Direct HTTP integration
Protect an endpoint in any server language with two Merchant API calls.
Call the Merchant API directly from any server language. This example uses Python and FastAPI; the access and completion requests are ordinary HTTP.
Configure your endpoint
You need Python 3.11 or newer and curl. The commands use a POSIX shell on macOS or Linux; on Windows, use WSL.
Before you start, complete Set up your product and keep the rvk_… revision key and merchant test key.
Create an empty directory and install the server dependencies:
python -m venv .venv. .venv/bin/activatepython -m pip install fastapi httpx uvicornSet the revision key and test key. PROTECTED_RESOURCE_URL must be the URL the payer calls; the local value below is correct for this guide.
The localhost URL works when the payer and the server run on the same machine. From a container or remote environment, use an HTTPS URL that the payer can reach.
export RECUUT_API_URL="https://api.recuut.com"export RECUUT_API_KEY="recuut_test_..."export RECUUT_REVISION_KEY="rvk_..."export PROTECTED_RESOURCE_URL="http://127.0.0.1:8000/generate"Create the complete server:
import osfrom time import perf_counterfrom typing import Annotated, Anyimport httpxfrom fastapi import FastAPI, Headerfrom fastapi.responses import JSONResponse, Responsefrom pydantic import BaseModelRECUUT_API_URL = os.environ["RECUUT_API_URL"]RECUUT_API_KEY = os.environ["RECUUT_API_KEY"]RECUUT_REVISION_KEY = os.environ["RECUUT_REVISION_KEY"]PROTECTED_RESOURCE_URL = os.environ["PROTECTED_RESOURCE_URL"]app = FastAPI()class GenerateRequest(BaseModel): prompt: strdef relay_recuut_response(response: httpx.Response) -> Response: forwarded_headers = {} for name in ("PAYMENT-REQUIRED", "Retry-After"): value = response.headers.get(name) if value is not None: forwarded_headers[name] = value return Response( content=response.content, status_code=response.status_code, headers=forwarded_headers, media_type="application/json", )def merchant_headers(payment_signature: str | None) -> dict[str, str]: headers = {"Authorization": f"Bearer {RECUUT_API_KEY}"} if payment_signature: headers["PAYMENT-SIGNATURE"] = payment_signature return headersasync def report_failed_work( merchant: httpx.AsyncClient, transaction_id: str, payment_signature: str, started_at: float,) -> None: await merchant.post( f"/merchant/v1/transactions/{transaction_id}/complete", headers=merchant_headers(payment_signature), json={ "outcome": "failed", "failure_code": "merchant_operation_failed", "resource_execution": { "duration_ms": int((perf_counter() - started_at) * 1_000), "response_status": 500, }, }, )async def report_free_response( merchant: httpx.AsyncClient, access: httpx.Response, response: Response, started_at: float,) -> None: execution_id = access.headers.get("RECUUT-RESOURCE-EXECUTION-ID") if not execution_id: return # Match the SDK's best-effort metadata reporting for free access. try: await merchant.post( f"/merchant/v1/resource-executions/{execution_id}/complete", headers=merchant_headers(None), json={ "resource_execution": { "duration_ms": int((perf_counter() - started_at) * 1_000), "response_content_type": "application/json", "response_size_bytes": len(response.body), "response_status": response.status_code, }, }, ) except httpx.HTTPError: pass@app.post("/generate")async def generate( payload: GenerateRequest, payment_signature: Annotated[str | None, Header(alias="PAYMENT-SIGNATURE")] = None,) -> Response: input_tokens = len(payload.prompt.split()) async with httpx.AsyncClient( base_url=RECUUT_API_URL, timeout=10, ) as merchant: access = await merchant.post( "/merchant/v1/access", headers=merchant_headers(payment_signature), json={ "inputs": {"input_tokens": str(input_tokens)}, "resource": { "url": PROTECTED_RESOURCE_URL, "mime_type": "application/json", }, "revision_key": RECUUT_REVISION_KEY, }, ) if access.status_code != 200: return relay_recuut_response(access) transaction_id = access.json()["transaction_id"] if transaction_id is None: started_at = perf_counter() response = JSONResponse({"text": payload.prompt.upper()}) await report_free_response(merchant, access, response, started_at) return response if payment_signature is None: return JSONResponse( {"code": "payment_signature_missing"}, status_code=502, ) started_at = perf_counter() try: result = { "text": payload.prompt.upper(), "transaction_id": transaction_id, } except Exception: await report_failed_work( merchant, transaction_id, payment_signature, started_at, ) raise protected_response = JSONResponse(result) completion = await merchant.post( f"/merchant/v1/transactions/{transaction_id}/complete", headers=merchant_headers(payment_signature), json={ "outcome": "succeeded", "resource_execution": { "duration_ms": int((perf_counter() - started_at) * 1_000), "response_content_type": "application/json", "response_size_bytes": len(protected_response.body), "response_status": protected_response.status_code, }, "values": { "input_tokens": str(input_tokens), "output_tokens": str(len(result["text"].split())), }, }, ) if completion.status_code != 200: return relay_recuut_response(completion) completed: dict[str, Any] = completion.json() payment_response = completion.headers.get("PAYMENT-RESPONSE") if completed.get("status") != "succeeded" or not payment_response: return JSONResponse( {"code": "payment_not_completed"}, status_code=502, ) protected_response.headers["PAYMENT-RESPONSE"] = payment_response return protected_responseThe server follows these steps:
- Validates the request and derives
input_tokenson the server. - Calls access. A
402goes back to the payer unchanged, without running the work. - Performs the work after access succeeds. Free access has no transaction ID; it returns the result and reports execution metadata.
- For paid access, completes the transaction with the input and output variables.
- Returns the paid result only after successful completion with
PAYMENT-RESPONSE. A202goes back without the result.
In production, make work with side effects idempotent and store its private result by transaction ID.
Start it in the same terminal that holds the environment variables:
uvicorn direct_http:app --reloadTest your endpoint
In another terminal, call the endpoint without a wallet:
curl --include \ --request POST \ --url http://127.0.0.1:8000/generate \ --header 'Content-Type: application/json' \ --data '{"prompt":"paid testnet response"}'The header value will be long, but the response shape should be:
HTTP/1.1 402 Payment RequiredPAYMENT-REQUIRED: <base64 payment instructions>content-type: application/json{"code":"payment_required","message":"Payment authorization is required."}Stop here if the response is not HTTP 402, if PAYMENT-REQUIRED is missing, or if the uppercase result appears. Check the server key, revision key, published Price, and Require payment setting before funding a wallet.
Then test a paid request against $PROTECTED_RESOURCE_URL.
For field-by-field details, use Authorize resource access, Complete a transaction, and Get a transaction. For maintained endpoint adapters, continue with FastAPI or Django.