""" apitoll_client.py — generic Python client for any apitoll service. A single file. Two dependencies (`requests`, `eth-account`). No x402 SDK. Handles the v2 wire format directly so the call site stays clean: from apitoll_client import call, generate_wallet, ensure_wallet rate = call("fx-rates", base="USD", quote="EUR") bin_info = call("bin-lookup", bin="411111") country = call("country-data", iso2="ZA") holidays = call("holidays", iso2="JP", year=2026) geo = call("ip-geolocation", address="8.8.8.8") Wallet handling: addr = ensure_wallet() # creates .apitoll-wallet if missing, # prints address, returns it bal = check_balance() # returns dict with eth, usdc Production: rate = call("fx-rates", base="USD", quote="EUR", network="production") # default is "sandbox" (Base Sepolia, free test USDC) This is a Base Sepolia testnet client by default — agents and users should explicitly pass network="production" only after confirming intent with the user, since production calls spend real USDC. """ from __future__ import annotations import base64 import json import os import secrets import sys import time from pathlib import Path from typing import Any, Optional import requests from eth_account import Account from eth_account.messages import encode_typed_data # -------------------------------------------------------------------------- # Service registry. Single source of truth for what apitoll offers and how # to call each one. To add a service: add a row here, restart. # -------------------------------------------------------------------------- BASE_URL = os.environ.get("APITOLL_BASE_URL", "https://apitoll.io") SERVICES: dict[str, dict[str, Any]] = { "fx-rates": { "path": "/v1/sandbox/fx/{base}/{quote}", "inputs": ["base", "quote"], "description": "Daily ECB reference exchange rates. base/quote are 3-letter ISO codes.", "price_usdc": 0.001, }, "fx-rates-list": { "path": "/v1/sandbox/fx/{base}", "inputs": ["base"], "description": "All 30+ FX rates against a single base currency in one call.", "price_usdc": 0.001, }, "bin-lookup": { "path": "/v1/sandbox/bin/{bin}", "inputs": ["bin"], "description": "Card-issuer metadata for a 6-8 digit BIN. Returns scheme, type, country, bank.", "price_usdc": 0.002, }, "country-data": { "path": "/v1/sandbox/country/{iso2}", "inputs": ["iso2"], "description": "ISO 3166, ISO 4217, dialling code, capital, timezones for a country.", "price_usdc": 0.001, }, "country-data-resolve": { "path": "/v1/sandbox/country/resolve/{input}", "inputs": ["input"], "description": "Look up a country from a fuzzy input — name, alias, or ISO code.", "price_usdc": 0.001, }, "holidays": { "path": "/v1/sandbox/holidays/{iso2}/{year}", "inputs": ["iso2", "year"], "description": "Public, bank, and regional holidays for a country and year.", "price_usdc": 0.001, }, "ip-geolocation": { "path": "/v1/sandbox/ip/{address}", "inputs": ["address"], "description": "City, ASN, and coordinates for IPv4 or IPv6.", "price_usdc": 0.002, }, } WALLET_PATH = Path(os.environ.get("APITOLL_WALLET_PATH", ".apitoll-wallet")) USDC_BASE_SEPOLIA = "0x036CbD53842c5426634e7929541eC2318f3dCF7e" USDC_BASE_MAINNET = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913" CHAIN_ID_BASE_SEPOLIA = 84532 CHAIN_ID_BASE_MAINNET = 8453 # -------------------------------------------------------------------------- # Wallet # -------------------------------------------------------------------------- def generate_wallet(force: bool = False) -> str: """Generate a fresh sandbox-only wallet, write to .apitoll-wallet, return address.""" if WALLET_PATH.exists() and not force: cur = _load_wallet() return cur["address"] acct = Account.create() payload = { "private_key": acct.key.hex(), "address": acct.address, "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "purpose": "apitoll-test-wallet-base-sepolia-only", } WALLET_PATH.write_text(json.dumps(payload, indent=2)) try: WALLET_PATH.chmod(0o600) except OSError: pass return acct.address def ensure_wallet() -> str: """Generate-if-missing convenience. Returns the wallet address.""" return generate_wallet(force=False) def wallet_address() -> Optional[str]: if not WALLET_PATH.exists(): return None return _load_wallet()["address"] def _load_wallet() -> dict[str, str]: return json.loads(WALLET_PATH.read_text()) def _account(): return Account.from_key(_load_wallet()["private_key"]) # -------------------------------------------------------------------------- # Balance check # -------------------------------------------------------------------------- def check_balance(network: str = "sandbox") -> dict[str, Any]: """ Returns {address, eth_wei, eth, usdc_micro, usdc, network, sufficient_for_call}. Uses the public Base Sepolia / Base RPC. """ addr = wallet_address() if not addr: return {"error": "no wallet — call generate_wallet() first"} rpc = ("https://sepolia.base.org" if network == "sandbox" else "https://mainnet.base.org") usdc = USDC_BASE_SEPOLIA if network == "sandbox" else USDC_BASE_MAINNET eth_wei = int(_eth_call(rpc, "eth_getBalance", [addr, "latest"]), 16) # ERC-20 balanceOf(address) → 0x70a08231 + 32-byte address data = "0x70a08231" + addr.replace("0x", "").lower().rjust(64, "0") raw = _eth_call(rpc, "eth_call", [{"to": usdc, "data": data}, "latest"]) usdc_micro = int(raw, 16) return { "address": addr, "network": network, "eth_wei": eth_wei, "eth": eth_wei / 1e18, "usdc_micro": usdc_micro, "usdc": usdc_micro / 1_000_000, "sufficient_for_call": usdc_micro >= 1000, # $0.001 "explorer": f"https://sepolia.basescan.org/address/{addr}" if network == "sandbox" else f"https://basescan.org/address/{addr}", } def _eth_call(rpc_url: str, method: str, params: list) -> str: r = requests.post(rpc_url, json={"jsonrpc": "2.0", "method": method, "params": params, "id": 1}, timeout=10) r.raise_for_status() j = r.json() if "error" in j: raise RuntimeError(f"RPC error: {j['error']}") return j["result"] # -------------------------------------------------------------------------- # x402 round-trip # -------------------------------------------------------------------------- def call(service: str, network: str = "sandbox", **inputs) -> dict[str, Any]: """ Call any apitoll service. Returns dict with `data`, `settlement`, and `meta` keys. Raises ValueError on bad service/inputs, RuntimeError on payment failure (with explanation). network: "sandbox" (Base Sepolia, default) or "production" (Base mainnet). """ if service not in SERVICES: raise ValueError(f"unknown service '{service}'. Available: {sorted(SERVICES)}") cfg = SERVICES[service] missing = [k for k in cfg["inputs"] if k not in inputs] if missing: raise ValueError(f"missing input(s) for {service}: {missing}. Required: {cfg['inputs']}") path = cfg["path"].format(**inputs) if network == "production": path = path.replace("/sandbox/", "/") elif network != "sandbox": raise ValueError(f"network must be 'sandbox' or 'production', got '{network}'") url = BASE_URL + path return _x402_round_trip(url, network) def _x402_round_trip(url: str, network: str) -> dict[str, Any]: # 1. Initial GET → 402 r1 = requests.get(url, headers={"accept": "application/json"}) if r1.status_code != 402: raise RuntimeError(f"expected 402, got HTTP {r1.status_code}: {r1.text[:200]}") quote_b64 = r1.headers.get("PAYMENT-REQUIRED") or r1.headers.get("payment-required") if not quote_b64: raise RuntimeError("server did not return PAYMENT-REQUIRED header") quote = json.loads(_b64decode(quote_b64)) # 2. Pick the matching network option chain_id = CHAIN_ID_BASE_SEPOLIA if network == "sandbox" else CHAIN_ID_BASE_MAINNET network_caip2 = f"eip155:{chain_id}" option = next((a for a in quote.get("accepts", []) if a.get("scheme") == "exact" and a.get("network") == network_caip2), None) if not option: raise RuntimeError(f"no {network_caip2} option in server quote") # 3. Build EIP-712 typed data acct = _account() now = int(time.time()) valid_after = 0 valid_before = now + min(int(option.get("maxTimeoutSeconds") or 300), 600) nonce = "0x" + secrets.token_hex(32) eip712 = { "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": option.get("extra", {}).get("name", "USDC"), "version": option.get("extra", {}).get("version", "2"), "chainId": chain_id, "verifyingContract": option["asset"], }, "message": { "from": acct.address, "to": option["payTo"], "value": int(option["amount"]), "validAfter": valid_after, "validBefore": valid_before, "nonce": nonce, }, } signable = encode_typed_data(full_message=eip712) signed = acct.sign_message(signable) signature = signed.signature.hex() if not signature.startswith("0x"): signature = "0x" + signature # 4. Build v2 payment-payload (note: top-level resource + accepted, NOT scheme/network) payment_payload = { "x402Version": quote["x402Version"], "payload": { "authorization": { "from": eip712["message"]["from"], "to": eip712["message"]["to"], "value": str(eip712["message"]["value"]), "validAfter": str(eip712["message"]["validAfter"]), "validBefore": str(eip712["message"]["validBefore"]), "nonce": eip712["message"]["nonce"], }, "signature": signature, }, "resource": quote["resource"], "accepted": option, } header_value = base64.b64encode(json.dumps(payment_payload).encode("utf-8")).decode("ascii") # 5. Retry with PAYMENT-SIGNATURE header r2 = requests.get(url, headers={ "accept": "application/json", "PAYMENT-SIGNATURE": header_value, }) # Parse the settlement receipt regardless of status — we want to know # if the buyer paid even when the data-side returned an error. settle = None settle_b64 = (r2.headers.get("PAYMENT-RESPONSE") or r2.headers.get("X-PAYMENT-RESPONSE") or r2.headers.get("payment-response") or r2.headers.get("x-payment-response")) if settle_b64: try: settle = json.loads(_b64decode(settle_b64)) except Exception: settle = None # Distinguish payment failure (402) from data lookup failure (4xx with # successful settlement). The latter means the buyer paid for an # attempted lookup that came back negative — that's the contract for # most apitoll services (you pay for the lookup, not for a hit). if r2.status_code == 402: raise RuntimeError( f"payment rejected (HTTP 402). Server body: {r2.text[:200]}. " f"Possible causes: malformed signature, expired authorization, " f"insufficient USDC, network mismatch." ) try: body = r2.json() except Exception: body = {"raw": r2.text[:500]} explorer = None if settle and settle.get("transaction"): explorer = (f"https://sepolia.basescan.org/tx/{settle['transaction']}" if network == "sandbox" else f"https://basescan.org/tx/{settle['transaction']}") result = { "data": body, "settlement": settle, "meta": { "url": url, "network": network, "http_status": r2.status_code, "explorer": explorer, }, } # If the data-side failed but we paid, surface that explicitly so the # caller knows the call cost USDC even though the answer was negative. if r2.status_code != 200: result["error"] = f"HTTP {r2.status_code}" if isinstance(body, dict) and body.get("error"): result["error_detail"] = body["error"] return result # -------------------------------------------------------------------------- # helpers # -------------------------------------------------------------------------- def _b64decode(s: str) -> bytes: s = s.strip() pad = "=" * (-len(s) % 4) return (base64.urlsafe_b64decode(s + pad) if "_" in s or "-" in s else base64.b64decode(s + pad)) def list_services() -> dict[str, dict[str, Any]]: """Return a copy of the service registry — useful for agents enumerating capabilities.""" return {k: {**v} for k, v in SERVICES.items()} # -------------------------------------------------------------------------- # CLI — `python apitoll_client.py key=value [key=value ...]` # -------------------------------------------------------------------------- if __name__ == "__main__": if len(sys.argv) < 2 or sys.argv[1] in ("-h", "--help"): print("Usage: python apitoll_client.py [key=value ...]") print(" python apitoll_client.py --list") print(" python apitoll_client.py --balance") print(" python apitoll_client.py --address") sys.exit(0) cmd = sys.argv[1] if cmd == "--list": for k, v in SERVICES.items(): print(f" {k:24} ${v['price_usdc']:.3f} {v['description']}") sys.exit(0) if cmd == "--balance": print(json.dumps(check_balance(), indent=2)); sys.exit(0) if cmd == "--address": print(ensure_wallet()); sys.exit(0) kv = dict(arg.split("=", 1) for arg in sys.argv[2:] if "=" in arg) network = kv.pop("network", "sandbox") ensure_wallet() out = call(cmd, network=network, **kv) print(json.dumps(out, indent=2))