"""Builds an x402 "exact" EVM payment header (EIP-3009) for v1 or v2 terms. The key stays with the caller."""
from __future__ import annotations

import os
import time

from eth_account.messages import encode_typed_data

from x402 import encode_header

CHAIN_IDS = {"base": 8453, "base-sepolia": 84532, "eip155:8453": 8453, "eip155:84532": 84532}


def exact_payment_header(terms: dict, account, *, max_atomic: int | None = None) -> str:
    req = next((a for a in terms.get("accepts", []) if a.get("scheme") == "exact" and str(a.get("network")) in CHAIN_IDS), None)
    if req is None:
        raise ValueError("no exact EVM payment option on Base in the 402 terms")
    value = str(req.get("amount") or req.get("maxAmountRequired"))
    if max_atomic is not None and int(value) > max_atomic:
        raise ValueError(f"refusing to pay {value} atomic USDC, above the cap {max_atomic}")
    now = int(time.time())
    auth = {"from": account.address, "to": req["payTo"], "value": value, "validAfter": str(now - 600),
            "validBefore": str(now + int(req.get("maxTimeoutSeconds", 60))), "nonce": "0x" + os.urandom(32).hex()}
    extra = req.get("extra") or {}
    typed = {
        "types": {
            "EIP712Domain": [{"name": "name", "type": "string"}, {"name": "version", "type": "string"},
                             {"name": "chainId", "type": "uint256"}, {"name": "verifyingContract", "type": "address"}],
            "TransferWithAuthorization": [{"name": "from", "type": "address"}, {"name": "to", "type": "address"},
                                          {"name": "value", "type": "uint256"}, {"name": "validAfter", "type": "uint256"},
                                          {"name": "validBefore", "type": "uint256"}, {"name": "nonce", "type": "bytes32"}],
        },
        "primaryType": "TransferWithAuthorization",
        "domain": {"name": extra.get("name", "USD Coin"), "version": extra.get("version", "2"),
                   "chainId": CHAIN_IDS[str(req["network"])], "verifyingContract": req["asset"]},
        "message": {"from": auth["from"], "to": auth["to"], "value": int(value), "validAfter": int(auth["validAfter"]),
                    "validBefore": int(auth["validBefore"]), "nonce": bytes.fromhex(auth["nonce"][2:])},
    }
    sig = account.sign_message(encode_typed_data(full_message=typed)).signature.hex()
    sig = sig if sig.startswith("0x") else "0x" + sig
    payload = {"signature": sig, "authorization": auth}
    if int(terms.get("x402Version", 1)) >= 2:
        return encode_header({"x402Version": 2, "resource": terms.get("resource"), "accepted": req, "payload": payload})
    return encode_header({"x402Version": 1, "scheme": "exact", "network": req["network"], "payload": payload})

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