"""Budget-bounded x402 Library buyer. Dry-run by default; never stores a key.""" from __future__ import annotations import base64 import json import os import sys import urllib.error import urllib.request from datetime import datetime, timezone from decimal import Decimal from typing import Any from urllib.parse import urljoin, urlparse NETWORK = "eip155:8453" USDC = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913" PAY_TO = "0xcdBd1625fb843491ae855c0EA110C07492aEcFb3" def atomic(usdc: str) -> int: value = Decimal(usdc) if value < 0 or value.as_tuple().exponent < -6: raise ValueError("USDC budget must be non-negative with at most six decimals") return int(value * 1_000_000) def request_json(url: str, method: str = "GET", body: dict[str, Any] | None = None) -> tuple[int, dict[str, str], Any]: data = json.dumps(body).encode() if body is not None else None request = urllib.request.Request(url, data=data, method=method, headers={"Accept": "application/json", "Content-Type": "application/json"}) try: with urllib.request.urlopen(request, timeout=10) as response: return response.status, dict(response.headers), json.loads(response.read()) except urllib.error.HTTPError as error: raw = error.read() parsed = json.loads(raw) if raw else None return error.code, dict(error.headers), parsed def same_origin(url: str, origin: str) -> str: resolved = urljoin(origin, url) if urlparse(resolved).netloc != urlparse(origin).netloc or urlparse(resolved).scheme != "https": raise ValueError("discovery attempted to change the trusted origin") return resolved def decode_required(value: str) -> dict[str, Any]: padded = value + "=" * (-len(value) % 4) return json.loads(base64.urlsafe_b64decode(padded)) def header_value(headers: dict[str, str], name: str) -> str | None: return next((value for key, value in headers.items() if key.lower() == name.lower()), None) def main() -> None: origin = os.getenv("X402LIBRARY_ORIGIN", "https://x402library.com") max_usdc = os.getenv("MAX_PAYMENT_USDC", "0.01") max_session = os.getenv("MAX_SESSION_SPEND_USDC", "0.01") input_body = { "asset": os.getenv("ASSET", "BTCUSDT"), "intended_action": os.getenv("INTENDED_ACTION", "ENTER_LONG"), "risk_tolerance": os.getenv("RISK_TOLERANCE", "BALANCED"), "max_data_age_seconds": int(os.getenv("MAX_DATA_AGE_SECONDS", "60")), "max_price_usdc": float(max_usdc), } status, _, pointer = request_json(f"{origin}/.well-known/x402.json") if status != 200: raise RuntimeError(f"well-known discovery failed: {status}") discovery_url = same_origin(pointer["discovery_contract"], origin) status, _, discovery = request_json(discovery_url) if status != 200: raise RuntimeError(f"discovery failed: {status}") products = [product for product in discovery["products"] if product.get("offer_endpoint")] product = next((product for product in products if product["id"] == "market.action-safety-managed-v1"), None) if product is None: raise RuntimeError("managed action-safety product was not discovered") offer_url = same_origin(product["offer_endpoint"], origin) paid_url = same_origin(product["endpoint"], origin) status, _, offer = request_json(offer_url, "POST", input_body) if status != 200: raise RuntimeError(f"offer failed: {status}") if not offer["purchase_allowed"]: print(json.dumps({"status": "PURCHASE_NOT_ALLOWED", "gate": offer["purchase_gate"]}, indent=2)) return if offer["data_status"]["availability"] != "LIVE" or offer["data_status"]["age_seconds"] > input_body["max_data_age_seconds"]: raise RuntimeError("offer violates freshness policy") valid_until = datetime.fromisoformat(offer["valid_until"].replace("Z", "+00:00")) if valid_until <= datetime.now(timezone.utc): raise RuntimeError("offer TTL has expired") if atomic(offer["price"]["amount"]) > min(atomic(max_usdc), atomic(max_session)): raise RuntimeError("offer exceeds buyer budget") status, headers, _ = request_json(paid_url, "POST", input_body) encoded_required = header_value(headers, "PAYMENT-REQUIRED") if status != 402 or not encoded_required: raise RuntimeError(f"expected a 402 payment challenge, received {status}") payment = decode_required(encoded_required) if payment["x402Version"] != 2 or payment["resource"]["url"] != paid_url: raise RuntimeError("402 protocol version or resource URL does not match") allowed = next((item for item in payment["accepts"] if item["scheme"] == "exact" and item["network"] == NETWORK and item["asset"].lower() == USDC.lower() and item["payTo"].lower() == PAY_TO.lower() and int(item["amount"]) <= min(atomic(max_usdc), atomic(max_session)) and 0 < item["maxTimeoutSeconds"] <= 300), None) if allowed is None: raise RuntimeError("402 challenge contains no policy-approved payment") if os.getenv("X402_AUTO_PAY") != "true": print(json.dumps({"status": "PAYMENT_APPROVAL_READY", "product_id": product["id"], "amount_atomic": allowed["amount"], "network": NETWORK, "pay_to": PAY_TO}, indent=2)) return private_key = os.getenv("EVM_PRIVATE_KEY") if not private_key: raise RuntimeError("X402_AUTO_PAY=true requires EVM_PRIVATE_KEY") # Optional payment path: pip install 'x402[requests,evm]'. from eth_account import Account from x402 import x402ClientSync from x402.http.clients import x402_requests from x402.mechanisms.evm import EthAccountSigner from x402.mechanisms.evm.exact.register import register_exact_evm_client client = x402ClientSync() register_exact_evm_client(client, EthAccountSigner(Account.from_key(private_key))) def payment_policy(_version: int, requirements: list[Any]) -> list[Any]: return [item for item in requirements if item.network == NETWORK and item.scheme == "exact" and item.asset.lower() == USDC.lower() and item.pay_to.lower() == PAY_TO.lower() and int(item.amount) <= min(atomic(max_usdc), atomic(max_session))] client.register_policy(payment_policy) request_id = header_value(headers, "X-X402LIBRARY-Request-Id") with x402_requests(client) as session: response = session.post(paid_url, json=input_body, timeout=10, headers={"X-X402LIBRARY-Request-Id": request_id} if request_id else {}) response.raise_for_status() if not response.headers.get("PAYMENT-RESPONSE"): raise RuntimeError("paid response omitted settlement receipt") print(json.dumps({"status": "PAID_RESULT_DELIVERED", "result": response.json()}, indent=2)) if __name__ == "__main__": try: main() except Exception as error: # Buyer fails closed with one concise reason. print(json.dumps({"status": "BUYER_STOPPED", "error": str(error)}, indent=2), file=sys.stderr) raise SystemExit(1)