Python SDK Reference#

from x402 import x402ResourceServer

class x402ResourceServer:
    def __init__(
        self,
        facilitator_clients: FacilitatorClient | list[FacilitatorClient] | None = None,
    ) -> None: ...

    def register(self, network: str, scheme) -> Self: ...
    def initialize(self) -> None: ...

    def on_before_verify(self, hook) -> Self
    def on_after_verify(self, hook) -> Self
    def on_verify_failure(self, hook) -> Self
    def on_before_settle(self, hook) -> Self
    def on_after_settle(self, hook) -> Self
    def on_settle_failure(self, hook) -> Self

from x402 import x402Facilitator

class x402Facilitator:
    def register(self, networks: list[str], scheme) -> Self: ...
    async def verify(self, payload, requirements) -> VerifyResponse: ...
    async def settle(self, payload, requirements) -> SettleResponse: ...

from x402 import x402Client

class x402Client:
    def __init__(self, payment_requirements_selector=None) -> None: ...
    def register(self, network: str, client_scheme) -> Self: ...
    async def create_payment_payload(self, payment_required) -> PaymentPayload: ...

from x402.http import OKXAuthConfig, OKXFacilitatorClient, OKXFacilitatorConfig

@dataclass
class OKXAuthConfig:
    api_key: str
    secret_key: str
    passphrase: str

@dataclass
class OKXFacilitatorConfig:
    auth: OKXAuthConfig
    base_url: str = "https://web3.okx.com"
    sync_settle: bool = True
    timeout: float = 30.0
    http_client: Any = None

class OKXFacilitatorClient:
    def __init__(self, config: OKXFacilitatorConfig) -> None: ...
    async def verify(self, payload, requirements) -> VerifyResponse: ...
    async def settle(self, payload, requirements) -> SettleResponse: ...
    def get_supported(self) -> SupportedResponse: ...
    async def aclose(self) -> None: ...

from x402.mechanisms.evm.exact.server import ExactEvmScheme
from x402.mechanisms.evm.deferred.server import AggrDeferredEvmScheme

class ExactEvmScheme:
    scheme = "exact"
    def __init__(self) -> None: ...
    def register_money_parser(self, parser) -> "ExactEvmScheme": ...
    def parse_price(self, price, network) -> AssetAmount: ...
    def enhance_payment_requirements(self, requirements, supported_kind, extension_keys) -> PaymentRequirements: ...

class AggrDeferredEvmScheme:
    scheme = "aggr_deferred"
    def __init__(self) -> None: ...

from x402.http import PaymentOption, RouteConfig

@dataclass
class PaymentOption:
    scheme: str
    pay_to: str | DynamicPayTo
    price: Price | DynamicPrice
    network: Network
    max_timeout_seconds: int | None = None
    extra: dict[str, Any] | None = None

@dataclass
class RouteConfig:
    accepts: PaymentOption | list[PaymentOption]
    resource: str | None = None
    description: str | None = None
    mime_type: str | None = None
    custom_paywall_html: str | None = None
    unpaid_response_body: UnpaidResponseBody | None = None
    settlement_failed_response_body: SettlementFailedResponseBody | None = None
    extensions: dict[str, Any] | None = None
    hook_timeout_seconds: float | None = None

RoutesConfig = dict[str, RouteConfig] | RouteConfig

from x402.http.middleware.fastapi import PaymentMiddlewareASGI

app.add_middleware(
    PaymentMiddlewareASGI,
    routes=routes,
    server=server,
)

from x402.http.middleware.flask import PaymentMiddleware, payment_middleware

class VerifyResponse(BaseX402Model):
    is_valid: bool
    invalid_reason: str | None = None
    invalid_message: str | None = None
    payer: str | None = None

class SettleResponse(BaseX402Model):
    success: bool
    error_reason: str | None = None
    error_message: str | None = None
    status: str | None = None
    payer: str | None = None
    transaction: str
    network: Network

class SettleStatusResponse(BaseX402Model):
    success: bool
    status: str | None = None
    error_reason / error_message / payer / transaction / network: ...

class SupportedKind(BaseX402Model):
    x402_version: int
    scheme: str
    network: Network
    extra: dict[str, Any] | None = None

class SupportedResponse(BaseX402Model):
    kinds: list[SupportedKind]
    extensions: list[str] = []
    signers: dict[str, list[str]] = {}

PAYMENT_SIGNATURE_HEADER = "PAYMENT-SIGNATURE"
PAYMENT_REQUIRED_HEADER = "PAYMENT-REQUIRED"
PAYMENT_RESPONSE_HEADER = "PAYMENT-RESPONSE"
X_PAYMENT_HEADER = "X-PAYMENT"
X_PAYMENT_RESPONSE_HEADER = "X-PAYMENT-RESPONSE"
HTTP_STATUS_PAYMENT_REQUIRED = 402

encode/decode_payment_signature_header(...)
encode/decode_payment_required_header(...)
encode/decode_payment_response_header(...)
detect_payment_required_version(...)
safe_base64_encode / safe_base64_decode

from x402.adapters import X402Adapter
from x402.http import x402HTTPResourceServer

class X402Adapter:
    def __init__(self, server: x402HTTPResourceServer, priority: int = 20) -> None: ...
    @property
    def name(self) -> str: ...
    @property
    def priority(self) -> int: ...
    def detect(self, request) -> bool: ...
    async def get_challenge(self, request, cfg) -> dict[str, str]: ...
    async def handle(self, request, cfg) -> Any: ...

from x402.server import x402ResourceServer
from x402.http import x402HTTPResourceServer

resource = x402ResourceServer(facilitator)
resource.register("eip155:196", ExactEvmScheme())
resource.register("eip155:196", AggrDeferredEvmScheme())
http_server = x402HTTPResourceServer(resource, routes={})
x402_adapter = X402Adapter(http_server)

from mpp_evm import (
    X_LAYER_CHAIN_ID,
    DEFAULT_ESCROW_CONTRACT,
    DEFAULT_DOMAIN_NAME,
    DEFAULT_DOMAIN_VERSION,
    METHOD_NAME_EVM,
    ACTION_OPEN,
    ACTION_TOP_UP,
    ACTION_VOUCHER,
    ACTION_CLOSE,
    ACTION_SETTLE,
    STATUS_OPEN,
    STATUS_CLOSED,
)

from mpp.server.mpp import Mpp

class Mpp:
    def __init__(
        self,
        method: Method,
        realm: str,
        secret_key: str,
        defaults: dict[str, Any] | None = None,
        store: Store | None = None,
    ) -> None: ...

    @classmethod
    def create(cls, method, realm=None, secret_key=None, store=None) -> "Mpp": ...

async def charge(
    self,
    authorization: str | None,
    amount: str,
    *,
    currency: str | None = None,
    recipient: str | None = None,
    expires: str | None = None,
    description: str | None = None,
    memo: str | None = None,
    splits: list[dict[str, str]] | None = None,
    fee_payer: bool = False,
    chain_id: int | None = None,
    extra: dict[str, str] | None = None,
) -> Challenge | tuple[Credential, Receipt]:
    ...

def pay(
    self,
    amount: str,
    *,
    intent: str = "charge",
    currency: str | None = None,
    recipient: str | None = None,
    description: str | None = None,
    expires_in: timedelta | None = None,
    chain_id: int | None = None,
    extra: dict[str, str] | None = None,
) -> Callable: ...

def on(self, name: str, handler: EventHandler) -> Unsubscribe
def on_challenge_created(self, handler) -> Unsubscribe
def on_payment_success(self, handler) -> Unsubscribe
def on_payment_failed(self, handler) -> Unsubscribe

from mpp import Challenge, Credential, Receipt

class Challenge:
    def to_www_authenticate(self, realm: str) -> str: ...

class Receipt:
    status: Literal["success"]
    timestamp: datetime
    reference: str
    method: str = "tempo"
    external_id: str | None = None
    extra: dict[str, Any] | None = None

    @classmethod
    def from_payment_receipt(cls, header: str) -> "Receipt": ...
    def to_payment_receipt(self) -> str: ...
    @classmethod
    def success(cls, reference, timestamp=None, method="tempo", external_id=None) -> "Receipt": ...

from mpp_evm import EvmMethod

class EvmMethod:
    name: str = "evm"
    def __init__(self, *, intents: dict[str, Any] | None = None) -> None: ...

    @property
    def intents(self) -> dict[str, Any]: ...
    async def create_credential(self, challenge) -> Any: ...

method = EvmMethod(intents={"charge": charge_intent, "session": session_intent})
method.currency = "0x...token"
method.recipient = "0x...payee"
method.decimals = 6
method.chain_id = 196

from mpp_evm.charge.intent import ChargeIntent

class ChargeIntent:
    name: str = "charge"
    def __init__(
        self,
        *,
        sa_client: SAClient,
        chain_id: int = X_LAYER_CHAIN_ID,
        recipient: str = "",
        fee_payer: bool = False,
    ) -> None: ...

    def challenge_method_details(self) -> dict[str, Any]: ...
    async def verify(self, credential: Credential, request: dict) -> Receipt: ...

from mpp_evm.session.intent import SessionIntent

class SessionIntent:
    name: str = "session"
    def __init__(
        self,
        *,
        sa_client: SAClient,
        recipient: str,
        signer: Signer | None = None,
        store: ChannelStore | None = None,
        chain_id: int = X_LAYER_CHAIN_ID,
        escrow_contract: str = DEFAULT_ESCROW_CONTRACT,
        per_request_cost: int | None = None,
        min_voucher_delta: int | None = None,
        nonce_provider: NonceProvider | None = None,
        deadline: int | None = None,
        domain_name: str = DEFAULT_DOMAIN_NAME,
        domain_version: str = DEFAULT_DOMAIN_VERSION,
        fee_payer: bool = False,
    ) -> None: ...

def challenge_method_details(self) -> dict[str, Any]: ...
async def verify(self, credential, request: dict) -> Any: ...
def respond(self, credential, receipt) -> Any: ...

async def settle_channel(self, channel_id: str) -> dict: ...
async def close_channel(self, channel_id: str) -> dict: ...

from mpp_evm.signer import Signer, PrivateKeySigner

@runtime_checkable
class Signer(Protocol):
    def sign(self, msg_hash: bytes) -> bytes: ...
    @property
    def address(self) -> str: ...

class PrivateKeySigner:
    def __init__(self, account: LocalAccount) -> None: ...
    @classmethod
    def from_hex(cls, hex_key: str) -> "PrivateKeySigner": ...

from mpp_evm.saclient.client import SAClient

@runtime_checkable
class SAClient(Protocol):
    async def settle(self, req: ChargeSettleRequest) -> ChargeReceipt: ...
    async def verify_hash(self, req: ChargeVerifyHashRequest) -> ChargeReceipt: ...
    async def session_open(self, req: SessionOpenRequest) -> SessionReceipt: ...
    async def session_top_up(self, req: SessionTopUpRequest) -> SessionReceipt: ...
    async def session_settle(self, req: SessionSettleRequest) -> SessionReceipt: ...
    async def session_close(self, req: SessionCloseRequest) -> SessionReceipt: ...
    async def session_status(self, channel_id: str) -> SessionStatus: ...

from mpp_evm.saclient.client import OKXSAClient

class OKXSAClient:
    def __init__(
        self,
        *,
        base_url: str,
        api_key: str,
        secret_key: str,
        passphrase: str,
        http_client: httpx.AsyncClient | None = None,
        timeout: float = 30.0,
    ) -> None: ...

@runtime_checkable
class ChannelStore(Protocol):
    async def get(self, key: str) -> Any | None: ...
    async def put(self, key: str, value: Any) -> None: ...
    async def delete(self, key: str) -> None: ...

@dataclass
class ChannelState:
    channel_id: str
    chain_id: int
    escrow_contract: str
    payer: str
    payee: str
    token: str = ""
    authorized_signer: str = ""
    deposit: str = "0"
    highest_voucher_amount: str = "0"
    highest_voucher_signature: str = ""
    min_voucher_delta: str = "0"
    spent: str = "0"
    units: int = 0
    finalized: bool = False
    close_requested_at: int = 0
    created_at: str = ""

from mpp_evm import FileStore

class FileStore:
    def __init__(self, directory: str) -> None: ...
    async def get(self, key) / put(self, key, value) / delete(self, key)
    async def put_if_absent(self, key, value) -> bool

from mpp_evm import MppAdapter, MppRouteConfig

@dataclass
class MppRouteConfig:
    intent: str = ""
    amount: str = ""
    currency: str = ""
    decimals: int = 6
    description: str = ""
    external_id: str = ""
    realm: str = ""
    unit_type: str = ""
    suggested_deposit: str = ""

class MppAdapter:
    def __init__(self, mpp: Mpp, priority: int = 10) -> None: ...
    @property
    def name(self) -> str: ...
    @property
    def priority(self) -> int: ...
    def detect(self, request) -> bool: ...
    async def get_challenge(self, request, cfg) -> dict[str, str]: ...
    async def handle(self, request, cfg) -> Any: ...

from fastapi import FastAPI, Request
from mpp import Credential, Receipt

app = FastAPI()

@app.get("/api/premium")
@mpp.pay(amount="0.00001", intent="charge", description="One premium API call")
async def premium(request: Request, credential: Credential, receipt: Receipt) -> dict:
    return {"data": "premium content", "receipt": {"reference": receipt.reference, "status": receipt.status}}