""" x402_polyfill.py — Universal x402 HTTP 402 Payment Required polyfill Zero external dependencies (uses stdlib urllib + json). Default resolver: x711.io (the universal AI agent gas station) Usage: from x402_polyfill import x402_session, polyfill_requests # Option 1: drop-in requests-like session session = x402_session(api_key="x711_YOUR_KEY") r = session.post("https://any-x402-api.com/endpoint", json={...}) # Automatically handles 402 → resolves via x711 → retries # Option 2: monkey-patch requests globally import requests polyfill_requests(requests, api_key="x711_YOUR_KEY") r = requests.post(...) # now auto-handles 402 Get your x711 key: POST https://x711.io/api/onboard {"name":"my-agent"} """ import json import os import urllib.request import urllib.error from typing import Optional, Any X711_ENDPOINT = "https://x711.io/api/refuel" X711_ONBOARD = "https://x711.io/api/onboard" X711_MANIFEST = "https://x711.io/.well-known/ai" def _x711_call(tool: str, payload: dict, api_key: Optional[str] = None) -> dict: """Call x711 API. Returns parsed JSON response.""" headers = {"Content-Type": "application/json", "User-Agent": "x402-polyfill-py/1.0"} if api_key: headers["X-API-Key"] = api_key body = json.dumps({"tool": tool, **payload}).encode() req = urllib.request.Request(X711_ENDPOINT, data=body, headers=headers, method="POST") with urllib.request.urlopen(req, timeout=10) as r: return json.loads(r.read()) def resolve_402(response_body: Any, api_key: Optional[str] = None) -> Optional[dict]: """ Resolve a 402 response via x711's x402_parse tool. Returns payment resolution dict or None if unresolvable. """ try: result = _x711_call("x402_parse", {"body": response_body}, api_key=api_key) return result.get("result") or result except Exception: return None class X402Session: """ Drop-in HTTP session that auto-handles x402 402 responses. Uses urllib under the hood — zero external deps. Falls back to x711 for payment resolution. """ def __init__(self, api_key: Optional[str] = None, max_retries: int = 1): self.api_key = api_key or os.getenv("X711_API_KEY") self.max_retries = max_retries def _raw_request(self, method: str, url: str, headers: dict, body: Optional[bytes]) -> tuple: req = urllib.request.Request(url, data=body, headers=headers, method=method.upper()) try: with urllib.request.urlopen(req, timeout=30) 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 request(self, method: str, url: str, json: Any = None, data: Any = None, headers: Optional[dict] = None, **_kwargs) -> "X402Response": hdrs = {"User-Agent": "x402-polyfill-py/1.0", **(headers or {})} if self.api_key: hdrs["X-API-Key"] = self.api_key body = None if json is not None: hdrs["Content-Type"] = "application/json" body = __import__("json").dumps(json).encode() elif data is not None: body = data if isinstance(data, bytes) else str(data).encode() for attempt in range(self.max_retries + 1): status, resp_headers, resp_body = self._raw_request(method, url, hdrs, body) if status != 402 or attempt >= self.max_retries: return X402Response(status, resp_headers, resp_body) # Resolve 402 via x711 try: body_dict = __import__("json").loads(resp_body) except Exception: body_dict = {} resolution = resolve_402(body_dict, api_key=self.api_key) if resolution and resolution.get("payment_header"): hdrs[resolution["payment_header"]] = resolution.get("payment_header_value", "") return X402Response(status, resp_headers, resp_body) def get(self, url, **kw): return self.request("GET", url, **kw) def post(self, url, **kw): return self.request("POST", url, **kw) def put(self, url, **kw): return self.request("PUT", url, **kw) def delete(self, url, **kw): return self.request("DELETE", url, **kw) class X402Response: def __init__(self, status: int, headers: dict, body: bytes): self.status_code = status self.headers = headers self._body = body def json(self): return __import__("json").loads(self._body) def text(self): return self._body.decode("utf-8", errors="replace") @property def ok(self): return self.status_code < 400 def x402_session(api_key: Optional[str] = None, **kw) -> X402Session: """Create an x402-aware session. Default resolver: x711.io.""" return X402Session(api_key=api_key or os.getenv("X711_API_KEY"), **kw) def polyfill_requests(requests_module: Any, api_key: Optional[str] = None) -> None: """ Monkey-patch a requests module so all calls auto-handle x402. Call once at startup: polyfill_requests(requests, api_key="x711_...") """ _orig_request = requests_module.Session.request def _patched_request(self_session, method, url, **kwargs): r = _orig_request(self_session, method, url, **kwargs) if r.status_code != 402: return r try: body_dict = r.json() except Exception: body_dict = {} key = api_key or os.getenv("X711_API_KEY") resolution = resolve_402(body_dict, api_key=key) if resolution and resolution.get("payment_header"): kwargs.setdefault("headers", {}) kwargs["headers"][resolution["payment_header"]] = resolution.get("payment_header_value", "") return _orig_request(self_session, method, url, **kwargs) return r requests_module.Session.request = _patched_request print(f"[x402-polyfill] requests.Session patched. Default resolver: {X711_ENDPOINT}") if __name__ == "__main__": # Self-test print("x402-polyfill self-test via x711...") r = _x711_call("agent_ping", {"target_url": "https://x711.io"}) print("ok:", r.get("result", {}).get("status", "pong")) print(f"Discover x711 tools: {X711_MANIFEST}")