"""Default facilitator: HTTP calls to POST <url>/verify and POST <url>/settle (x402 v1 body)."""
from __future__ import annotations

from typing import Protocol

import httpx

DEFAULT_FACILITATOR_URL = "https://x402.org/facilitator"


class Facilitator(Protocol):
    async def verify(self, payment_payload: dict, payment_requirements: dict) -> dict: ...
    async def settle(self, payment_payload: dict, payment_requirements: dict) -> dict: ...


class HttpFacilitator:
    def __init__(self, url: str = DEFAULT_FACILITATOR_URL, client: httpx.AsyncClient | None = None):
        self.url = url.rstrip("/")
        self.client = client

    async def _post(self, path: str, payment_payload: dict, payment_requirements: dict) -> dict:
        body = {"x402Version": 1, "paymentPayload": payment_payload, "paymentRequirements": payment_requirements}
        if self.client is not None:
            r = await self.client.post(f"{self.url}/{path}", json=body)
        else:
            async with httpx.AsyncClient(timeout=30) as c:
                r = await c.post(f"{self.url}/{path}", json=body)
        try:
            return r.json()
        except ValueError:
            return {}

    async def verify(self, payment_payload: dict, payment_requirements: dict) -> dict:
        return await self._post("verify", payment_payload, payment_requirements)

    async def settle(self, payment_payload: dict, payment_requirements: dict) -> dict:
        return await self._post("settle", payment_payload, payment_requirements)

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