#!/usr/bin/env python3 """MCP server for Obolpay Gateway — compatible with x402™: expose the gateway as native tools for AI agents. x402™ is a trademark of LF Projects, LLC. Obolpay Gateway is an independent service that implements the open x402 payment protocol; it is not affiliated with or endorsed by LF Projects, LLC. Business use only. Terms: https://pay.obolpay.xyz/terms · Legal notice: https://pay.obolpay.xyz/legal Any MCP-compatible agent (Claude Desktop, etc.) can add this server to DISCOVER -> PREVIEW -> PURCHASE paid data and VERIFY receipts. purchase() spends real USDC from the wallet whose key you configure (at most X402_MAX_UNITS per call, default 0.10 USDC): configure only a key and a limit that the wallet's owner has approved. Install & run: pip install "mcp[cli]" eth-account requests # + web3 only for the legacy scheme # purchase() needs a funded Base wallet: export X402_AGENT_PRIVATE_KEY=0x... # holds >= the quoted USDC (standard `exact`: no ETH needed) export X402_BASE_URL=https://pay.obolpay.xyz # optional; this is the default python x402_mcp_server.py Claude Desktop config (claude_desktop_config.json): { "mcpServers": { "x402-obolpay": { "command": "python", "args": ["/absolute/path/to/x402_mcp_server.py"], "env": { "X402_AGENT_PRIVATE_KEY": "0x..." } } } } Tools: discover() -> the machine-readable service manifest (free) preview() -> the free data preview from the 402 challenge (free, no spend) purchase() -> pay the quoted USDC and return {data, receipt} (spends real USDC) verify_receipt(message, sig) -> third-party verification of a proof-of-purchase receipt purchase() pays ONLY with a scheme the live HTTP 402 lists in accepts[]: standard x402 `exact` first (sign an EIP-3009 authorization; the gateway's facilitator submits it and pays the gas), the non-standard `obolpay-tx-receipt` (send a transfer yourself) only while the 402 still lists it. If neither is offered it returns an error WITHOUT signing or sending anything. """ import base64 import json import os import secrets import time import requests from mcp.server.fastmcp import FastMCP BASE = os.environ.get("X402_BASE_URL", "https://pay.obolpay.xyz").rstrip("/") ENDPOINT = BASE + "/api/v1/protected-data" RPC = os.environ.get("X402_BASE_RPC_URL", "https://mainnet.base.org") UA = {"User-Agent": "x402-mcp/1.0"} # send a UA (Cloudflare blocks empty/raw-urllib) #: Never authorize more than this per call (atomic USDC units; 100000 = 0.10 USDC). MAX_UNITS = int(os.environ.get("X402_MAX_UNITS", "100000")) LEGACY_SCHEME = "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"}]} mcp = FastMCP("x402-obolpay") @mcp.tool() def discover() -> dict: """Return the gateway's machine-readable manifest: price, token, network, recipient, payment flow, free-preview and proof-of-purchase capabilities. No payment required.""" return requests.get(BASE + "/.well-known/x402", headers=UA, timeout=30).json() @mcp.tool() def preview() -> dict: """Fetch the HTTP 402 challenge and return its FREE preview (a sample of the paid dataset) plus the price/invoice, so the agent can decide whether to pay. No payment required.""" r = requests.get(ENDPOINT, headers=UA, timeout=30) if r.status_code != 402: return {"error": f"expected 402, got {r.status_code}", "body": r.text[:400]} body = r.json() p = body.get("payment", {}) accepts = body.get("accepts") or [] return { "preview": p.get("preview"), "price": {"amount": p.get("amount"), "token": p.get("token"), "network": p.get("network")}, "invoice_id": p.get("invoice_id"), "recipient": p.get("recipient") or next((a.get("payTo") for a in accepts if a.get("payTo")), None), "payment_schemes": [a.get("scheme") for a in accepts], } def _exact_payment_header(acct, req: dict) -> str: """Sign an EIP-3009 transferWithAuthorization for one `exact` accepts[] entry (x402 v1). The EIP-712 domain is the USDC contract's own (name "USD Coin" — not "USDC" — on Base). Nothing moves here: the gateway's facilitator submits the signed authorization.""" from eth_account.messages import encode_typed_data network, value = req["network"], int(req["maxAmountRequired"]) if network not in CHAIN_IDS: raise ValueError(f"unsupported network {network!r}") if value > MAX_UNITS: raise ValueError(f"price {value} units exceeds X402_MAX_UNITS={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)} 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: dict, tries: int = 20): """Send the paid request; resend the SAME proof while the gateway says it is retryable.""" rr = None for _ in range(tries): rr = requests.get(ENDPOINT, 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 _purchase_tx_receipt(pk: str, acct, body: dict, req: dict) -> dict: """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) from eth_account import Account from eth_account.messages import encode_defunct value = int(req["maxAmountRequired"]) if value > MAX_UNITS: return {"error": f"price {value} units exceeds X402_MAX_UNITS={MAX_UNITS}; not paying"} w3 = Web3(Web3.HTTPProvider(RPC)) ch = body.get("payment") or {} invoice = (req.get("extra") or {}).get("invoice_id") or ch["invoice_id"] domain = (ch.get("signature_scheme") or {}).get("domain") or (BASE.split("://", 1)[-1]) 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() if not txh.startswith("0x"): txh = "0x" + txh w3.eth.wait_for_transaction_receipt(txh) msg = "x402:" + domain + ":" + invoice + ":" + txh.lower() sig = Account.sign_message(encode_defunct(text=msg), pk).signature.hex() if not sig.startswith("0x"): sig = "0x" + sig rr = _claim({"X-Payment-Invoice-ID": invoice, "X-Payment-Tx-Hash": txh, "X-Payment-Signature": sig}, tries=40) if rr is not None and rr.status_code == 200: out = rr.json() return {"scheme": LEGACY_SCHEME, "tx_hash": txh, "data": out.get("data"), "receipt": out.get("receipt")} return {"error": f"rejected: {rr.status_code if rr is not None else '?'}", "body": rr.text[:400] if rr is not None else "", "tx_hash": txh} @mcp.tool() def purchase() -> dict: """Pay the quoted USDC on Base and return the unlocked {data, receipt}. Pays only with a scheme the live 402 lists in accepts[]: standard x402 `exact` first (you sign an EIP-3009 authorization; the gateway's facilitator submits it and pays the gas); the non-standard `obolpay-tx-receipt` (you send the transfer and pay gas) only if the 402 lists it. Requires env X402_AGENT_PRIVATE_KEY (a Base wallet with >= the quoted USDC). WARNING: this spends real USDC.""" pk = os.environ.get("X402_AGENT_PRIVATE_KEY") if not pk: return {"error": "X402_AGENT_PRIVATE_KEY not set; cannot pay."} from eth_account import Account acct = Account.from_key(pk) r = requests.get(ENDPOINT, headers=UA, timeout=30) if r.status_code != 402: return {"error": f"expected 402, got {r.status_code}"} body = r.json() 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_SCHEME), None) if exact is not None: try: header = _exact_payment_header(acct, exact) except ValueError as exc: return {"error": f"not paying: {exc}"} rr = _claim({"X-PAYMENT": header}) if rr is not None and rr.status_code == 200: out = rr.json() return {"scheme": "exact", "tx_hash": out.get("tx_hash"), "data": out.get("data"), "receipt": out.get("receipt")} return {"error": f"rejected: {rr.status_code if rr is not None else '?'}", "body": rr.text[:400] if rr is not None else ""} if legacy is not None: return _purchase_tx_receipt(pk, acct, body, legacy) return {"error": "no payment scheme offered right now; nothing was signed or sent", "detail": body.get("payment_unavailable") or body.get("message")} @mcp.tool() def verify_receipt(message: str, signature: str) -> dict: """Verify a proof-of-purchase receipt with the gateway (recovers the EIP-191 signer and checks it equals the server's receipt signer). Anyone can call this — no payment required.""" return requests.post(BASE + "/verify-receipt", headers={**UA, "Content-Type": "application/json"}, json={"message": message, "signature": signature}, timeout=30).json() # ---- Account balance (gasless vouchers) — top-ups only where the gateway offers them ---- def _agent_address() -> str: from eth_account import Account return Account.from_key(os.environ["X402_AGENT_PRIVATE_KEY"]).address.lower() def _topup_offered() -> bool: """Does the gateway currently accept top-ups? Its manifest lists `prepaid_balance` only then.""" try: m = requests.get(BASE + "/.well-known/x402", headers=UA, timeout=30).json() except Exception: # noqa: BLE001 - unknown means "do not assume top-ups exist" return False return isinstance(m, dict) and isinstance(m.get("prepaid_balance"), dict) @mcp.tool() def balance(address: str = "") -> dict: """Check an account balance at the gateway: balance_units, calls_remaining, and next_nonce. Pass an address, or leave blank to use the wallet from X402_AGENT_PRIVATE_KEY.""" addr = (address or "").strip().lower() if not addr: if not os.environ.get("X402_AGENT_PRIVATE_KEY"): return {"error": "provide address or set X402_AGENT_PRIVATE_KEY"} addr = _agent_address() return requests.get(BASE + "/account/" + addr, headers=UA, timeout=30).json() @mcp.tool() def topup(tx_hash: str) -> dict: """Credit a balance from an on-chain USDC deposit — ONLY if discover() lists `prepaid_balance`. If it does not, the gateway accepts no top-ups: do NOT send USDC for one (it would not be credited); pay per call with purchase() instead. This tool checks first and refuses otherwise.""" if not _topup_offered(): return {"error": "topup_not_offered", "detail": "This gateway does not accept top-ups right now (no `prepaid_balance` in " "its manifest). Do not send USDC for a top-up; use purchase() to pay per call."} return requests.post(BASE + "/account/topup", headers={**UA, "Content-Type": "application/json"}, json={"tx_hash": tx_hash}, timeout=60).json() @mcp.tool() def spend_gasless() -> dict: """Fetch the paid data by drawing from an existing balance — GASLESS, NO on-chain tx. Works with any balance the gateway holds for your wallet (earlier top-ups or automatic refund credits). Requires X402_AGENT_PRIVATE_KEY with a funded balance; otherwise use purchase().""" from eth_account import Account from eth_account.messages import encode_defunct pk = os.environ["X402_AGENT_PRIVATE_KEY"] addr = _agent_address() acc = requests.get(BASE + "/account/" + addr, headers=UA, timeout=30).json() if acc.get("balance_units", 0) < acc.get("price_units", 1): return {"error": "insufficient_balance", "account": acc, "hint": ("call topup(tx_hash) after depositing USDC" if _topup_offered() else "top-ups are not offered; use purchase() to pay per call")} # 署名するメッセージは**サーバの文字列をそのまま署名しない**。ローカルで # 期待形 `x402-spend:{domain}:{address}:{nonce}:{amount_units}` を組み直し、 # サーバが返した voucher_message がそれと一致するときだけ、組み直した方に # 署名する。X402_BASE_URL は差し替え可能なので、悪意あるゲートウェイに # 向ければ「任意の文字列」への personal_sign を資金入りウォレットから # 引き出せてしまう（別ドメイン/別額のバウチャーや、別サービスで有効な # レシート主張文など）。これはそれを断つ。 from urllib.parse import urlparse domain = urlparse(BASE).hostname or "" expected = f"x402-spend:{domain}:{addr}:{acc['next_nonce']}:{acc['price_units']}" if acc.get("voucher_message") != expected: return {"error": "voucher_mismatch", "detail": "Server voucher_message does not match the expected " "x402-spend:{domain}:{address}:{nonce}:{amount} for this account. " "Refusing to sign an unexpected message.", "expected": expected, "got": acc.get("voucher_message")} sig = Account.sign_message(encode_defunct(text=expected), pk).signature.hex() if not sig.startswith("0x"): sig = "0x" + sig headers = {**UA, "X-Account-Address": addr, "X-Account-Nonce": str(acc["next_nonce"]), "X-Account-Voucher": sig} return requests.get(ENDPOINT, headers=headers, timeout=30).json() def _hide_tools_the_gateway_does_not_offer() -> None: """At startup, drop the topup tool from the tool list unless the gateway accepts top-ups. An agent that sees a `topup` tool will reasonably send USDC first and call it second; if the gateway does not accept top-ups, that transfer is never credited. Unknown (gateway unreachable) counts as "not offered". The function stays importable and still refuses on its own.""" if _topup_offered(): return remover = getattr(mcp, "remove_tool", None) if remover is not None: try: remover("topup") except Exception: # noqa: BLE001 - already absent / older SDK: topup() still self-refuses pass if __name__ == "__main__": _hide_tools_the_gateway_does_not_offer() mcp.run()