feat: flatten x402 verify+settle+validate builders to kwargs

agentscore_commerce/payment/__init__.py

@@ -58,7 +58,6 @@
 from agentscore_commerce.payment.x402_settle import (
     ClassifiedX402Error,
     ProcessX402SettleFailure,
-    ProcessX402SettleInput,
     ProcessX402SettleResult,
     ProcessX402SettleSuccess,
     classify_orchestration_error,
@@ -70,9 +69,7 @@
 )
 from agentscore_commerce.payment.x402_validation import (
     X402_SUPPORTED_BASE_NETWORKS,
-    ValidateX402NetworkConfigInput,
     VerifyX402RequestFailure,
-    VerifyX402RequestInput,
     VerifyX402RequestResult,
     VerifyX402RequestSuccess,
     validate_x402_network_config,
@@ -102,7 +99,6 @@
     "PaymentRequiredHeaderInput",
     "PaymentSigner",
     "ProcessX402SettleFailure",
-    "ProcessX402SettleInput",
     "ProcessX402SettleResult",
     "ProcessX402SettleSuccess",
     "RailDefinition",
@@ -111,9 +107,7 @@
     "StripeRail",
     "TempoChargeRail",
     "TempoSessionRail",
-    "ValidateX402NetworkConfigInput",
     "VerifyX402RequestFailure",
-    "VerifyX402RequestInput",
     "VerifyX402RequestResult",
     "VerifyX402RequestSuccess",
     "X402AcceptsBlock",

agentscore_commerce/payment/x402_settle.py

@@ -29,27 +29,6 @@
 from urllib.parse import urlparse
 
 
-@dataclass
-class ProcessX402SettleInput:
-    """Input for :func:`process_x402_settle`."""
-
-    #: The x402 server instance from ``create_x402_server``.
-    x402_server: Any
-    #: The verified x402 payload extracted from the X-Payment header.
-    payload: Any
-    #: Resource configuration the facilitator validates against (network, price, payTo,
-    #: asset, max_timeout_seconds, etc.). Shape is x402-server-specific.
-    resource_config: Any
-    #: Resource metadata exposed to the facilitator.
-    resource_meta: dict[str, str]
-    #: Optional extension to enrich during verify (e.g. Bazaar).
-    extension: Any = None
-    #: Transport context for the extension enrich step. Defaults to ``{"method": "POST",
-    #: "adapter": {"getPath": <pathname>}, "routePattern": <pathname>}`` derived from
-    #: ``resource_meta["url"]``.
-    transport_context: Any = None
-
-
 @dataclass
 class ProcessX402SettleSuccess:
     """Success outcome from :func:`process_x402_settle`."""
@@ -322,14 +301,31 @@ def coerce_payment_payload(payload: Any) -> Any:
         return payload
 
 
-async def process_x402_settle(input: ProcessX402SettleInput) -> ProcessX402SettleResult:
-    """Run the x402 verify→settle flow and return a tagged outcome."""
-    server = input.x402_server
-    resource_config = coerce_resource_config(input.resource_config)
-    payload = coerce_payment_payload(input.payload)
+async def process_x402_settle(
+    *,
+    x402_server: Any,
+    payload: Any,
+    resource_config: Any,
+    resource_meta: dict[str, str],
+    extension: Any = None,
+    transport_context: Any = None,
+) -> ProcessX402SettleResult:
+    """Run the x402 verify→settle flow and return a tagged outcome.
+
+    ``resource_config`` accepts either a ``dict`` (JS-style with ``payTo`` /
+    ``maxTimeoutSeconds`` camelCase keys) or an x402 ``ResourceConfig`` instance —
+    dicts are coerced before the build step.
+
+    Set ``extension`` to fold a Bazaar (or other) extension into the verify step;
+    ``transport_context`` defaults to a POST context derived from
+    ``resource_meta["url"]`` when an extension is supplied.
+    """
+    server = x402_server
+    coerced_config = coerce_resource_config(resource_config)
+    coerced_payload = coerce_payment_payload(payload)
 
     try:
-        built_requirements = server.build_payment_requirements(resource_config)
+        built_requirements = server.build_payment_requirements(coerced_config)
     except Exception as err:
         return ProcessX402SettleFailure(phase="facilitator_error", step="build_requirements", error=err)
     if not built_requirements:
@@ -344,22 +340,22 @@ async def process_x402_settle(input: ProcessX402SettleInput) -> ProcessX402Settl
     # second argument to ``build_payment_requirements`` rather than as a verify-step
     # input, but the fold happens at build time — so we replay the build with the
     # enriched extensions and use those requirements going forward.
-    if input.extension is not None:
-        transport_context = input.transport_context
-        if transport_context is None:
-            path = urlparse(input.resource_meta["url"]).path
-            transport_context = {
+    if extension is not None:
+        resolved_transport_context = transport_context
+        if resolved_transport_context is None:
+            path = urlparse(resource_meta["url"]).path
+            resolved_transport_context = {
                 "method": "POST",
                 "adapter": {"getPath": lambda: path},
                 "routePattern": path,
             }
         try:
-            enriched_ext = server.enrich_extensions(input.extension, transport_context)
+            enriched_ext = server.enrich_extensions(extension, resolved_transport_context)
         except Exception as err:
             return ProcessX402SettleFailure(phase="facilitator_error", step="enrich_extensions", error=err)
         try:
             built_requirements = server.build_payment_requirements(
-                resource_config, list(enriched_ext.keys()) if isinstance(enriched_ext, dict) else None
+                coerced_config, list(enriched_ext.keys()) if isinstance(enriched_ext, dict) else None
             )
             if built_requirements:
                 matched_requirement = built_requirements[0]
@@ -370,7 +366,7 @@ async def process_x402_settle(input: ProcessX402SettleInput) -> ProcessX402Settl
     # — not ``process_payment_request`` (a fictional method that earlier versions of this
     # helper called and only ever worked against test stubs).
     try:
-        verify_result = await server.verify_payment(payload, matched_requirement)
+        verify_result = await server.verify_payment(coerced_payload, matched_requirement)
     except Exception as err:
         return ProcessX402SettleFailure(phase="facilitator_error", step="verify_payment", error=err)
 
@@ -391,7 +387,7 @@ async def process_x402_settle(input: ProcessX402SettleInput) -> ProcessX402Settl
         return ProcessX402SettleFailure(phase="verify_failed", verify_result=verify_result)
 
     try:
-        settle_result = await server.settle_payment(payload, matched_requirement)
+        settle_result = await server.settle_payment(coerced_payload, matched_requirement)
         payment_response_header: str | None = None
         if settle_result is not None:
             payment_response_header = base64.b64encode(settle_result_to_json_bytes(settle_result)).decode()

agentscore_commerce/payment/x402_validation.py

@@ -30,43 +30,23 @@
 X402_SUPPORTED_BASE_NETWORKS: frozenset[str] = frozenset({networks.base.mainnet.caip2, networks.base.sepolia.caip2})
 
 
-@dataclass
-class ValidateX402NetworkConfigInput:
-    """Input for :func:`validate_x402_network_config`."""
-
-    base_network: str
-
-
-def validate_x402_network_config(input: ValidateX402NetworkConfigInput) -> None:
+def validate_x402_network_config(*, base_network: str) -> None:
     """Boot-time guard: raise if the base network isn't supported.
 
     Raises ``ValueError`` with a message that names the unsupported value AND lists the
     valid options — agents tracking down a misconfigured deploy don't need to grep for
     the supported list.
     """
-    if input.base_network not in X402_SUPPORTED_BASE_NETWORKS:
+    if base_network not in X402_SUPPORTED_BASE_NETWORKS:
         raise ValueError(
-            f"X402_BASE_NETWORK={input.base_network} is not supported. "
+            f"X402_BASE_NETWORK={base_network} is not supported. "
             f"Use one of: {', '.join(sorted(X402_SUPPORTED_BASE_NETWORKS))}"
         )
 
 
 _EVM_ADDRESS_RE = re.compile(r"^0x[0-9a-fA-F]{40}$")
 
 
-@dataclass
-class VerifyX402RequestInput:
-    """Input for :func:`verify_x402_request`."""
-
-    #: The incoming request headers (case-insensitive lookup).
-    headers: dict[str, str]
-    #: Async lookup that returns ``True`` when the address was minted by this merchant
-    #: (typically ``pi_cache.has_address``).
-    is_cached_address: Callable[[str], Awaitable[bool]]
-    #: The merchant's accepted Base CAIP-2 network.
-    accepted_network: str
-
-
 @dataclass
 class VerifyX402RequestSuccess:
     """Successful verification — caller passes ``payload`` straight into ``process_x402_settle``."""
@@ -116,7 +96,12 @@ def _regenerate_body(message: str, user_message: str) -> dict[str, Any]:
     }
 
 
-async def verify_x402_request(input: VerifyX402RequestInput) -> VerifyX402RequestResult:
+async def verify_x402_request(
+    *,
+    headers: dict[str, str],
+    is_cached_address: Callable[[str], Awaitable[bool]],
+    accepted_network: str,
+) -> VerifyX402RequestResult:
     """Parse the x402 X-Payment header and validate network + payTo + cache hit.
 
     Returns ``VerifyX402RequestSuccess`` when valid; the caller passes ``payload``
@@ -127,7 +112,7 @@ async def verify_x402_request(input: VerifyX402RequestInput) -> VerifyX402Reques
     Reads the header from ``payment-signature`` first, falling back to ``x-payment``
     (both are in the wild as the binary-friendly transport name evolved).
     """
-    header_value = _header_lookup(input.headers, "payment-signature", "x-payment")
+    header_value = _header_lookup(headers, "payment-signature", "x-payment")
     if not header_value:
         return VerifyX402RequestFailure(
             body=_regenerate_body(
@@ -156,14 +141,14 @@ async def verify_x402_request(input: VerifyX402RequestInput) -> VerifyX402Reques
     signed_network = accepted.get("network")
     signed_pay_to = accepted.get("payTo")
 
-    if not signed_network or signed_network != input.accepted_network:
+    if not signed_network or signed_network != accepted_network:
         if signed_network and signed_network.lower().startswith("solana:"):
             return VerifyX402RequestFailure(
                 body=_regenerate_body(
                     (
                         f"x402 on {signed_network} is not accepted; "
                         f"Solana payments must use the `solana/charge` rail advertised in the 402 challenge. "
-                        f"This server accepts x402 on {input.accepted_network} only."
+                        f"This server accepts x402 on {accepted_network} only."
                     ),
                     (
                         "Solana payments are not accepted over x402 at this merchant. "
@@ -173,10 +158,7 @@ async def verify_x402_request(input: VerifyX402RequestInput) -> VerifyX402Reques
             )
         return VerifyX402RequestFailure(
             body=_regenerate_body(
-                (
-                    f"Unsupported x402 network {signed_network or '<missing>'}; "
-                    f"this server accepts {input.accepted_network}."
-                ),
+                (f"Unsupported x402 network {signed_network or '<missing>'}; this server accepts {accepted_network}."),
                 (
                     "The credential signed for an unsupported network. Pick the accepted "
                     "network from the 402 challenge and re-sign."
@@ -195,7 +177,7 @@ async def verify_x402_request(input: VerifyX402RequestInput) -> VerifyX402Reques
             ),
         )
 
-    if not await input.is_cached_address(signed_pay_to):
+    if not await is_cached_address(signed_pay_to):
         return VerifyX402RequestFailure(
             body=_regenerate_body(
                 "payTo address not found in cache or expired. Request a fresh 402 challenge and retry.",

examples/multi_rail_merchant.py

@@ -51,9 +51,6 @@
 from agentscore_commerce.identity.fastapi import AgentScoreGate, get_agentscore_data
 from agentscore_commerce.payment import (
     USDC,
-    ProcessX402SettleInput,
-    ValidateX402NetworkConfigInput,
-    VerifyX402RequestInput,
     build_x402_accepts_for_402,
     networks,
     process_x402_settle,
@@ -73,7 +70,7 @@
 
 # Boot-time guard: validate the configured x402 networks are in the supported set.
 # Raises on misconfigured deploys before the first request.
-validate_x402_network_config(ValidateX402NetworkConfigInput(base_network=X402_BASE_NETWORK))
+validate_x402_network_config(base_network=X402_BASE_NETWORK)
 
 # Singleton Stripe PI / deposit-address cache. Backed by Redis when REDIS_URL is set
 # (multi-instance deployments need this so a deposit lands on whichever instance
@@ -134,32 +131,28 @@ async def purchase(request: Request, assess: dict = Depends(get_agentscore_data)
     # ──────────────────────────────────────────────────────────────────────────
     if request.headers.get("payment-signature") or request.headers.get("x-payment"):
         verified = await verify_x402_request(
-            VerifyX402RequestInput(
-                headers=dict(request.headers),
-                is_cached_address=pi_cache.has_address,
-                accepted_network=X402_BASE_NETWORK,
-            )
+            headers=dict(request.headers),
+            is_cached_address=pi_cache.has_address,
+            accepted_network=X402_BASE_NETWORK,
         )
         if not verified.ok:
             return JSONResponse(verified.body, status_code=verified.status)
 
         settle = await process_x402_settle(
-            ProcessX402SettleInput(
-                x402_server=x402_server,
-                payload=verified.payload,
-                resource_config={
-                    "scheme": "exact",
-                    "network": verified.signed_network,
-                    "price": f"${total_usd}",
-                    "payTo": verified.signed_pay_to,
-                    "maxTimeoutSeconds": 300,
-                },
-                resource_meta={
-                    "url": str(request.url),
-                    "description": "Agent purchase via x402",
-                    "mimeType": "application/json",
-                },
-            )
+            x402_server=x402_server,
+            payload=verified.payload,
+            resource_config={
+                "scheme": "exact",
+                "network": verified.signed_network,
+                "price": f"${total_usd}",
+                "payTo": verified.signed_pay_to,
+                "maxTimeoutSeconds": 300,
+            },
+            resource_meta={
+                "url": str(request.url),
+                "description": "Agent purchase via x402",
+                "mimeType": "application/json",
+            },
         )
         if not settle.success:
             return JSONResponse(