Fastapi

pip install 'x402[fastapi,httpx,evm,svm]' 'solana<0.40' uvicorn python-dotenv pydantic

EVM_ADDRESS=0x...       # EVM wallet address to receive payments
SVM_ADDRESS=...         # Solana wallet address to receive payments
FACILITATOR_URL=https://facilitator.payai.network

import os

from dotenv import load_dotenv
from fastapi import FastAPI
from pydantic import BaseModel

from x402.http import FacilitatorConfig, HTTPFacilitatorClient, PaymentOption
from x402.http.middleware.fastapi import PaymentMiddlewareASGI
from x402.http.types import RouteConfig
from x402.mechanisms.evm.exact import ExactEvmServerScheme
from x402.mechanisms.svm.exact import ExactSvmServerScheme
from x402.schemas import AssetAmount, Network
from x402.server import x402ResourceServer

load_dotenv()

# Config
EVM_ADDRESS = os.getenv("EVM_ADDRESS")
SVM_ADDRESS = os.getenv("SVM_ADDRESS")
EVM_NETWORK: Network = "eip155:84532"  # Base Sepolia
SVM_NETWORK: Network = "solana:EtWTRABZaYq6iMfeYKouRu166VU2xqa1"  # Solana Devnet
FACILITATOR_URL = os.getenv("FACILITATOR_URL", "https://x402.org/facilitator")

if not EVM_ADDRESS or not SVM_ADDRESS:
    raise ValueError("Missing required environment variables")


# Response schemas
class WeatherReport(BaseModel):
    weather: str
    temperature: int


class WeatherResponse(BaseModel):
    report: WeatherReport


class PremiumContentResponse(BaseModel):
    content: str


# App
app = FastAPI()


# x402 Middleware
facilitator = HTTPFacilitatorClient(FacilitatorConfig(url=FACILITATOR_URL))
server = x402ResourceServer(facilitator)
server.register(EVM_NETWORK, ExactEvmServerScheme())
server.register(SVM_NETWORK, ExactSvmServerScheme())

routes = {
    "GET /weather": RouteConfig(
        accepts=[
            PaymentOption(
                scheme="exact",
                pay_to=EVM_ADDRESS,
                price="$0.01",
                network=EVM_NETWORK,
            ),
            PaymentOption(
                scheme="exact",
                pay_to=SVM_ADDRESS,
                price="$0.01",
                network=SVM_NETWORK,
            ),
        ],
        mime_type="application/json",
        description="Weather report",
    ),
    "GET /premium/*": RouteConfig(
        accepts=[
            PaymentOption(
                scheme="exact",
                pay_to=EVM_ADDRESS,
                price=AssetAmount(
                    amount="10000",  # $0.01 USDC
                    asset="0x036CbD53842c5426634e7929541eC2318f3dCF7e",
                    extra={"name": "USDC", "version": "2"},
                ),
                network=EVM_NETWORK,
            ),
            PaymentOption(
                scheme="exact",
                pay_to=SVM_ADDRESS,
                price="$0.01",
                network=SVM_NETWORK,
            ),
        ],
        mime_type="application/json",
        description="Premium content",
    ),
}
app.add_middleware(PaymentMiddlewareASGI, routes=routes, server=server)


# Routes
@app.get("/health")
async def health_check() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/weather")
async def get_weather() -> WeatherResponse:
    return WeatherResponse(report=WeatherReport(weather="sunny", temperature=70))


@app.get("/premium/content")
async def get_premium_content() -> PremiumContentResponse:
    return PremiumContentResponse(content="This is premium content")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=4021)

uvicorn main:app --reload

PAYAI_API_KEY_ID=your-key-id
PAYAI_API_KEY_SECRET=payai_sk_your-api-key-secret

pip install cryptography

import base64, json, time, uuid
from cryptography.hazmat.primitives.serialization import load_der_private_key
from x402.http.facilitator_client_base import AuthHeaders


class PayAIAuthProvider:
    """Signs short-lived EdDSA JWTs for the PayAI facilitator.

    Implements the x402 AuthProvider protocol, so the facilitator client
    attaches the token to every request without further wiring.
    """

    def __init__(self, api_key_id: str, api_key_secret: str, ttl: int = 120):
        self._kid = api_key_id
        secret = api_key_secret.strip()
        if secret.startswith("payai_sk_"):
            secret = secret[len("payai_sk_"):]
        self._key = load_der_private_key(base64.b64decode(secret), password=None)
        self._ttl = ttl
        self._token: str | None = None
        self._renew_at = 0.0

    @staticmethod
    def _b64(raw: bytes) -> str:
        return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")

    def _mint(self) -> str:
        now = int(time.time())
        header = self._b64(json.dumps(
            {"alg": "EdDSA", "typ": "JWT", "kid": self._kid},
            separators=(",", ":")).encode())
        payload = self._b64(json.dumps(
            {"sub": self._kid, "iss": "payai-merchant", "iat": now,
             "exp": now + self._ttl, "jti": str(uuid.uuid4())},
            separators=(",", ":")).encode())
        message = f"{header}.{payload}"
        self._token = f"{message}.{self._b64(self._key.sign(message.encode()))}"
        # refresh a little early so an in-flight request never carries a token
        # that expires mid-lifetime
        self._renew_at = now + self._ttl - 30
        return self._token

    def _jwt(self) -> str:
        if self._token and time.time() < self._renew_at:
            return self._token
        return self._mint()

    def get_auth_headers(self) -> AuthHeaders:
        h = {"Authorization": f"Bearer {self._jwt()}"}
        return AuthHeaders(verify=h, settle=h, supported=h, bazaar=h)

from payai_auth import PayAIAuthProvider

facilitator = HTTPFacilitatorClient(
    FacilitatorConfig(
        url=FACILITATOR_URL,
        auth_provider=PayAIAuthProvider(
            os.environ["PAYAI_API_KEY_ID"],
            os.environ["PAYAI_API_KEY_SECRET"],
        ),
    )
)