#!/usr/bin/env python3 """Complete x402/EIP-3009 purchase against the Energy Research Hub. Install: pip install requests web3 eth-account Run: PRIVATE_KEY=0x... python buy-python.py """ import base64 import json import os import secrets import sys import time from typing import Any import requests from eth_account import Account from eth_account.messages import encode_typed_data from web3 import Web3 API_URL = os.getenv( "API_URL", "https://energy.netzhandwerker.de/predict/negative-price", ) RPC_URL = os.getenv("BASE_RPC_URL", "https://mainnet.base.org") MAX_AMOUNT_UNITS = int(os.getenv("MAX_AMOUNT_UNITS", "1000")) REQUEST_BODY = json.loads(os.getenv("REQUEST_BODY_JSON", "{}")) BALANCE_OF_ABI = [ { "inputs": [{"name": "account", "type": "address"}], "name": "balanceOf", "outputs": [{"name": "", "type": "uint256"}], "stateMutability": "view", "type": "function", } ] class BuyerError(Exception): def __init__(self, code: str, message: str, details: Any = None): super().__init__(f"{code}: {message}") self.details = details def decode_base64_json(value: str, label: str) -> dict[str, Any]: if not value: raise BuyerError("MISSING_HEADER", f"{label} is missing") try: result = json.loads(base64.b64decode(value).decode("utf-8")) except Exception as exc: raise BuyerError("INVALID_HEADER", f"{label} is not base64 JSON", str(exc)) from exc if not isinstance(result, dict): raise BuyerError("INVALID_HEADER", f"{label} does not decode to a JSON object") return result def response_json(response: requests.Response) -> Any: try: return response.json() except ValueError: return response.text def request(payment_header: str | None = None) -> requests.Response: headers = { "Accept": "application/json", "Content-Type": "application/json", "User-Agent": "energy-hub-python-x402-example/1.0", } if payment_header: headers["PAYMENT-SIGNATURE"] = payment_header return requests.post(API_URL, headers=headers, json=REQUEST_BODY, timeout=60) def parse_challenge(response: requests.Response) -> dict[str, Any]: if response.status_code != 402: raise BuyerError("EXPECTED_402", f"expected HTTP 402, received HTTP {response.status_code}", response_json(response)) encoded = response.headers.get("Payment-Required") challenge = decode_base64_json(encoded, "Payment-Required") if encoded else response_json(response) if not isinstance(challenge, dict) or not isinstance(challenge.get("accepts"), list): raise BuyerError("INVALID_CHALLENGE", "challenge has no accepts[]") return challenge def choose_accept(challenge: dict[str, Any]) -> dict[str, Any]: for item in challenge["accepts"]: if ( item.get("scheme") == "exact" and str(item.get("network", "")).startswith("eip155:") and item.get("extra", {}).get("assetTransferMethod") == "eip3009" ): return item raise BuyerError("UNSUPPORTED_CHALLENGE", "no exact EIP-3009 option was offered") def token_balance(address: str, asset: str) -> int: web3 = Web3(Web3.HTTPProvider(RPC_URL, request_kwargs={"timeout": 30})) if not web3.is_connected(): raise BuyerError("RPC_ERROR", f"could not connect to {RPC_URL}") contract = web3.eth.contract(address=Web3.to_checksum_address(asset), abi=BALANCE_OF_ABI) return int(contract.functions.balanceOf(Web3.to_checksum_address(address)).call()) def sign_payment( private_key: str, challenge: dict[str, Any], accept: dict[str, Any], ) -> tuple[str, dict[str, Any]]: account = Account.from_key(private_key) amount = int(str(accept.get("amount", "0"))) if amount <= 0: raise BuyerError("INVALID_AMOUNT", f"challenge amount must be positive, got {amount}") if amount > MAX_AMOUNT_UNITS: raise BuyerError("AMOUNT_LIMIT", f"challenge amount {amount} exceeds MAX_AMOUNT_UNITS={MAX_AMOUNT_UNITS}") balance = token_balance(account.address, str(accept["asset"])) if balance < amount: raise BuyerError( "INSUFFICIENT_USDC", f"wallet has {balance} atomic units but challenge requires {amount}; no payment header was sent", ) domain = accept.get("extra", {}).get("eip712Domain", {}) required_domain_fields = ("name", "version", "chainId", "verifyingContract") missing = [field for field in required_domain_fields if not domain.get(field)] if missing: raise BuyerError("INVALID_CHALLENGE", f"EIP-712 domain is missing: {', '.join(missing)}") domain = dict(domain) domain["chainId"] = int(domain["chainId"]) if str(domain["verifyingContract"]).lower() != str(accept["asset"]).lower(): raise BuyerError("INVALID_CHALLENGE", "EIP-712 verifyingContract does not match accepts[].asset") now = int(time.time()) authorization = { "from": account.address, "to": str(accept["payTo"]), "value": str(amount), "validAfter": "0", "validBefore": str(now + int(accept["maxTimeoutSeconds"])), "nonce": "0x" + secrets.token_hex(32), } if int(authorization["validBefore"]) <= now: raise BuyerError("EXPIRED_AUTHORIZATION", "validBefore is not in the future; request a fresh challenge") 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": domain, "message": authorization, } signed = Account.sign_message(encode_typed_data(full_message=typed_data), private_key=private_key) signature = "0x" + signed.signature.hex().removeprefix("0x") payload = { "x402Version": int(challenge.get("x402Version", 2)), "resource": challenge.get("resource"), "accepted": accept, "payload": {"signature": signature, "authorization": authorization}, } encoded = base64.b64encode( json.dumps(payload, separators=(",", ":")).encode("utf-8") ).decode("ascii") details = { **authorization, "v": signed.v, "r": hex(signed.r), "s": hex(signed.s), "header_format": "base64(JSON({x402Version,resource,accepted,payload:{signature,authorization}}))", } return encoded, details def main() -> int: private_key = os.getenv("PRIVATE_KEY") if not private_key: raise BuyerError("MISSING_PRIVATE_KEY", "set PRIVATE_KEY in the environment") first = request() challenge = parse_challenge(first) accept = choose_accept(challenge) print( "1/4 HTTP 402 received", json.dumps( { "amount_field": "accepts[0].amount", "amount_atomic_units": accept["amount"], "pay_to": accept["payTo"], "asset": accept["asset"], "network": accept["network"], "docs": challenge.get("docs"), "signing_guide": challenge.get("signing_guide"), } ), ) payment_header, signature_details = sign_payment(private_key, challenge, accept) print("2/4 EIP-3009 TransferWithAuthorization signed", json.dumps(signature_details)) print("3/4 PAYMENT-SIGNATURE built as base64 JSON") paid = request(payment_header) paid_body = response_json(paid) if paid.status_code != 200: raise BuyerError( "PAID_REQUEST_FAILED", f"retry returned HTTP {paid.status_code}; expired signatures, wrong amounts, reused nonces, and insufficient balance are rejected", paid_body, ) payment_response_value = paid.headers.get("Payment-Response") or paid.headers.get("X-Payment-Response") payment_response = decode_base64_json(payment_response_value, "Payment-Response") if payment_response_value else None print("4/4 HTTP 200 received", json.dumps({"payment_response": payment_response, "data": paid_body})) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except BuyerError as exc: print(str(exc), file=sys.stderr) if exc.details is not None: print(json.dumps(exc.details, indent=2, default=str), file=sys.stderr) raise SystemExit(1)