"""A paid agent in Python (FastAPI): one route priced in USDC over x402 (v1 exact scheme).

GET /health  free
GET /paid    402 without X-PAYMENT; with it: verify, settle, then the work and X-PAYMENT-RESPONSE

Run: PAY_TO=0x... NETWORK=base-sepolia uvicorn app:app --port 8403
"""
from __future__ import annotations

import os
from datetime import datetime, timezone
from typing import Any, Awaitable, Callable

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from facilitator import DEFAULT_FACILITATOR_URL, Facilitator, HttpFacilitator
from x402 import NETWORKS, build_requirements, encode_header, is_address, read_payment, to_atomic_usdc

Work = Callable[[Request], Awaitable[Any]]


async def _default_work(request: Request) -> Any:
    return {"ok": True, "message": "Paid work done.", "path": request.url.path,
            "at": datetime.now(timezone.utc).isoformat()}


def create_app(*, price: str, pay_to: str, network: str, facilitator: Facilitator | str | None = None,
               description: str = "A paid agent call", work: Work | None = None,
               public_url: str | None = None) -> FastAPI:
    to_atomic_usdc(price)
    if not is_address(pay_to):
        raise ValueError("pay_to must be 0x and 40 hex characters")
    if network not in NETWORKS:
        raise ValueError("network must be base or base-sepolia")
    fac: Facilitator = facilitator if facilitator is not None and not isinstance(facilitator, str) \
        else HttpFacilitator(facilitator or DEFAULT_FACILITATOR_URL)
    do_work = work or _default_work
    app = FastAPI(title="x402 paid agent starter", docs_url=None, redoc_url=None, openapi_url=None)

    def requirements(request: Request) -> dict:
        origin = (public_url or f"{request.url.scheme}://{request.url.netloc}").rstrip("/")
        return build_requirements(price=price, pay_to=pay_to, network=network,
                                  resource=f"{origin}{request.url.path}", description=description)

    def payment_required(req: dict, error: str) -> JSONResponse:
        return JSONResponse({"x402Version": 1, "error": error, "accepts": [req]}, status_code=402)

    @app.get("/health")
    async def health() -> dict:
        return {"ok": True}

    @app.get("/paid")
    async def paid(request: Request):
        req = requirements(request)
        header = request.headers.get("x-payment")
        if not header:
            return payment_required(req, "X-PAYMENT header is required")
        payment, reason = read_payment(header, req)
        if payment is None:
            return payment_required(req, reason or "bad payment")
        try:
            verified = await fac.verify(payment, req)
        except Exception:
            return payment_required(req, "the payment could not be verified")
        if not isinstance(verified, dict) or verified.get("isValid") is not True:
            return payment_required(req, (verified or {}).get("invalidReason") or "the payment is not valid")
        try:
            settled = await fac.settle(payment, req)
        except Exception:
            return payment_required(req, "the payment could not be settled")
        if not isinstance(settled, dict) or settled.get("success") is not True:
            return payment_required(req, (settled or {}).get("errorReason") or "the payment did not settle")
        result = await do_work(request)
        receipt = {"success": True, "transaction": settled.get("transaction"),
                   "network": settled.get("network") or network, "payer": settled.get("payer") or verified.get("payer")}
        return JSONResponse(result, headers={"X-PAYMENT-RESPONSE": encode_header(receipt),
                                             "Access-Control-Expose-Headers": "X-PAYMENT-RESPONSE"})

    return app


def _from_env() -> FastAPI:
    return create_app(price=os.environ.get("PRICE_USDC", "0.01"),
                      pay_to=os.environ.get("PAY_TO", "0x0000000000000000000000000000000000000000"),
                      network=os.environ.get("NETWORK", "base-sepolia"),
                      facilitator=os.environ.get("FACILITATOR_URL", DEFAULT_FACILITATOR_URL),
                      description=os.environ.get("DESCRIPTION", "One paid agent call"),
                      public_url=os.environ.get("PUBLIC_URL"))


app = _from_env()

agentgate-source: code mirror on Cloudflare Pages. All repos