From 26bd89adc2d02164e1af7effabd9cbde2b8ef118 Mon Sep 17 00:00:00 2001 From: ashutosh0x Date: Mon, 5 Jan 2026 10:26:05 +0530 Subject: [PATCH] feat: upgrade x402 action provider to v2 Upgrade the x402 action provider to support the v2 protocol, including header-based payment requirements, signatures, and CAIP-2 network identifiers. Maintains backward compatibility with v1 endpoints. --- .../action_providers/x402/schemas.py | 5 +- .../x402/x402_action_provider.py | 50 +++++++++++++------ 2 files changed, 39 insertions(+), 16 deletions(-) diff --git a/python/coinbase-agentkit/coinbase_agentkit/action_providers/x402/schemas.py b/python/coinbase-agentkit/coinbase_agentkit/action_providers/x402/schemas.py index d6dbee554..8018cef42 100644 --- a/python/coinbase-agentkit/coinbase_agentkit/action_providers/x402/schemas.py +++ b/python/coinbase-agentkit/coinbase_agentkit/action_providers/x402/schemas.py @@ -43,7 +43,7 @@ class RetryWithX402Schema(BaseModel): default=None, description="Optional request body for POST/PUT/PATCH requests" ) scheme: str = Field(..., description="The payment scheme to use") - network: str = Field(..., description="The network to use for payment") + network: str = Field(..., description="The network to use (can be CAIP-2 format, e.g., eip155:8453)") max_amount_required: str = Field(..., description="The maximum amount required for payment") resource: str = Field(..., description="The resource URL that requires payment") description: str = Field(default="", description="Description of the payment requirement") @@ -54,6 +54,9 @@ class RetryWithX402Schema(BaseModel): pay_to: str = Field(..., description="Address to send payment to") max_timeout_seconds: int = Field(..., description="Maximum timeout in seconds") asset: str = Field(..., description="Asset contract address to use for payment") + identity: str | None = Field( + default=None, description="Optional wallet-controlled identity for the session" + ) extra: dict[str, Any] | None = Field(default=None, description="Additional payment metadata") class Config: diff --git a/python/coinbase-agentkit/coinbase_agentkit/action_providers/x402/x402_action_provider.py b/python/coinbase-agentkit/coinbase_agentkit/action_providers/x402/x402_action_provider.py index 352f2fec3..b5e372ea6 100644 --- a/python/coinbase-agentkit/coinbase_agentkit/action_providers/x402/x402_action_provider.py +++ b/python/coinbase-agentkit/coinbase_agentkit/action_providers/x402/x402_action_provider.py @@ -1,5 +1,4 @@ -"""x402 action provider.""" - +import base64 import json from typing import Any @@ -14,7 +13,7 @@ from ..action_provider import ActionProvider from .schemas import DirectX402RequestSchema, HttpRequestSchema, RetryWithX402Schema -SUPPORTED_NETWORKS = ["base-mainnet", "base-sepolia"] +SUPPORTED_NETWORKS = ["base-mainnet", "base-sepolia", "eip155:8453", "eip155:84532"] class x402ActionProvider(ActionProvider[EvmWalletProvider]): # noqa: N801 @@ -76,14 +75,30 @@ def make_http_request(self, wallet_provider: EvmWalletProvider, args: dict[str, ) # Parse payment requirements from 402 response - payment_requirements = [ - PaymentRequirements(**accept) for accept in response.json().get("accepts", []) - ] + payment_requirements = [] + + # x402 v2: Check for PAYMENT-REQUIRED header (Base64 encoded JSON) + v2_header = response.headers.get("PAYMENT-REQUIRED") + if v2_header: + try: + decoded_reqs = json.loads(base64.b64decode(v2_header).decode("utf-8")) + # Normalize v2 keys to v1 internal model if needed, but here we expect list of accepts + accepts = decoded_reqs.get("accepts") or [decoded_reqs] + payment_requirements = [PaymentRequirements(**accept) for accept in accepts] + except Exception as e: + print(f"Failed to parse x402 v2 header: {e}") + + # Fallback to v1: Body-based "accepts" + if not payment_requirements: + payment_requirements = [ + PaymentRequirements(**accept) for accept in response.json().get("accepts", []) + ] return json.dumps( { "status": "error_402_payment_required", "acceptablePaymentOptions": [req.dict() for req in payment_requirements], + "protocolVersion": "v2" if v2_header else "v1", "nextSteps": [ "Inform the user that the requested server replied with a 402 Payment Required response.", f"The payment options are: {', '.join(f'{req.asset} {req.max_amount_required} {req.network}' for req in payment_requirements)}", @@ -171,20 +186,25 @@ def payment_selector( session = x402_requests(account, payment_requirements_selector=payment_selector) # Pass the payment data to the session request + # x402-v2: Re-inject identity if provided + headers = args.get("headers") or {} + if args.get("identity"): + headers["x-wallet-identity"] = args["identity"] + response = session.request( url=args["url"], method=args["method"] or "GET", - headers=args.get("headers"), + headers=headers, data=args.get("body"), ) # Extract payment proof if available payment_proof = None - if "x-payment-response" in response.headers: + # Check both x-payment-response (v1) and PAYMENT-RESPONSE (v2) + payment_resp_header = response.headers.get("PAYMENT-RESPONSE") or response.headers.get("x-payment-response") + if payment_resp_header: try: - payment_proof = decode_x_payment_response( - response.headers["x-payment-response"] - ) + payment_proof = decode_x_payment_response(payment_resp_header) except Exception as e: print("Failed to decode payment proof:", str(e)) pass @@ -268,11 +288,11 @@ def make_http_request_with_x402( # Extract payment proof if available payment_proof = None - if "x-payment-response" in response.headers: + # Check both x-payment-response (v1) and PAYMENT-RESPONSE (v2) + payment_resp_header = response.headers.get("PAYMENT-RESPONSE") or response.headers.get("x-payment-response") + if payment_resp_header: try: - payment_proof = decode_x_payment_response( - response.headers["x-payment-response"] - ) + payment_proof = decode_x_payment_response(payment_resp_header) except Exception as e: print("Failed to decode payment proof:", str(e)) pass