"""x402 discovery, preview, and buyer execution helpers.""" from __future__ import annotations import base64 import hashlib import json import logging import time from types import SimpleNamespace from typing import Any from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from agent_wallet.config import normalize_solana_network, resolve_solana_rpc_url from agent_wallet.exceptions import ProviderError from agent_wallet.http_client import get_client from agent_wallet.wallet_layer.base import AgentWalletBackend CDP_BAZAAR_DISCOVERY_BASE_URL = "https://api.cdp.coinbase.com/platform/v2/x402/discovery" AGENTIC_MARKET_API_BASE_URL = "https://api.agentic.market/v1" X402_EXECUTE_TIMEOUT_SECONDS = 45.0 DISCOVERY_CACHE_TTL_SECONDS = 300.0 _DISCOVERY_CACHE_MAX_ENTRIES = 64 # Discovery responses cached as JSON text so hits hand out independent copies. # Long-lived only inside the resident read worker; cold CLI runs start empty. _discovery_cache: dict[str, tuple[float, str]] = {} SOLANA_CAIP_BY_NETWORK = { "mainnet": "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp", } EVM_CAIP_BY_NETWORK = { "ethereum": "eip155:1", "base": "eip155:8453", } # x402 v1 requirements carry legacy network names instead of CAIP-2 ids. LEGACY_NETWORK_TO_CAIP = { "ethereum": "eip155:1", "base": "eip155:8453", "solana": "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp", } _USDC_IDENTIFIERS = { "usdc", "usd coin", "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913", "0x036cbd53842c5426634e7929541ec2318f3dcf7e", "epjfwdd5aufqssqem2qn1xzybapc8g4wegkgkzwytdt1v", } log = logging.getLogger("agent_wallet.x402") def _backend_chain(backend: AgentWalletBackend) -> str: chain = _trim(getattr(backend, "chain", "")).lower() if chain: return chain try: capabilities = backend.get_capabilities() except Exception: return "" return _trim(getattr(capabilities, "chain", "")).lower() def _backend_network(backend: AgentWalletBackend) -> str: return _trim(getattr(backend, "network", "")).lower() def _backend_solana_sdk_rpc_url(backend: AgentWalletBackend) -> str | None: candidates = getattr(backend, "rpc_urls", None) if isinstance(candidates, list): for candidate in candidates: text = _trim(candidate) if text.startswith(("http://", "https://")): return text primary = _trim(getattr(backend, "rpc_url", None)) if primary.startswith(("http://", "https://")): return primary network = _backend_network(backend) fallback = resolve_solana_rpc_url(normalize_solana_network(network or "mainnet"), "") return _trim(fallback) or None def _trim(value: Any) -> str: return str(value or "").strip() def _canonical_json_text(payload: Any) -> str: return json.dumps(payload, sort_keys=True, separators=(",", ":")) def _hash_text(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest() def _normalize_discovery_provider(value: Any) -> str: provider = _trim(value).lower() or "auto" aliases = { "bazaar": "cdp_bazaar", "cdp": "cdp_bazaar", "agentic": "agentic_market", "agenticmarket": "agentic_market", "market": "agentic_market", } provider = aliases.get(provider, provider) if provider not in {"auto", "cdp_bazaar", "agentic_market"}: raise ProviderError("x402-discovery", f"Unsupported discovery provider: {provider}") return provider def _normalize_http_method(value: Any) -> str: method = _trim(value).upper() or "GET" if method not in {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}: raise ProviderError("x402-http", f"Unsupported HTTP method: {method}") return method def _normalize_headers(value: Any) -> dict[str, str]: if value is None: return {} if not isinstance(value, dict): raise ProviderError("x402-http", "headers must be an object when provided.") headers: dict[str, str] = {} for key, raw_value in value.items(): name = _trim(key) if not name: raise ProviderError("x402-http", "headers must not contain empty names.") headers[name] = str(raw_value) return headers def _normalize_query_params(value: Any) -> dict[str, str]: if value is None: return {} if not isinstance(value, dict): raise ProviderError("x402-http", "query must be an object when provided.") params: dict[str, str] = {} for key, raw_value in value.items(): name = _trim(key) if not name: raise ProviderError("x402-http", "query must not contain empty names.") params[name] = str(raw_value) return params def _append_query(url: str, query: dict[str, str]) -> str: if not query: return url parts = urlsplit(url) merged = dict(parse_qsl(parts.query, keep_blank_values=True)) merged.update(query) return urlunsplit( (parts.scheme, parts.netloc, parts.path, urlencode(merged, doseq=True), parts.fragment) ) def _request_host(url: str) -> str: try: return _trim(urlsplit(url).netloc).lower() except Exception: return "" def _response_text(response: Any) -> str: try: text = response.text except Exception: return "" return str(text or "") def _parse_json_response(response: Any, *, provider: str, operation: str) -> Any: body = _response_text(response) if not body.strip(): raise ProviderError(provider, f"{operation} returned an empty response body.") try: return response.json() except ValueError as exc: snippet = body.strip().replace("\n", " ")[:200] detail = f": {snippet}" if snippet else "" raise ProviderError(provider, f"{operation} returned invalid JSON{detail}") from exc def _decode_payment_required(header_value: str) -> dict[str, Any]: raw = _trim(header_value) if not raw: raise ProviderError("x402-http", "PAYMENT-REQUIRED header is empty.") decoded_bytes: bytes | None = None for decoder in (base64.b64decode, base64.urlsafe_b64decode): try: padding = "=" * (-len(raw) % 4) decoded_bytes = decoder(raw + padding) break except Exception: continue if decoded_bytes is None: raise ProviderError("x402-http", "PAYMENT-REQUIRED header is not valid base64.") try: payload = json.loads(decoded_bytes.decode("utf-8")) except Exception as exc: raise ProviderError("x402-http", "PAYMENT-REQUIRED header is not valid JSON.") from exc if isinstance(payload, list): accepts = payload x402_version = None elif isinstance(payload, dict): accepts = payload.get("accepts") x402_version = payload.get("x402Version") else: raise ProviderError("x402-http", "PAYMENT-REQUIRED payload must decode to JSON.") if not isinstance(accepts, list) or not accepts: raise ProviderError("x402-http", "PAYMENT-REQUIRED payload does not contain accepts[].") return { "x402_version": x402_version, "accepts": accepts, "raw": payload, "encoded": raw, } def _caip_network(value: Any) -> str: """Map a requirement network to CAIP-2 form, passing CAIP ids through.""" network = _trim(value) if ":" in network: return network return LEGACY_NETWORK_TO_CAIP.get(network.lower(), network) def _decode_payment_required_body(response: Any) -> dict[str, Any] | None: """Parse a legacy x402 v1 402 response whose requirements live in the JSON body. Returns the same shape as _decode_payment_required, with `encoded` set to a base64 form the x402 SDK can decode (it detects the version from the JSON). """ try: payload = response.json() except Exception: return None if not isinstance(payload, dict) or payload.get("x402Version") != 1: return None accepts = payload.get("accepts") if not isinstance(accepts, list) or not accepts: return None encoded = base64.b64encode(json.dumps(payload).encode("utf-8")).decode("ascii") return { "x402_version": 1, "accepts": accepts, "raw": payload, "encoded": encoded, } def _extract_requirement_extra(requirement: dict[str, Any]) -> dict[str, Any]: extra = requirement.get("extra") return dict(extra) if isinstance(extra, dict) else {} def _requirement_field(requirement: Any, field_name: str) -> Any: if isinstance(requirement, dict): aliases = { "pay_to": ("pay_to", "payTo"), "max_timeout_seconds": ("max_timeout_seconds", "maxTimeoutSeconds"), "amount": ("amount", "maxAmountRequired"), } for candidate in aliases.get(field_name, (field_name,)): if candidate in requirement: return requirement.get(candidate) return None value = getattr(requirement, field_name, None) if value is None and field_name == "amount": # v1 SDK models expose the amount as max_amount_required / get_amount(). getter = getattr(requirement, "get_amount", None) if callable(getter): return getter() return value def _requirement_amount(requirement: dict[str, Any]) -> str: # x402 v2 uses "amount"; v1 uses "maxAmountRequired". return _trim(requirement.get("amount") or requirement.get("maxAmountRequired")) def _looks_like_usdc(requirement: dict[str, Any]) -> bool: asset = _trim(requirement.get("asset")).lower() if asset in _USDC_IDENTIFIERS: return True extra = _extract_requirement_extra(requirement) return _trim(extra.get("name")).lower() in _USDC_IDENTIFIERS def _normalize_amount_hint(requirement: dict[str, Any]) -> str | None: amount = _requirement_amount(requirement) if not amount.isdigit(): return None if _looks_like_usdc(requirement): raw = int(amount) whole = raw // 1_000_000 fraction = raw % 1_000_000 if fraction: return f"{whole}.{fraction:06d}".rstrip("0").rstrip(".") return f"{whole}" return None def normalize_payment_requirement( requirement: dict[str, Any], *, source: str, resource_url: str | None = None, ) -> dict[str, Any]: extra = _extract_requirement_extra(requirement) amount = _requirement_amount(requirement) return { "scheme": _trim(requirement.get("scheme")).lower() or None, "network": _trim(requirement.get("network")) or None, "asset": _trim(requirement.get("asset")) or None, "amount": amount or None, "amount_display": _normalize_amount_hint(requirement), "pay_to": _trim(requirement.get("payTo")) or None, "max_timeout_seconds": requirement.get("maxTimeoutSeconds"), "resource_url": resource_url, "extra": extra, "source": source, "raw": requirement, } def _normalize_cdp_resource(item: dict[str, Any]) -> dict[str, Any]: metadata = item.get("metadata") metadata_dict = dict(metadata) if isinstance(metadata, dict) else {} accepts = item.get("accepts") if isinstance(item.get("accepts"), list) else [] resource = _trim(item.get("resource")) return { "discovery_provider": "cdp_bazaar", "resource": resource, "type": _trim(item.get("type")) or "http", "x402_version": item.get("x402Version"), "description": _trim(item.get("description") or metadata_dict.get("description")) or None, "last_updated": item.get("lastUpdated"), "accepts": [ normalize_payment_requirement(requirement, source="cdp_bazaar", resource_url=resource) for requirement in accepts if isinstance(requirement, dict) ], "metadata": metadata_dict, "raw": item, } def _normalize_agentic_service(service: dict[str, Any]) -> dict[str, Any]: endpoints = service.get("endpoints") if isinstance(service.get("endpoints"), list) else [] normalized_endpoints: list[dict[str, Any]] = [] accepts: list[dict[str, Any]] = [] for endpoint in endpoints: if not isinstance(endpoint, dict): continue pricing = endpoint.get("pricing") if isinstance(endpoint.get("pricing"), dict) else {} normalized = { "url": _trim(endpoint.get("url")) or None, "description": _trim(endpoint.get("description")) or None, "method": _normalize_http_method(endpoint.get("method")), "pricing": { "amount": _trim(pricing.get("amount")) or None, "currency": _trim(pricing.get("currency")) or None, "network": _trim(pricing.get("network")) or None, }, } normalized_endpoints.append(normalized) amount_text = _trim(pricing.get("amount")) currency = _trim(pricing.get("currency")).lower() network = _trim(pricing.get("network")).lower() if amount_text and currency == "usdc": accepts.append( { "scheme": "exact", "network": EVM_CAIP_BY_NETWORK.get(network, SOLANA_CAIP_BY_NETWORK.get(network, network)), "asset": "USDC", "amount": amount_text, "amount_display": amount_text, "pay_to": None, "max_timeout_seconds": None, "resource_url": normalized["url"], "extra": { "marketplace": "agentic_market", "pricingCurrency": pricing.get("currency"), "pricingNetwork": pricing.get("network"), }, "source": "agentic_market", "raw": endpoint, } ) return { "discovery_provider": "agentic_market", "service_id": _trim(service.get("id")) or None, "service_name": _trim(service.get("name")) or None, "description": _trim(service.get("description")) or None, "domain": _trim(service.get("domain")) or None, "category": _trim(service.get("category")) or None, "networks": [str(item) for item in service.get("networks") or []], "integration_type": _trim(service.get("integrationType")) or None, "is_new": bool(service.get("isNew")), "endpoints": normalized_endpoints, "accepts": accepts, "raw": service, } def _wallet_caip_networks(backend: AgentWalletBackend) -> list[str]: chain = _backend_chain(backend) network = _backend_network(backend) if chain == "evm": caip = EVM_CAIP_BY_NETWORK.get(network) return [caip] if caip else [] if chain == "solana": caip = SOLANA_CAIP_BY_NETWORK.get(network) return [caip] if caip else [] return [] def _solana_exact_execution_supported(backend: AgentWalletBackend) -> bool: return ( _backend_chain(backend) == "solana" and _backend_network(backend) == "mainnet" and getattr(backend, "signer", None) is not None ) def _evm_exact_execution_supported(backend: AgentWalletBackend) -> bool: return ( _backend_chain(backend) == "evm" and _backend_network(backend) == "base" and callable(getattr(backend, "sign_x402_evm_exact_typed_data", None)) ) def _evm_payment_requirement_supported(requirement: dict[str, Any]) -> bool: scheme = _trim(requirement.get("scheme")).lower() extra = _extract_requirement_extra(requirement) if scheme == "exact": transfer_method = _trim(extra.get("assetTransferMethod")).lower() return transfer_method in {"", "eip3009", "transferwithauthorization"} if scheme == "upto": # The upto Permit2 payload needs the facilitator's address up front; # without it the SDK signer raises instead of falling back cleanly. return bool(_trim(extra.get("facilitatorAddress"))) return False def _wallet_x402_support_summary(backend: AgentWalletBackend) -> dict[str, Any]: chain = _backend_chain(backend) network = _backend_network(backend) supported_networks = _wallet_caip_networks(backend) planned_execution_networks = { "eip155:8453", "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp", } execution_modes: list[str] = [] if _solana_exact_execution_supported(backend): execution_modes.append("solana_exact") if _evm_exact_execution_supported(backend): execution_modes.append("evm_exact") return { "chain": chain, "network": network, "supported_caip_networks": supported_networks, "wallet_type_supported": chain in {"evm", "solana"}, "execution_available": bool(execution_modes), "execution_modes": execution_modes, "planned_execution_networks": sorted(planned_execution_networks), } def _requirement_compatibility(requirement: dict[str, Any], backend: AgentWalletBackend) -> dict[str, Any]: wallet_summary = _wallet_x402_support_summary(backend) network = _caip_network(requirement.get("network")) scheme = _trim(requirement.get("scheme")).lower() wallet_network_matches = network in wallet_summary["supported_caip_networks"] chain = _backend_chain(backend) planned_execution_supported = False currently_executable = False if chain == "solana": planned_execution_supported = scheme == "exact" and network in set( wallet_summary["planned_execution_networks"] ) currently_executable = ( planned_execution_supported and wallet_network_matches and _solana_exact_execution_supported(backend) ) elif chain == "evm": planned_execution_supported = ( scheme in {"exact", "upto"} and network in set(wallet_summary["planned_execution_networks"]) and _evm_payment_requirement_supported(requirement) ) currently_executable = ( planned_execution_supported and wallet_network_matches and _evm_exact_execution_supported(backend) ) if currently_executable: reason = ( "Executable now through the local Solana exact buyer flow." if chain == "solana" else f"Executable now through the local EVM {scheme or 'exact'} buyer flow." ) elif chain == "evm" and scheme == "exact" and not _evm_payment_requirement_supported(requirement): reason = "This EVM exact payment requires a transfer method that is not enabled in the current wallet runtime." elif chain == "evm" and scheme == "upto" and not _evm_payment_requirement_supported(requirement): reason = "This EVM upto payment is missing a facilitatorAddress in its extra data, so it cannot be signed." elif planned_execution_supported and wallet_network_matches: reason = "Wallet network matches, but this backend does not yet expose a supported x402 signer path." elif planned_execution_supported: reason = "Planned execution path exists, but the requirement targets a different network than the active wallet." else: reason = "Unsupported scheme or network for the planned execution path." return { "wallet_type_supported": wallet_summary["wallet_type_supported"], "wallet_network_matches": wallet_network_matches, "planned_execution_supported": planned_execution_supported, "currently_executable": currently_executable, "reason": reason, } def _select_preferred_requirement( requirements: list[dict[str, Any]], backend: AgentWalletBackend, ) -> dict[str, Any] | None: compatible: list[dict[str, Any]] = [] exact_match: list[dict[str, Any]] = [] for requirement in requirements: compatibility = _requirement_compatibility(requirement, backend) if not compatibility["planned_execution_supported"]: continue compatible.append(requirement) if compatibility["wallet_network_matches"]: exact_match.append(requirement) candidates = exact_match or compatible if not candidates: return None def sort_key(item: dict[str, Any]) -> tuple[int, int, int]: # Prefer upto over exact: upto settles for actual usage against a # signed cap, which fits metered/usage-priced endpoints better than # exact's fixed charge. Ties within a scheme still favor the cheapest # declared amount. scheme_rank = 0 if _trim(item.get("scheme")).lower() == "upto" else 1 amount = _trim(item.get("amount")) if amount.isdigit(): return (scheme_rank, 0, int(amount)) return (scheme_rank, 1, 0) return sorted(candidates, key=sort_key)[0] def _build_request_metadata( *, url: str, method: str = "GET", headers: dict[str, Any] | None = None, query: dict[str, Any] | None = None, json_body: Any | None = None, text_body: str | None = None, ) -> dict[str, Any]: request_url = _trim(url) if not request_url: raise ProviderError("x402-http", "url is required.") http_method = _normalize_http_method(method) normalized_headers = _normalize_headers(headers) normalized_query = _normalize_query_params(query) if json_body is not None and text_body is not None: raise ProviderError("x402-http", "Provide either json_body or text_body, not both.") final_url = _append_query(request_url, normalized_query) body_hash = None content_type = None if json_body is not None: body_hash = _hash_text(_canonical_json_text(json_body)) content_type = normalized_headers.get("Content-Type") or normalized_headers.get("content-type") if not content_type: normalized_headers["Content-Type"] = "application/json" elif text_body is not None: body_hash = _hash_text(text_body) content_type = normalized_headers.get("Content-Type") or normalized_headers.get("content-type") request_fingerprint = _hash_text( _canonical_json_text( { "method": http_method, "url": final_url, "body_hash": body_hash, } ) ) return { "url": final_url, "host": _request_host(final_url), "method": http_method, "headers": normalized_headers, "query": normalized_query, "json_body": json_body, "text_body": text_body, "body_hash": body_hash, "content_type": content_type, "request_fingerprint": request_fingerprint, } async def _send_request( *, client: Any, request: dict[str, Any], extra_headers: dict[str, str] | None = None, timeout: float | None = None, ) -> Any: headers = dict(request["headers"]) if extra_headers: headers.update(extra_headers) return await client.request( request["method"], request["url"], headers=headers, json=request["json_body"] if request["json_body"] is not None else None, content=request["text_body"] if request["text_body"] is not None else None, timeout=timeout, ) def _build_x402_action_payload( *, backend: AgentWalletBackend, request: dict[str, Any], wallet_summary: dict[str, Any], address: str | None, status_code: int, selected_payment: dict[str, Any] | None, accepted_payments: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: return { "asset_type": "x402-request", "source": "x402-http", "chain": _backend_chain(backend), "network": _backend_network(backend), "x402_network": selected_payment.get("network") if isinstance(selected_payment, dict) else None, "x402_scheme": selected_payment.get("scheme") if isinstance(selected_payment, dict) else None, "x402_asset": selected_payment.get("asset") if isinstance(selected_payment, dict) else None, "x402_amount": selected_payment.get("amount") if isinstance(selected_payment, dict) else None, "x402_amount_display": selected_payment.get("amount_display") if isinstance(selected_payment, dict) else None, "x402_pay_to": selected_payment.get("pay_to") if isinstance(selected_payment, dict) else None, "request_url": request["url"], "method": request["method"], "request_fingerprint": request["request_fingerprint"], "body_hash": request["body_hash"], "content_type": request["content_type"], "wallet": { **wallet_summary, "address": address, }, "status_code": status_code, "selected_payment": selected_payment, "accepted_payments": accepted_payments, "payment_required": selected_payment is not None, } def _response_preview(response: Any) -> Any: try: return response.json() except Exception: return _response_text(response)[:2000] def _parse_payment_required_response( response: Any, *, backend: AgentWalletBackend, request: dict[str, Any], address: str | None, wallet_summary: dict[str, Any], ) -> dict[str, Any]: payment_required = response.headers.get("PAYMENT-REQUIRED") if payment_required: decoded = _decode_payment_required(payment_required) else: decoded = _decode_payment_required_body(response) if decoded is None: raise ProviderError( "x402-http", "Server returned HTTP 402 without a PAYMENT-REQUIRED header or an x402 v1 JSON body.", details={"status_code": response.status_code, "url": request["url"]}, ) normalized_accepts = [ normalize_payment_requirement(requirement, source="payment_required", resource_url=request["url"]) for requirement in decoded["accepts"] if isinstance(requirement, dict) ] compatibility = [ { **requirement, "compatibility": _requirement_compatibility(requirement, backend), } for requirement in normalized_accepts ] selected = _select_preferred_requirement(normalized_accepts, backend) preview = _build_x402_action_payload( backend=backend, request=request, wallet_summary=wallet_summary, address=address, status_code=response.status_code, selected_payment=selected, accepted_payments=compatibility, ) preview.update( { "execute_available": bool( isinstance(selected, dict) and _requirement_compatibility(selected, backend)["currently_executable"] ), "x402_version": decoded["x402_version"], "response_headers": { "payment-required": decoded["encoded"], "content-type": response.headers.get("content-type"), }, } ) return preview def _validate_payment_requirement( selected: dict[str, Any] | None, *, backend: AgentWalletBackend, request_url: str, ) -> dict[str, Any]: if not isinstance(selected, dict): raise ProviderError( "x402-validate", "This endpoint returned HTTP 402 but no compatible payment option was found for the active wallet.", details={ "request_url": request_url, "wallet_chain": _backend_chain(backend), "wallet_network": _backend_network(backend), }, ) scheme = _trim(selected.get("scheme")).lower() if scheme not in {"exact", "upto"}: raise ProviderError( "x402-validate", f"Unsupported x402 payment scheme '{scheme or 'unknown'}'. Only 'exact' and 'upto' are supported.", details={"request_url": request_url, "selected_payment": selected}, ) if not _trim(selected.get("pay_to")): raise ProviderError( "x402-validate", "Payment destination (payTo) is missing from the x402 requirement.", details={"request_url": request_url, "selected_payment": selected}, ) compatibility = _requirement_compatibility(selected, backend) if compatibility["currently_executable"]: return selected chain = _backend_chain(backend) or "unknown" network = _backend_network(backend) or "unknown" requirement_network = _caip_network(selected.get("network")) or "unknown" if chain == "solana" and requirement_network not in SOLANA_CAIP_BY_NETWORK.values(): message = ( f"This endpoint requires payment on {requirement_network}, but the active wallet is Solana ({network})." ) elif chain == "evm" and requirement_network not in EVM_CAIP_BY_NETWORK.values(): message = ( f"This endpoint requires payment on {requirement_network}, but the active wallet is EVM ({network})." ) else: message = str(compatibility["reason"]) raise ProviderError( "x402-validate", message, details={ "request_url": request_url, "selected_payment": selected, "compatibility": compatibility, }, ) def _validate_request_execution_policy( *, request: dict[str, Any], backend: AgentWalletBackend, ) -> None: host = _trim(request.get("host")).lower() if host == "x402.alchemy.com": headers = request.get("headers") has_auth = isinstance(headers, dict) and any( str(key).strip().lower() == "authorization" and str(value).strip() for key, value in headers.items() ) if not has_auth: raise ProviderError( "x402-validate", ( "Alchemy's x402 gateway needs wallet-auth headers in addition to the payment challenge. " "The generic x402 tool does not mint Alchemy SIWE/SIWS auth tokens yet, so this endpoint " "is not safe to execute through the generic flow." ), details={ "request_url": request.get("url"), "host": host, "wallet_chain": _backend_chain(backend), "wallet_network": _backend_network(backend), "hint": "Use a dedicated Alchemy agent gateway integration or authenticated CLI flow.", }, ) def _select_sdk_payment_requirement( payment_required: Any, *, selected_payment: dict[str, Any], ) -> Any: accepts = getattr(payment_required, "accepts", None) if not isinstance(accepts, list) or not accepts: raise ProviderError("x402-http", "Decoded x402 payment payload does not contain accepts[].") selected_raw = selected_payment.get("raw") if isinstance(selected_payment, dict) else None if isinstance(selected_raw, dict): for requirement in accepts: model_dump = getattr(requirement, "model_dump", None) if callable(model_dump): dumped = model_dump(by_alias=True, exclude_none=True) if dumped == selected_raw: return requirement for requirement in accepts: if ( _trim(_requirement_field(requirement, "scheme")).lower() == _trim(selected_payment.get("scheme")).lower() and _trim(_requirement_field(requirement, "network")) == _trim(selected_payment.get("network")) and _trim(_requirement_field(requirement, "asset")) == _trim(selected_payment.get("asset")) and _trim(_requirement_field(requirement, "amount")) == _trim(selected_payment.get("amount")) and _trim(_requirement_field(requirement, "pay_to")) == _trim(selected_payment.get("pay_to")) ): return requirement if len(accepts) == 1: return accepts[0] raise ProviderError( "x402-http", "Could not match the selected x402 payment back to the decoded PAYMENT-REQUIRED payload.", details={ "selected_payment": selected_payment, "accepts_count": len(accepts), }, ) # The `upto` Permit2 signature authorizes the (merchant + facilitator) pair to # settle any amount in [0, cap] until `max_timeout_seconds` elapses, with no # on-chain check beyond the deadline itself. A merchant-declared timeout with # no upper bound would extend that facilitator-discretion window indefinitely, # so cap it defensively before signing. `exact` amounts are fixed in the # signature, so this cap only applies to `upto`. MAX_UPTO_DEADLINE_SECONDS = 7 * 24 * 60 * 60 # 7 days def _cap_upto_deadline(requirement: Any) -> Any: if _trim(getattr(requirement, "scheme", "")).lower() != "upto": return requirement timeout = getattr(requirement, "max_timeout_seconds", None) if not isinstance(timeout, int) or timeout <= MAX_UPTO_DEADLINE_SECONDS: return requirement return requirement.model_copy(update={"max_timeout_seconds": MAX_UPTO_DEADLINE_SECONDS}) def _build_selected_payment_required_payload( payment_required: Any, *, selected_payment: dict[str, Any], ) -> Any: selected_requirement = _select_sdk_payment_requirement( payment_required, selected_payment=selected_payment, ) selected_requirement = _cap_upto_deadline(selected_requirement) return payment_required.model_copy(update={"accepts": [selected_requirement]}) def _load_x402_common_sdk() -> dict[str, Any]: try: from x402 import x402Client from x402.http.x402_http_client_base import x402HTTPClientBase from x402.http.utils import decode_payment_required_header except ImportError as exc: raise ProviderError( "x402-sdk", "x402 execution requires the x402 Python package with HTTP client support.", details={"hint": 'Install dependencies so `x402[httpx]` is available in the wallet runtime.'}, ) from exc return { "x402Client": x402Client, "x402HTTPClientBase": x402HTTPClientBase, "decode_payment_required_header": decode_payment_required_header, } def _load_x402_solana_sdk() -> dict[str, Any]: sdk = _load_x402_common_sdk() try: from x402.mechanisms.svm.exact import register_exact_svm_client except ImportError as exc: raise ProviderError( "x402-sdk", "x402 Solana execution requires SVM support.", details={"hint": 'Install dependencies so `x402[httpx,svm]` is available in the wallet runtime.'}, ) from exc sdk.update( { "register_exact_svm_client": register_exact_svm_client, } ) return sdk def _load_x402_evm_sdk() -> dict[str, Any]: sdk = _load_x402_common_sdk() try: from x402.mechanisms.evm.exact import register_exact_evm_client except ImportError as exc: raise ProviderError( "x402-sdk", "x402 EVM execution requires EVM support.", details={"hint": 'Install dependencies so `x402[httpx,evm]` is available in the wallet runtime.'}, ) from exc try: from x402.mechanisms.evm.upto import UptoEvmScheme except ImportError as exc: raise ProviderError( "x402-sdk", "x402 EVM upto-scheme execution requires the upto mechanism module.", details={"hint": 'Install dependencies so `x402[httpx,evm]` is available in the wallet runtime.'}, ) from exc sdk.update( { "register_exact_evm_client": register_exact_evm_client, "UptoEvmScheme": UptoEvmScheme, } ) return sdk def _load_x402_sdk() -> dict[str, Any]: return _load_x402_common_sdk() def _load_x402_siwx_sdk() -> dict[str, Any]: try: from x402.extensions.sign_in_with_x import ( SIGN_IN_WITH_X, CreateSIWxClientExtensionOptions, create_siwx_client_extension, ) except ImportError as exc: raise ProviderError( "x402-sdk", "x402 sign-in-with-x execution requires the SIWx extension module.", details={"hint": 'Install dependencies so `x402[httpx,evm,svm]` is available in the wallet runtime.'}, ) from exc return { "SIGN_IN_WITH_X": SIGN_IN_WITH_X, "CreateSIWxClientExtensionOptions": CreateSIWxClientExtensionOptions, "create_siwx_client_extension": create_siwx_client_extension, } async def _maybe_attach_siwx_header( *, payment_required: Any, signer: Any, headers: dict[str, str], ) -> dict[str, str]: """Attach a Sign-In-With-X header when the merchant declares that extension. Some x402 gateways require proving wallet ownership via SIWx alongside the payment signature; without it those requests are rejected even with a valid payment. Best-effort: any failure to build the SIWx credential falls back to paying without it, matching the upstream SDK's own on_payment_required hook contract (which swallows exceptions the same way). """ extensions = getattr(payment_required, "extensions", None) or {} if not extensions: return headers try: sdk = _load_x402_siwx_sdk() except ProviderError: return headers if sdk["SIGN_IN_WITH_X"] not in extensions: return headers extension = sdk["create_siwx_client_extension"]( sdk["CreateSIWxClientExtensionOptions"](signers=[signer]) ) context = SimpleNamespace(payment_required=payment_required) try: result = await extension.transport_hooks.http.on_payment_required(None, context) except Exception as exc: log.warning( "x402 SIWx header build failed; continuing without it", extra={"error_type": type(exc).__name__, "error": str(exc) or None}, ) return headers siwx_headers = getattr(result, "headers", None) if result is not None else None if not siwx_headers: return headers return {**headers, **siwx_headers} def _build_solana_sdk_signer(backend: AgentWalletBackend) -> Any: signer = getattr(backend, "signer", None) if signer is None or not hasattr(signer, "export_keypair_bytes"): raise ProviderError( "x402-solana", "The active Solana backend does not expose a local signer for x402 payments.", ) try: from solders.keypair import Keypair except ImportError as exc: raise ProviderError( "x402-solana", "Solders is required for Solana x402 signing.", ) from exc class _OpenClawSolanaX402Signer: def __init__(self, wallet_signer: Any): self._wallet_signer = wallet_signer self._keypair = Keypair.from_bytes(wallet_signer.export_keypair_bytes()) @property def address(self) -> str: return str(self._wallet_signer.address) @property def keypair(self) -> Any: return self._keypair def sign_transaction(self, tx: Any) -> Any: tx.sign([self._keypair]) return tx return _OpenClawSolanaX402Signer(signer) def _build_evm_sdk_signer(backend: AgentWalletBackend, address: str) -> Any: sign_typed_data = getattr(backend, "sign_x402_evm_exact_typed_data", None) if not callable(sign_typed_data): raise ProviderError( "x402-evm", "The active EVM backend does not expose an x402 exact typed-data signer.", ) class _NoEthAccountSentinel: """Truthy placeholder with neither `.sign_message` nor `.address`. x402's SIWx sign_evm_message() treats a signer with `.account` unset but both `.sign_message` and `.address` present as an eth_account-style wallet, and calls `.sign_message(SignableMessage)` positionally instead of our keyword-based `sign_message(message=..., account=...)`. Exposing a benign `.account` here short-circuits that fallback so the SDK reaches our intended call path. """ class _OpenClawEvmX402Signer: def __init__(self, wallet_backend: AgentWalletBackend, wallet_address: str): self._wallet_backend = wallet_backend self._address = wallet_address self.account = _NoEthAccountSentinel() @property def address(self) -> str: return self._address async def sign_message(self, *, message: str, account: Any = None) -> str: return await self._wallet_backend.sign_message(message) def sign_typed_data( self, domain: Any, types: dict[str, list[Any]], primary_type: str, message: dict[str, Any], ) -> bytes: normalized_types: dict[str, list[dict[str, str]]] = {} for type_name, fields in types.items(): normalized_types[type_name] = [ { "name": _trim(getattr(field, "name", "")), "type": _trim(getattr(field, "type", "")), } for field in fields ] domain_payload = { "name": getattr(domain, "name", None), "version": getattr(domain, "version", None), "chainId": getattr(domain, "chain_id", None), "verifyingContract": getattr(domain, "verifying_contract", None), } return self._wallet_backend.sign_x402_evm_exact_typed_data( domain=domain_payload, types=normalized_types, primary_type=primary_type, message=message, ) return _OpenClawEvmX402Signer(backend, address) async def _create_payment_headers( *, backend: AgentWalletBackend, payment_required_header: str, selected_payment: dict[str, Any], ) -> dict[str, str]: chain = _backend_chain(backend) if chain == "solana": sdk = _load_x402_solana_sdk() payment_required = sdk["decode_payment_required_header"](payment_required_header) selected_payload = _build_selected_payment_required_payload( payment_required, selected_payment=selected_payment, ) client = sdk["x402Client"]() sdk_rpc_url = _backend_solana_sdk_rpc_url(backend) if not sdk_rpc_url: raise ProviderError( "x402-solana", "No direct Solana RPC URL is available for the x402 SDK signer path.", details={"network": _backend_network(backend)}, ) solana_signer = _build_solana_sdk_signer(backend) sdk["register_exact_svm_client"]( client, solana_signer, networks=str(selected_payment["network"]), rpc_url=sdk_rpc_url, ) try: payment_payload = await client.create_payment_payload(selected_payload) except Exception as exc: raise ProviderError( "x402-solana", "Failed to build the Solana x402 payment payload.", details={ "network": _backend_network(backend), "sdk_rpc_url": sdk_rpc_url, "error_type": type(exc).__name__, "error": str(exc) or None, }, ) from exc headers = sdk["x402HTTPClientBase"]().encode_payment_signature_header(payment_payload) return await _maybe_attach_siwx_header( payment_required=payment_required, signer=solana_signer, headers=headers ) if chain == "evm": sdk = _load_x402_evm_sdk() payment_required = sdk["decode_payment_required_header"](payment_required_header) selected_payload = _build_selected_payment_required_payload( payment_required, selected_payment=selected_payment, ) client = sdk["x402Client"]() address = await backend.get_address() if not isinstance(address, str) or not address.strip(): raise ProviderError("x402-evm", "The active EVM backend did not resolve a payer address.") evm_signer = _build_evm_sdk_signer(backend, address.strip()) network = str(selected_payment["network"]) sdk["register_exact_evm_client"](client, evm_signer, networks=network) # `upto` reuses the same typed-data signer as `exact`; registering both # lets the SDK settle whichever scheme the merchant's accepted # requirement (already narrowed to one entry above) turns out to use. client.register(network, sdk["UptoEvmScheme"](evm_signer)) payment_payload = await client.create_payment_payload(selected_payload) headers = sdk["x402HTTPClientBase"]().encode_payment_signature_header(payment_payload) return await _maybe_attach_siwx_header( payment_required=payment_required, signer=evm_signer, headers=headers ) raise ProviderError( "x402-http", "Only Solana and EVM buyer flows are executable in this milestone.", details={"chain": chain}, ) def _extract_settlement_header(response: Any) -> dict[str, Any]: sdk = _load_x402_sdk() settle = sdk["x402HTTPClientBase"]().get_payment_settle_response( lambda name: response.headers.get(name) ) return settle.model_dump(by_alias=True, exclude_none=True) def _extract_settlement_header_safe(response: Any) -> dict[str, Any] | None: try: return _extract_settlement_header(response) except Exception as exc: log.warning( "x402 settlement header parse failed", extra={ "status_code": getattr(response, "status_code", None), "payment_response": response.headers.get("PAYMENT-RESPONSE") if hasattr(response, "headers") else None, "x_payment_response": response.headers.get("X-PAYMENT-RESPONSE") if hasattr(response, "headers") else None, "error_type": type(exc).__name__, "error": str(exc) or None, }, ) return None def _log_x402_execute( *, request: dict[str, Any], selected_payment: dict[str, Any] | None, response: Any, settlement: dict[str, Any] | None, ) -> None: log.info( "x402 execute completed", extra={ "url": request.get("url"), "method": request.get("method"), "request_fingerprint": request.get("request_fingerprint"), "x402_network": selected_payment.get("network") if isinstance(selected_payment, dict) else None, "x402_asset": selected_payment.get("asset") if isinstance(selected_payment, dict) else None, "x402_amount": selected_payment.get("amount") if isinstance(selected_payment, dict) else None, "x402_pay_to": selected_payment.get("pay_to") if isinstance(selected_payment, dict) else None, "status_code": getattr(response, "status_code", None), "transaction": settlement.get("transaction") if isinstance(settlement, dict) else None, "confirmed": bool(settlement and settlement.get("success")), }, ) def _discovery_cache_get(cache_key: str) -> dict[str, Any] | None: entry = _discovery_cache.get(cache_key) if entry is None: return None expires_at, payload_text = entry if expires_at <= time.monotonic(): _discovery_cache.pop(cache_key, None) return None return json.loads(payload_text) def _discovery_cache_set(cache_key: str, payload: dict[str, Any]) -> None: try: payload_text = json.dumps(payload) except (TypeError, ValueError): return if len(_discovery_cache) >= _DISCOVERY_CACHE_MAX_ENTRIES: for stale_key in sorted(_discovery_cache, key=lambda key: _discovery_cache[key][0])[ : len(_discovery_cache) - _DISCOVERY_CACHE_MAX_ENTRIES + 1 ]: _discovery_cache.pop(stale_key, None) _discovery_cache[cache_key] = (time.monotonic() + DISCOVERY_CACHE_TTL_SECONDS, payload_text) async def search_services( *, query: str | None = None, discovery_provider: str = "auto", network: str | None = None, asset: str | None = None, scheme: str | None = None, max_usd_price: str | None = None, limit: int = 10, ) -> dict[str, Any]: provider = _normalize_discovery_provider(discovery_provider) if provider == "auto": provider = "cdp_bazaar" if limit <= 0: raise ProviderError("x402-discovery", "limit must be greater than zero.") cache_key = _canonical_json_text( { "op": "search_services", "provider": provider, "query": _trim(query), "network": _trim(network), "asset": _trim(asset), "scheme": _trim(scheme), "max_usd_price": _trim(max_usd_price), "limit": limit, } ) cached = _discovery_cache_get(cache_key) if cached is not None: return cached result = await _search_services_uncached( query=query, provider=provider, network=network, asset=asset, scheme=scheme, max_usd_price=max_usd_price, limit=limit, ) _discovery_cache_set(cache_key, result) return result async def _search_services_uncached( *, query: str | None, provider: str, network: str | None, asset: str | None, scheme: str | None, max_usd_price: str | None, limit: int, ) -> dict[str, Any]: client = get_client() if provider == "cdp_bazaar": if query and _trim(query): # CDP rejects empty filter values with HTTP 400, so omit unset params # entirely instead of sending them as blank strings. search_params: dict[str, Any] = {"query": _trim(query), "limit": min(limit, 20)} for name, value in ( ("network", network), ("asset", asset), ("scheme", scheme), ("maxUsdPrice", max_usd_price), ): if _trim(value): search_params[name] = _trim(value) response = await client.get( f"{CDP_BAZAAR_DISCOVERY_BASE_URL}/search", params=search_params, ) payload = _parse_json_response( response, provider="x402-cdp-bazaar", operation="CDP Bazaar search" ) resources = payload.get("resources") if isinstance(payload, dict) else None if not isinstance(resources, list): raise ProviderError("x402-cdp-bazaar", "Unexpected CDP Bazaar search response.") items = [_normalize_cdp_resource(item) for item in resources if isinstance(item, dict)] return { "discovery_provider": provider, "query": _trim(query), "count": len(items), "partial_results": bool(payload.get("partialResults")), "search_method": payload.get("searchMethod"), "items": items, } response = await client.get( f"{CDP_BAZAAR_DISCOVERY_BASE_URL}/resources", params={"type": "http", "limit": min(limit, 1000), "offset": 0}, ) payload = _parse_json_response( response, provider="x402-cdp-bazaar", operation="CDP Bazaar resources" ) resources = payload.get("items") if isinstance(payload, dict) else None if not isinstance(resources, list): raise ProviderError("x402-cdp-bazaar", "Unexpected CDP Bazaar resources response.") items = [_normalize_cdp_resource(item) for item in resources if isinstance(item, dict)] return { "discovery_provider": provider, "query": "", "count": len(items), "pagination": payload.get("pagination") if isinstance(payload, dict) else None, "items": items, } endpoint = "/services/search" if query and _trim(query) else "/services/" params = {"q": _trim(query)} if query and _trim(query) else None response = await client.get(f"{AGENTIC_MARKET_API_BASE_URL}{endpoint}", params=params) payload = _parse_json_response( response, provider="x402-agentic-market", operation="Agentic Market search" ) services = payload.get("services") if isinstance(payload, dict) else None if not isinstance(services, list): raise ProviderError("x402-agentic-market", "Unexpected Agentic Market response.") normalized = [ _normalize_agentic_service(service) for service in services if isinstance(service, dict) ] if network: needle = _trim(network).lower() normalized = [ item for item in normalized if needle in {entry.lower() for entry in item.get("networks") or []} ] if scheme: needle = _trim(scheme).lower() normalized = [ item for item in normalized if any(_trim(req.get("scheme")).lower() == needle for req in item.get("accepts") or []) ] if asset: needle = _trim(asset).lower() normalized = [ item for item in normalized if any(_trim(req.get("asset")).lower() == needle for req in item.get("accepts") or []) ] return { "discovery_provider": provider, "query": _trim(query), "count": len(normalized[:limit]), "items": normalized[:limit], } async def get_service_details( *, reference: str, discovery_provider: str = "auto", ) -> dict[str, Any]: ref = _trim(reference) if not ref: raise ProviderError("x402-discovery", "reference is required.") provider = _normalize_discovery_provider(discovery_provider) if provider == "auto": provider = ( "cdp_bazaar" if ref.startswith("http://") or ref.startswith("https://") else "agentic_market" ) if provider == "cdp_bazaar": def _match(items: list[dict[str, Any]]) -> dict[str, Any] | None: exact = next((item for item in items if item.get("resource") == ref), None) if exact is not None: return exact needle = ref.lower() return next( ( item for item in items if needle in _trim(item.get("resource")).lower() or needle in _trim(item.get("description")).lower() ), None, ) # Server-side search covers the full Bazaar catalog; a plain resource # listing only ever returns the first page, so use it as a last resort. queries = [ref] if ref.startswith(("http://", "https://")): netloc = _trim(urlsplit(ref).netloc) if netloc and netloc != ref: queries.append(netloc) exact: dict[str, Any] | None = None for candidate_query in queries: results = await search_services( query=candidate_query, discovery_provider="cdp_bazaar", limit=20 ) exact = _match(results["items"]) if exact is not None: break if exact is None: resources = await search_services(discovery_provider="cdp_bazaar", limit=200) exact = _match(resources["items"]) if exact is None: raise ProviderError("x402-cdp-bazaar", f"No Bazaar resource matched: {ref}") return {"discovery_provider": provider, "service": exact} query = ref if ref.startswith("http://") or ref.startswith("https://"): query = urlsplit(ref).netloc or ref services = await search_services(query=query, discovery_provider="agentic_market", limit=20) needle = ref.lower() exact = next( ( item for item in services["items"] if needle in { _trim(item.get("service_id")).lower(), _trim(item.get("domain")).lower(), _trim(item.get("service_name")).lower(), } ), None, ) if exact is None and services["items"]: exact = services["items"][0] if exact is None: raise ProviderError("x402-agentic-market", f"No Agentic Market service matched: {ref}") return {"discovery_provider": provider, "service": exact} async def preview_request( *, backend: AgentWalletBackend, url: str, method: str = "GET", headers: dict[str, Any] | None = None, query: dict[str, Any] | None = None, json_body: Any | None = None, text_body: str | None = None, ) -> dict[str, Any]: request = _build_request_metadata( url=url, method=method, headers=headers, query=query, json_body=json_body, text_body=text_body, ) client = get_client() response = await _send_request(client=client, request=request) wallet_summary = _wallet_x402_support_summary(backend) address = await backend.get_address() preview: dict[str, Any] = { "mode": "preview", **_build_x402_action_payload( backend=backend, request=request, wallet_summary=wallet_summary, address=address, status_code=response.status_code, selected_payment=None, ), "execute_available": False, "request": { "url": request["url"], "method": request["method"], "request_fingerprint": request["request_fingerprint"], "query": request["query"], "body_hash": request["body_hash"], "content_type": request["content_type"], }, } if response.status_code != 402: preview.update( { "payment_required": False, "response_preview": _response_preview(response), "response_headers": { "content-type": response.headers.get("content-type"), }, } ) return preview payment_preview = _parse_payment_required_response( response, backend=backend, request=request, address=address, wallet_summary=wallet_summary, ) payment_preview["mode"] = "preview" payment_preview["request"] = { "url": request["url"], "method": request["method"], "request_fingerprint": request["request_fingerprint"], "query": request["query"], "body_hash": request["body_hash"], "content_type": request["content_type"], } return payment_preview def _reusable_approved_preview( approved_preview: Any, *, request: dict[str, Any], ) -> dict[str, Any] | None: """Return an approval-time preview usable instead of a fresh unpaid probe. Reuse is only safe when the preview describes exactly this request (fingerprint covers method, final URL, and body hash) and still carries the encoded PAYMENT-REQUIRED challenge plus a selected payment. Anything else falls back to a fresh probe. """ if not isinstance(approved_preview, dict): return None if not approved_preview.get("payment_required"): return None if _trim(approved_preview.get("request_fingerprint")) != request["request_fingerprint"]: return None if not isinstance(approved_preview.get("selected_payment"), dict): return None header = dict(approved_preview.get("response_headers") or {}).get("payment-required") if not isinstance(header, str) or not header.strip(): return None return dict(approved_preview) #: USD amount below which an x402 payment never requires host/session/ #: permission approval, regardless of network (including mainnet) -- the #: same rationale as in-person card payments skipping a signature/PIN below #: a floor limit. Only ever applies when the payment asset is confidently #: identified as USDC (see _looks_like_usdc); any other asset always #: requires approval, since its USD value can't be determined here. DE_MINIMIS_USD_THRESHOLD = 2.0 def de_minimis_usd_amount(preview: dict[str, Any]) -> float | None: """Return *preview*'s payment amount in USD, or None if it isn't confidently USDC. ``x402_amount_display`` is only populated by normalize_payment_requirement when the asset is recognized as USDC (see _looks_like_usdc) -- for any other asset it's None, since its USD value is unknown here. """ amount_display = preview.get("x402_amount_display") if not isinstance(amount_display, str) or not amount_display.strip(): return None try: return float(amount_display) except ValueError: return None def is_de_minimis_payment( preview: dict[str, Any], *, threshold_usd: float | None = None, ) -> bool: """Whether *preview*'s payment is small enough to skip approval entirely. threshold_usd defaults to the current DE_MINIMIS_USD_THRESHOLD, read at call time (not bound at import time) so tests can monkeypatch the module attribute directly. """ usd_amount = de_minimis_usd_amount(preview) if usd_amount is None: return False effective_threshold = DE_MINIMIS_USD_THRESHOLD if threshold_usd is None else threshold_usd return usd_amount < effective_threshold async def resolve_payment_preview( *, backend: AgentWalletBackend, url: str, method: str = "GET", headers: dict[str, Any] | None = None, query: dict[str, Any] | None = None, json_body: Any | None = None, text_body: str | None = None, approved_preview: dict[str, Any] | None = None, ) -> dict[str, Any]: """Return the payment preview for this exact request. Reuses *approved_preview* when it still matches (same fingerprint check ``pay_and_fetch`` applies before deciding whether to skip a fresh probe), otherwise makes a fresh unpaid probe. Callers that need a summary to bind an approval token to -- before paying -- can call this instead of duplicating the reuse-or-probe logic ``pay_and_fetch`` already has. """ request = _build_request_metadata( url=url, method=method, headers=headers, query=query, json_body=json_body, text_body=text_body, ) reused = _reusable_approved_preview(approved_preview, request=request) if reused is not None: return reused return await preview_request( backend=backend, url=url, method=method, headers=headers, query=query, json_body=json_body, text_body=text_body, ) async def pay_and_fetch( *, backend: AgentWalletBackend, url: str, method: str = "GET", headers: dict[str, Any] | None = None, query: dict[str, Any] | None = None, json_body: Any | None = None, text_body: str | None = None, approved_preview: dict[str, Any] | None = None, ) -> dict[str, Any]: request = _build_request_metadata( url=url, method=method, headers=headers, query=query, json_body=json_body, text_body=text_body, ) preview = _reusable_approved_preview(approved_preview, request=request) reused_approved_preview = preview is not None if preview is None: preview = await preview_request( backend=backend, url=url, method=method, headers=headers, query=query, json_body=json_body, text_body=text_body, ) if not preview.get("payment_required"): executed = dict(preview) executed["mode"] = "execute" executed["paid"] = False executed["broadcasted"] = False executed["confirmed"] = False return executed selected_payment = _validate_payment_requirement( preview.get("selected_payment") if isinstance(preview.get("selected_payment"), dict) else None, backend=backend, request_url=str(preview.get("request_url") or url), ) payment_required_header = ( dict(preview.get("response_headers") or {}).get("payment-required") ) if not isinstance(payment_required_header, str) or not payment_required_header.strip(): raise ProviderError("x402-http", "Missing PAYMENT-REQUIRED header in preview state.") _validate_request_execution_policy(request=request, backend=backend) payment_headers = await _create_payment_headers( backend=backend, payment_required_header=payment_required_header, selected_payment=selected_payment, ) client = get_client() response = await _send_request( client=client, request=request, extra_headers=payment_headers, timeout=X402_EXECUTE_TIMEOUT_SECONDS, ) settlement = _extract_settlement_header_safe(response) _log_x402_execute( request=request, selected_payment=selected_payment, response=response, settlement=settlement, ) executed = dict(preview) executed.update( { "mode": "execute", "paid": True, "reused_approved_preview": reused_approved_preview, "broadcasted": bool(settlement and settlement.get("transaction")), "confirmed": bool(settlement and settlement.get("success")), "payment_settlement": settlement, "status_code": response.status_code, "response_preview": _response_preview(response), "response_headers": { "content-type": response.headers.get("content-type"), "payment-response": response.headers.get("PAYMENT-RESPONSE") or response.headers.get("X-PAYMENT-RESPONSE"), }, } ) if response.status_code == 402: message = "The paid x402 retry still returned HTTP 402." if reused_approved_preview: message += ( " The payment was signed from the approved preview quote; the server may have" " re-priced the endpoint since then. Run x402_preview_request again and retry." ) raise ProviderError( "x402-http", message, details={ "request_url": request["url"], "selected_payment": selected_payment, "response_preview": executed["response_preview"], }, ) return executed