""" x402 Python client (reference implementation). End-to-end pay-per-call against https://x402.adametherzlab.com/* endpoints. Uses EIP-3009 transferWithAuthorization signed via eth_account; the CDP facilitator settles the on-chain USDC transfer (you do not need ETH for gas). Tested with eth-account 0.13.x, requests 2.31+, Python 3.10+. Usage: python x402_py_client.py """ import base64 import json import os import secrets import time import requests from eth_account import Account from eth_account.messages import encode_typed_data # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- X402_HOST = "https://x402.adametherzlab.com" USDC_BASE = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913" CHAIN_ID = 8453 # Base mainnet # --------------------------------------------------------------------------- # EIP-3009 typed-data signing for USDC on Base. # --------------------------------------------------------------------------- def _sign_eip3009( private_key: str, payer: str, pay_to: str, amount_micros: int, valid_after_seconds_skew: int = 30, valid_before_seconds: int = 870, ): """Sign an EIP-3009 transferWithAuthorization message. Returns the dict shape that the x402 'exact' EVM scheme expects: { 'authorization': { from, to, value, validAfter, validBefore, nonce }, 'signature': '0x...' (65-byte r||s||v hex with v=27/28), } validAfter is set to (now - 30s) to tolerate clock skew between the agent and the facilitator. validBefore is (now + 870s) -- ~15 min window to match the reference x402 JS client. """ now = int(time.time()) valid_after = max(0, now - valid_after_seconds_skew) valid_before = now + valid_before_seconds nonce_bytes = secrets.token_bytes(32) nonce_hex = "0x" + nonce_bytes.hex() typed_data = { "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": "USD Coin", "version": "2", "chainId": CHAIN_ID, "verifyingContract": USDC_BASE, }, "message": { "from": payer, "to": pay_to, "value": amount_micros, "validAfter": valid_after, "validBefore": valid_before, "nonce": nonce_bytes, }, } encoded = encode_typed_data(full_message=typed_data) signed = Account.sign_message(encoded, private_key=private_key) sig_hex = signed.signature.hex() if not sig_hex.startswith("0x"): sig_hex = "0x" + sig_hex return { "authorization": { "from": payer, "to": pay_to, "value": str(amount_micros), "validAfter": str(valid_after), "validBefore": str(valid_before), "nonce": nonce_hex, }, "signature": sig_hex, } # --------------------------------------------------------------------------- # x402 wire helpers. # --------------------------------------------------------------------------- def _decode_payment_required(headers: dict, body_text: str) -> dict: """Decode the 402 challenge into a Python dict. Try base64 header first, then JSON body.""" hdr = headers.get("payment-required") or headers.get("Payment-Required") if hdr: try: return json.loads(base64.b64decode(hdr).decode("utf-8")) except Exception: pass if body_text: try: j = json.loads(body_text) if "accepts" in j: return j except Exception: pass raise RuntimeError("Could not decode 402 challenge (no payment-required header and body is not a valid challenge JSON)") def _encode_payment_signature_header(payload_dict: dict) -> str: raw = json.dumps(payload_dict, separators=(",", ":")).encode("utf-8") return base64.b64encode(raw).decode("ascii") # --------------------------------------------------------------------------- # Public entry point. # --------------------------------------------------------------------------- def pay_and_call( url: str, private_key: str, method: str = "POST", body: dict | None = None, timeout: float = 90.0, ) -> dict: """Hit an x402 endpoint, pay if asked, return the JSON response. Returns the parsed JSON of the 200 response (or raises on non-2xx after pay). """ payer = Account.from_key(private_key).address headers = {"Content-Type": "application/json"} data = None if body is None else json.dumps(body) # 1) Initial request -> expect 402 challenge. r1 = requests.request(method, url, headers=headers, data=data, timeout=timeout) if r1.status_code == 200: return r1.json() # free endpoint, no payment if r1.status_code != 402: raise RuntimeError(f"Expected 402 challenge, got HTTP {r1.status_code}: {r1.text[:300]}") challenge = _decode_payment_required({k.lower(): v for k, v in r1.headers.items()}, r1.text) accept = challenge["accepts"][0] amount_micros = int(accept["amount"]) pay_to = accept["payTo"] # 2) Sign EIP-3009 authorization. auth_payload = _sign_eip3009(private_key, payer, pay_to, amount_micros) # 3) Build the full payment payload (must match the JS client's wire shape): # {x402Version, payload, extensions, resource, accepted} # extensions / resource flow back from the challenge; accepted is the # specific accepts[] entry we chose. wire_payload = { "x402Version": challenge.get("x402Version", 2), "payload": auth_payload, "extensions": challenge.get("extensions", {}), "resource": challenge.get("resource", {}), "accepted": accept, } payment_hdr = _encode_payment_signature_header(wire_payload) # 4) Retry with the payment header. headers2 = {**headers, "payment-signature": payment_hdr} r2 = requests.request(method, url, headers=headers2, data=data, timeout=timeout) if not r2.ok: raise RuntimeError(f"x402 settle failed (HTTP {r2.status_code}): {r2.text[:300]}") try: return r2.json() except ValueError: return {"raw": r2.text} # --------------------------------------------------------------------------- # Smoke test: call /api/companion-planting ($0.01) when run as a script. # Uses X402_PRIVATE_KEY env var. # --------------------------------------------------------------------------- if __name__ == "__main__": pk = os.environ.get("X402_PRIVATE_KEY") if not pk: raise SystemExit("Set X402_PRIVATE_KEY env var to a Base wallet private key with ~$0.05 USDC.") print(f"payer: {Account.from_key(pk).address}") print("calling /api/companion-planting with {plant: 'yaupon holly'} (paid endpoint, $0.01) ...") started = time.time() result = pay_and_call( url=f"{X402_HOST}/api/companion-planting", private_key=pk, method="POST", body={"plant": "yaupon holly"}, ) elapsed_ms = int((time.time() - started) * 1000) print(f"status: OK in {elapsed_ms} ms") print(f"server-extracted query: {result.get('query')!r}") print(f"result preview: {(result.get('result') or '')[:200]}")