#!/usr/bin/env python3 """x402 reference payer — settle an HTTP 402 (scheme "exact", EVM / EIP-3009). Fetch, run, done: pip install eth-account python3 pay.py https://host/subscribe --key-file eth_private.hex \ --body '{"handle": "you", "webhook_url": "https://your-endpoint"}' What it does: requests the URL, reads the 402 payment requirements, signs an EIP-3009 transfer authorization with your key, and retries the request with the X-PAYMENT header. The private key never leaves this machine — only the signature is sent. The authorization is exact-amount, payable only to the server's published address, and expires within minutes. The server's facilitator broadcasts it and pays the gas: your wallet needs the token (USDC), never ETH. Key input (never a CLI argument — those leak into shell history and ps): --key-file PATH file containing the hex private key, or X402_PRIVATE_KEY env the hex key itself Only dependency: eth-account (>=0.9). Everything else is stdlib. """ from __future__ import annotations import argparse import base64 import json import secrets import sys import time import urllib.error import urllib.request CHAIN_IDS = {"base": 8453, "base-sepolia": 84532} EIP3009_TYPES = { "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"}, ] } def http(url: str, method: str, body: bytes | None, headers: dict[str, str], timeout: int): req = urllib.request.Request(url, data=body, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as r: return r.status, dict(r.headers), r.read() except urllib.error.HTTPError as e: return e.code, dict(e.headers), e.read() def choose_requirement(accepts, amount: str | None = None, network: str | None = None): """Pick one offer from a 402 `accepts` list; None if nothing fits.""" offers = [r for r in accepts if r.get("scheme") == "exact"] if network: offers = [r for r in offers if r.get("network") == network] else: offers = [r for r in offers if r.get("network") in CHAIN_IDS] if amount is not None: offers = [r for r in offers if str(r.get("maxAmountRequired")) == str(amount)] return min(offers, key=lambda r: int(r["maxAmountRequired"])) if offers else None def sign_payment(key: str, req: dict, chain_id: int) -> dict: try: from eth_account import Account except ImportError: sys.exit("pay.py needs eth-account: pip install eth-account") acct = Account.from_key(key) now = int(time.time()) extra = req.get("extra") or {} if not extra.get("name") or not extra.get("version"): print("warning: offer lacks extra.name/version; assuming USDC domain", file=sys.stderr) domain = { "name": extra.get("name") or "USD Coin", "version": extra.get("version") or "2", "chainId": chain_id, "verifyingContract": req["asset"], } nonce = secrets.token_bytes(32) # validAfter is backdated to tolerate clock skew between us and the chain message = { "from": acct.address, "to": req["payTo"], "value": int(req["maxAmountRequired"]), "validAfter": now - 600, "validBefore": now + int(req.get("maxTimeoutSeconds") or 300), "nonce": nonce, } signed = Account.sign_typed_data(acct.key, domain, EIP3009_TYPES, message) sig = signed.signature.hex() return { "x402Version": 1, "scheme": "exact", "network": req["network"], "payload": { "signature": sig if sig.startswith("0x") else "0x" + sig, "authorization": { "from": acct.address, "to": req["payTo"], "value": str(message["value"]), "validAfter": str(message["validAfter"]), "validBefore": str(message["validBefore"]), "nonce": "0x" + nonce.hex(), }, }, } def load_key(args) -> str: import os if args.key_file: key = open(args.key_file).read().strip() else: key = os.environ.get("X402_PRIVATE_KEY", "").strip() if not key: sys.exit("no key: pass --key-file PATH or set X402_PRIVATE_KEY") return key def show(status: int, body: bytes, output: str | None) -> dict: print(f"HTTP {status}", file=sys.stderr) try: obj = json.loads(body) print(json.dumps(obj, indent=2)) except ValueError: obj = {} sys.stdout.buffer.write(body + b"\n") if output: with open(output, "wb") as f: f.write(body) return obj if isinstance(obj, dict) else {} def main(argv=None) -> int: p = argparse.ArgumentParser(description="pay an x402 HTTP 402 (exact scheme, EVM)") p.add_argument("url") p.add_argument("--body", default="{}", help="JSON request body") p.add_argument("--method", default="POST") p.add_argument("--key-file", help="file containing the hex private key") p.add_argument("--amount", help="atomic token units; must match an offer exactly") p.add_argument("--network", help="restrict to one network (e.g. base)") p.add_argument("--chain-id", type=int, help="EVM chain id for networks pay.py doesn't know") p.add_argument("-H", "--header", action="append", default=[], help='extra header "Name: value"') p.add_argument("-o", "--output", help="also write the final response body to this file") p.add_argument("--timeout", type=int, default=60) p.add_argument("--dry-run", action="store_true", help="show the chosen offer and stop (no signing)") args = p.parse_args(argv) try: body = json.dumps(json.loads(args.body)).encode() except ValueError: sys.exit("--body is not valid JSON") headers = {"Content-Type": "application/json", "User-Agent": "x402-pay.py/1.0"} for h in args.header: name, _, value = h.partition(":") headers[name.strip()] = value.strip() status, _, resp = http(args.url, args.method, body, headers, args.timeout) if status != 402: print("no payment was required" if status < 300 else "unexpected response", file=sys.stderr) show(status, resp, args.output) return 0 if status < 300 else 1 try: challenge = json.loads(resp) accepts = challenge["accepts"] except (ValueError, KeyError): sys.exit(f"402 body is not an x402 challenge: {resp[:500]!r}") if challenge.get("x402Version") not in (None, 1): print(f"warning: x402Version {challenge['x402Version']}, expected 1", file=sys.stderr) req = choose_requirement(accepts, args.amount, args.network) if req is None: sys.exit("no matching offer; server accepts:\n" + json.dumps(accepts, indent=2)) chain_id = args.chain_id or CHAIN_IDS.get(req["network"]) if chain_id is None: sys.exit(f"unknown network {req['network']!r}: pass --chain-id") print( f"paying: {req['maxAmountRequired']} atomic units of {req['asset']} " f"on {req['network']} to {req['payTo']}\n ({req.get('description', 'no description')})", file=sys.stderr, ) if args.dry_run: return 0 payment = sign_payment(load_key(args), req, chain_id) headers["X-PAYMENT"] = base64.b64encode(json.dumps(payment).encode()).decode() status, resp_headers, resp = http(args.url, args.method, body, headers, args.timeout) settle = resp_headers.get("X-PAYMENT-RESPONSE") or resp_headers.get("X-Payment-Response") if settle: try: print(f"settlement: {base64.b64decode(settle).decode()}", file=sys.stderr) except Exception: pass obj = show(status, resp, args.output) if "secret" in obj: print("NOTE: the response contains a one-time secret — save it now.", file=sys.stderr) return 0 if status < 300 else 1 if __name__ == "__main__": sys.exit(main())