#!/usr/bin/env python3 """Reference client for Obolpay Gateway — compatible with x402™ — evaluate the free preview, pay per call, fetch the data. x402™ is a trademark of LF Projects, LLC. Served by the gateway itself. Read it before you run it: it signs payments with the key you give it (at most MAX_UNITS per call). Business use only; terms: https://pay.obolpay.xyz/terms · legal notice: https://pay.obolpay.xyz/legal pip install eth-account requests PRIVATE_KEY=0x... python client.py # optional: TYPES=jp-business-day MAX_UNITS=100000 It pays ONLY with a scheme the live HTTP 402 lists in accepts[]: * `exact` (standard x402, preferred): you only SIGN an EIP-3009 authorization; the gateway's facilitator submits it on-chain and pays the gas. The wallet needs >= the quoted USDC on Base. * `obolpay-tx-receipt` (non-standard; used only while the 402 still lists it): you send a USDC transfer yourself and pay the gas (also needs `pip install web3` and a little ETH). If the 402 offers neither, it stops WITHOUT signing or sending anything. """ import base64, json, os, secrets, time import requests from eth_account import Account from eth_account.messages import encode_defunct, encode_typed_data BASE = "https://pay.obolpay.xyz" ENDPOINT = BASE + "/api/v1/protected-data" RPC = "https://mainnet.base.org" UA = {"User-Agent": "x402-client/1.1"} # NOTE: send a User-Agent (Cloudflare blocks empty/raw-urllib) MAX_UNITS = int(os.environ.get("MAX_UNITS", "100000")) # never authorize more than 0.10 USDC per call LEGACY = "obolpay-tx-receipt" 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 exact_payment_header(acct, req): """Sign an EIP-3009 transferWithAuthorization for one `exact` accepts[] entry (x402 v1).""" network, value = req["network"], int(req["maxAmountRequired"]) if network not in CHAIN_IDS: raise SystemExit("unsupported network %r - refusing to sign" % network) if value > MAX_UNITS: raise SystemExit("price %d units exceeds MAX_UNITS=%d - refusing to sign" % (value, MAX_UNITS)) extra, now = req.get("extra") or {}, int(time.time()) auth = {"from": acct.address, "to": req["payTo"], "value": value, "validAfter": now - 60, "validBefore": now + int(req.get("maxTimeoutSeconds") or 300), "nonce": "0x" + secrets.token_hex(32)} # EIP-712 domain = the USDC contract's own: name "USD Coin" (NOT "USDC") on Base. signable = encode_typed_data( domain_data={"name": extra.get("name", "USD Coin"), "version": extra.get("version", "2"), "chainId": CHAIN_IDS[network], "verifyingContract": req["asset"]}, message_types=EIP3009_TYPES, message_data={**auth, "nonce": bytes.fromhex(auth["nonce"][2:])}) sig = acct.sign_message(signable).signature.hex() payload = {"x402Version": 1, "scheme": "exact", "network": network, "payload": {"signature": sig if sig.startswith("0x") else "0x" + sig, "authorization": {**auth, "value": str(value), "validAfter": str(auth["validAfter"]), "validBefore": str(auth["validBefore"])}}} return base64.b64encode(json.dumps(payload).encode()).decode() def claim(headers, params, tries=20): """Send the paid request; resend the SAME proof while the gateway says it is retryable.""" for _ in range(tries): rr = requests.get(ENDPOINT, params=params, headers={**UA, **headers}, timeout=90) if rr.status_code == 200: return rr try: retry = bool(rr.json().get("retryable")) except ValueError: retry = False if not retry: return rr time.sleep(3) return rr def pay_tx_receipt(pk, acct, body, req, params): """Non-standard `obolpay-tx-receipt`: used ONLY because this 402 lists it in accepts[].""" from web3 import Web3 # only this legacy path moves funds itself (you pay gas) w3 = Web3(Web3.HTTPProvider(RPC)) value = int(req["maxAmountRequired"]) if value > MAX_UNITS: raise SystemExit("price %d units exceeds MAX_UNITS=%d - refusing to pay" % (value, MAX_UNITS)) extra = req.get("extra") or {} invoice = extra.get("invoice_id") or body["payment"]["invoice_id"] domain = ((body.get("payment") or {}).get("signature_scheme") or {}).get("domain") or "pay.obolpay.xyz" erc20 = w3.eth.contract(address=Web3.to_checksum_address(req["asset"]), abi=[{ "name": "transfer", "type": "function", "stateMutability": "nonpayable", "inputs": [{"name": "to", "type": "address"}, {"name": "value", "type": "uint256"}], "outputs": [{"type": "bool"}]}]) tx = erc20.functions.transfer(Web3.to_checksum_address(req["payTo"]), value).build_transaction({ "from": acct.address, "nonce": w3.eth.get_transaction_count(acct.address), "chainId": CHAIN_IDS[req["network"]], "gas": 120000, "maxFeePerGas": w3.eth.gas_price * 2, "maxPriorityFeePerGas": w3.to_wei(0.001, "gwei")}) signed = acct.sign_transaction(tx) raw = getattr(signed, "raw_transaction", None) or signed.rawTransaction txh = w3.eth.send_raw_transaction(raw).hex() txh = txh if txh.startswith("0x") else "0x" + txh print("TX:", txh) w3.eth.wait_for_transaction_receipt(txh) msg = "x402:" + domain + ":" + invoice + ":" + txh.lower() # EIP-191 binding sig = Account.sign_message(encode_defunct(text=msg), pk).signature.hex() sig = sig if sig.startswith("0x") else "0x" + sig return claim({"X-Payment-Invoice-ID": invoice, "X-Payment-Tx-Hash": txh, "X-Payment-Signature": sig}, params, tries=40) def main(): pk = os.environ["PRIVATE_KEY"] acct = Account.from_key(pk) params = {"types": os.environ["TYPES"]} if os.environ.get("TYPES") else None # 1) Read the free preview + payment challenge (nothing is signed or spent yet) r = requests.get(ENDPOINT, params=params, headers=UA, timeout=30) if r.status_code != 402: raise SystemExit("expected HTTP 402, got %s: %s" % (r.status_code, r.text[:200])) body = r.json() print("PREVIEW:", (body.get("payment") or {}).get("preview")) # <- evaluate before paying accepts = body.get("accepts") or [] exact = next((a for a in accepts if a.get("scheme") == "exact"), None) legacy = next((a for a in accepts if a.get("scheme") == LEGACY), None) # 2) Pay with a scheme the 402 actually offers — standard `exact` first if exact is not None: rr = claim({"X-PAYMENT": exact_payment_header(acct, exact)}, params) elif legacy is not None: rr = pay_tx_receipt(pk, acct, body, legacy, params) else: raise SystemExit("no payment scheme offered right now (nothing signed or sent): %s" % (body.get("payment_unavailable") or body.get("message"))) # 3) Result if rr.status_code == 200: out = rr.json() print("DATA:", out["data"]) print("RECEIPT:", out.get("receipt")) # verify: POST {message, signature} to /verify-receipt return out print("REJECTED:", rr.status_code, rr.text[:400]) return None if __name__ == "__main__": main()