refactor(identity/ucp): payment handlers consume *RailSpec

agentscore_commerce/identity/ucp.py

@@ -20,6 +20,16 @@
 from dataclasses import dataclass, field
 from typing import Any, Literal
 
+from agentscore_commerce.payment.networks import networks
+from agentscore_commerce.payment.rail_spec import (
+    RecipientLike,
+    SolanaMppRailSpec,
+    StripeRailSpec,
+    TempoRailSpec,
+    TempoSessionRailSpec,
+    X402BaseRailSpec,
+)
+
 _DEFAULT_VERSION = "2026-04-08"
 
 # Reverse-DNS namespacing per UCP convention. The bare ``agentscore-identity`` form
@@ -469,17 +479,89 @@ async def ucp_profile():
 _SCHEMA_BASE = "https://agentscore.sh/schemas/payment-handlers"
 
 
-def mpp_payment_handler(*, networks: list[dict[str, Any]]) -> dict[str, list[UCPPaymentHandlerBinding]]:
+# CAIP-2 → UCP-namespace network-name mapping. UCP payment_handler bindings publish
+# network strings in the UCP namespace ("base-8453", "solana-mainnet-beta"); RailSpecs
+# carry the CAIP-2 form internally ("eip155:8453", "solana:5eykt4..."). Unknown values
+# pass through verbatim — vendors who pin a non-standard rail can override the spec's
+# network field directly.
+_CAIP2_TO_UCP_NETWORK = {
+    networks.base.mainnet.caip2: "base-8453",
+    networks.base.sepolia.caip2: "base-84532",
+    networks.solana.mainnet.caip2: "solana-mainnet-beta",
+    networks.solana.devnet.caip2: "solana-devnet",
+}
+
+
+def _ucp_network_name(caip2_or_ucp: str) -> str:
+    return _CAIP2_TO_UCP_NETWORK.get(caip2_or_ucp, caip2_or_ucp)
+
+
+def _static_recipient(r: RecipientLike) -> str | None:
+    """Return the recipient as a string when it's already concrete; `None` for factories.
+
+    Per-order factory recipients (e.g. Stripe-multichain mints fresh deposits per
+    PaymentIntent) cannot be advertised in the static UCP profile — the authoritative
+    recipient ships in the 402 body at request time instead.
+    """
+    return r if isinstance(r, str) else None
+
+
+def _tempo_to_network_entry(spec: TempoRailSpec) -> dict[str, Any]:
+    entry: dict[str, Any] = {
+        "network": "tempo-testnet" if spec.testnet else spec.network,
+        "chain_id": spec.chain_id,
+    }
+    static = _static_recipient(spec.recipient)
+    if static is not None:
+        entry["recipient"] = static
+    return entry
+
+
+def _solana_mpp_to_network_entry(spec: SolanaMppRailSpec) -> dict[str, Any]:
+    entry: dict[str, Any] = {"network": _ucp_network_name(spec.network)}
+    static = _static_recipient(spec.recipient)
+    if static is not None:
+        entry["recipient"] = static
+    return entry
+
+
+def _tempo_session_to_network_entry(spec: TempoSessionRailSpec) -> dict[str, Any]:
+    entry: dict[str, Any] = {
+        "network": "tempo-testnet" if spec.testnet else "tempo-mainnet",
+        "escrow_contract": spec.escrow_contract,
+    }
+    static = _static_recipient(spec.recipient)
+    if static is not None:
+        entry["recipient"] = static
+    return entry
+
+
+def _mpp_rail_to_network_entry(spec: TempoRailSpec | SolanaMppRailSpec | TempoSessionRailSpec) -> dict[str, Any]:
+    if isinstance(spec, TempoRailSpec):
+        return _tempo_to_network_entry(spec)
+    if isinstance(spec, SolanaMppRailSpec):
+        return _solana_mpp_to_network_entry(spec)
+    if isinstance(spec, TempoSessionRailSpec):
+        return _tempo_session_to_network_entry(spec)
+    msg = f"mpp_payment_handler: unsupported rail spec type {type(spec).__name__}"
+    raise TypeError(msg)
+
+
+def mpp_payment_handler(
+    *,
+    networks: list[TempoRailSpec | SolanaMppRailSpec | TempoSessionRailSpec],
+) -> dict[str, list[UCPPaymentHandlerBinding]]:
     """Build the `sh.agentscore.payment.mpp` payment handler block for a UCP profile.
 
-    Each network entry: `{"network": <id>, "chain_id"?: <int>, "recipient"?: <addr>, ...}`.
-    Tempo: `tempo-mainnet` / `tempo-testnet`. Solana via `solana/charge`:
-    `mpp-solana-mainnet` / `mpp-solana-devnet`.
+    Pass any mix of `TempoRailSpec`, `SolanaMppRailSpec`, and `TempoSessionRailSpec`.
+    Tempo + Solana SPL both flow through the MPP handler; tempo-session covers the
+    pay-as-you-go channel variant.
 
     Spread into payment_handlers:
         payment_handlers={
             **mpp_payment_handler(networks=[
-                {"network": "tempo-mainnet", "chain_id": 4217},
+                TempoRailSpec(recipient="0xabc..."),
+                SolanaMppRailSpec(recipient="solanaaddr..."),
             ]),
         }
     """
@@ -490,23 +572,34 @@ def mpp_payment_handler(*, networks: list[dict[str, Any]]) -> dict[str, list[UCP
                 version=_HANDLER_VERSION,
                 spec=f"{_SPEC_BASE}/mpp",
                 schema=f"{_SCHEMA_BASE}/mpp.json",
-                config={"networks": networks},
+                config={"networks": [_mpp_rail_to_network_entry(s) for s in networks]},
             )
         ]
     }
 
 
-def x402_payment_handler(*, networks: list[dict[str, Any]]) -> dict[str, list[UCPPaymentHandlerBinding]]:
+def _x402_rail_to_network_entry(spec: X402BaseRailSpec) -> dict[str, Any]:
+    entry: dict[str, Any] = {"network": _ucp_network_name(spec.network)}
+    static = _static_recipient(spec.recipient)
+    if static is not None:
+        entry["recipient"] = static
+    return entry
+
+
+def x402_payment_handler(
+    *,
+    networks: list[X402BaseRailSpec],
+) -> dict[str, list[UCPPaymentHandlerBinding]]:
     """Build the `sh.agentscore.payment.x402` payment handler block for a UCP profile.
 
-    Each network entry: `{"network": <id>, "recipient"?: <addr>, ...}`.
-    EVM: `base-8453`, `base-84532`. Solana: `solana-mainnet-beta`, `solana-devnet`.
-    Stellar: `stellar-pubnet`, `stellar-testnet`.
+    Today only x402 on EVM (Base mainnet / sepolia) ships through this SDK; the
+    `X402BaseRailSpec.network` defaults to `eip155:8453` (CAIP-2) and is converted to
+    `base-8453` for the UCP profile internally.
 
     Spread into payment_handlers:
         payment_handlers={
             **x402_payment_handler(networks=[
-                {"network": "base-8453", "recipient": "0xabc..."},
+                X402BaseRailSpec(recipient="0xabc..."),
             ]),
         }
     """
@@ -517,18 +610,22 @@ def x402_payment_handler(*, networks: list[dict[str, Any]]) -> dict[str, list[UC
                 version=_HANDLER_VERSION,
                 spec=f"{_SPEC_BASE}/x402",
                 schema=f"{_SCHEMA_BASE}/x402.json",
-                config={"networks": networks},
+                config={"networks": [_x402_rail_to_network_entry(s) for s in networks]},
             )
         ]
     }
 
 
-def stripe_spt_payment_handler(*, profile_id: str) -> dict[str, list[UCPPaymentHandlerBinding]]:
+def stripe_spt_payment_handler(*, spec: StripeRailSpec) -> dict[str, list[UCPPaymentHandlerBinding]]:
     """Build the `sh.agentscore.payment.stripe_spt` payment handler block for a UCP profile.
 
+    `spec.profile_id` is the merchant-side network identifier the agent's SPT is scoped
+    to; advertised verbatim in the UCP profile so trust-mode verifiers know which Stripe
+    network they're scoped against.
+
     Spread into payment_handlers:
         payment_handlers={
-            **stripe_spt_payment_handler(profile_id="profile_5xKvNqM9BaH"),
+            **stripe_spt_payment_handler(spec=StripeRailSpec(profile_id="profile_5xKvNqM9BaH")),
         }
     """
     return {
@@ -538,7 +635,7 @@ def stripe_spt_payment_handler(*, profile_id: str) -> dict[str, list[UCPPaymentH
                 version=_HANDLER_VERSION,
                 spec=f"{_SPEC_BASE}/stripe_spt",
                 schema=f"{_SCHEMA_BASE}/stripe_spt.json",
-                config={"rail": "stripe-spt", "profile_id": profile_id},
+                config={"rail": "stripe-spt", "profile_id": spec.profile_id},
             )
         ]
     }

examples/signed_ucp_merchant.py

@@ -46,6 +46,7 @@
     sign_ucp_profile,
     verify_ucp_profile,
 )
+from agentscore_commerce.payment import TempoRailSpec
 
 logger = logging.getLogger("signed_ucp_merchant")
 
@@ -78,9 +79,7 @@ async def well_known_ucp() -> JSONResponse:
         },
         payment_handlers={
             **mpp_payment_handler(
-                networks=[
-                    {"network": "tempo-mainnet", "chain_id": 4217, "recipient": "0xfeedface"},
-                ]
+                networks=[TempoRailSpec(recipient="0xfeedface")],
             ),
         },
         signing_keys=[UCPSigningKey.from_jwk(key.public_jwk)],

tests/test_payment_handlers.py

@@ -0,0 +1,151 @@
+"""Tests for the UCP payment-handler builders consuming *RailSpec."""
+
+from __future__ import annotations
+
+import pytest
+
+from agentscore_commerce.identity.ucp import (
+    mpp_payment_handler,
+    stripe_spt_payment_handler,
+    x402_payment_handler,
+)
+from agentscore_commerce.payment import (
+    SolanaMppRailSpec,
+    StripeRailSpec,
+    TempoRailSpec,
+    TempoSessionRailSpec,
+    X402BaseRailSpec,
+)
+
+
+def test_mpp_tempo_static_recipient() -> None:
+    out = mpp_payment_handler(networks=[TempoRailSpec(recipient="0xfeedface")])
+    binding = out["sh.agentscore.payment.mpp"][0]
+    assert binding.config is not None
+    assert binding.config["networks"] == [
+        {"network": "tempo-mainnet", "chain_id": 4217, "recipient": "0xfeedface"},
+    ]
+
+
+def test_mpp_tempo_testnet_overrides_network_name() -> None:
+    out = mpp_payment_handler(networks=[TempoRailSpec(recipient="0xfeedface", testnet=True)])
+    binding = out["sh.agentscore.payment.mpp"][0]
+    assert binding.config is not None
+    entry = binding.config["networks"][0]
+    assert entry["network"] == "tempo-testnet"
+
+
+def test_mpp_tempo_factory_recipient_omitted_from_static_profile() -> None:
+    """Per-order factory recipients are omitted from the UCP profile (only the 402 body carries them)."""
+
+    async def factory() -> str:
+        return "0xdynamic"
+
+    out = mpp_payment_handler(networks=[TempoRailSpec(recipient=factory)])
+    binding = out["sh.agentscore.payment.mpp"][0]
+    assert binding.config is not None
+    entry = binding.config["networks"][0]
+    assert "recipient" not in entry
+    assert entry["network"] == "tempo-mainnet"
+    assert entry["chain_id"] == 4217
+
+
+def test_mpp_solana_caip2_to_ucp_namespace() -> None:
+    spec = SolanaMppRailSpec(recipient="solanaaddr")
+    out = mpp_payment_handler(networks=[spec])
+    binding = out["sh.agentscore.payment.mpp"][0]
+    assert binding.config is not None
+    entry = binding.config["networks"][0]
+    assert entry["network"] == "solana-mainnet-beta"
+    assert entry["recipient"] == "solanaaddr"
+
+
+def test_mpp_solana_devnet_caip2_to_ucp_namespace() -> None:
+    from agentscore_commerce.payment import networks
+
+    spec = SolanaMppRailSpec(recipient="solanaaddr", network=networks.solana.devnet.caip2)
+    out = mpp_payment_handler(networks=[spec])
+    binding = out["sh.agentscore.payment.mpp"][0]
+    assert binding.config is not None
+    assert binding.config["networks"][0]["network"] == "solana-devnet"
+
+
+def test_mpp_mixed_tempo_solana_session() -> None:
+    """One call can mix Tempo, Solana MPP, and Tempo session rails."""
+    out = mpp_payment_handler(
+        networks=[
+            TempoRailSpec(recipient="0xtempo"),
+            SolanaMppRailSpec(recipient="solanaaddr"),
+            TempoSessionRailSpec(recipient="0xsession", escrow_contract="0xescrow", store=object()),
+        ],
+    )
+    binding = out["sh.agentscore.payment.mpp"][0]
+    assert binding.config is not None
+    entries = binding.config["networks"]
+    assert len(entries) == 3
+    assert entries[2]["escrow_contract"] == "0xescrow"
+
+
+def test_mpp_unknown_spec_type_raises() -> None:
+    with pytest.raises(TypeError, match="unsupported rail spec type"):
+        mpp_payment_handler(networks=["not-a-spec"])  # type: ignore[list-item]
+
+
+def test_x402_base_mainnet_caip2_to_ucp_namespace() -> None:
+    out = x402_payment_handler(networks=[X402BaseRailSpec(recipient="0xbase")])
+    binding = out["sh.agentscore.payment.x402"][0]
+    assert binding.config is not None
+    entry = binding.config["networks"][0]
+    assert entry["network"] == "base-8453"
+    assert entry["recipient"] == "0xbase"
+
+
+def test_x402_base_sepolia_caip2_to_ucp_namespace() -> None:
+    out = x402_payment_handler(
+        networks=[X402BaseRailSpec(recipient="0xbase", network="eip155:84532")],
+    )
+    binding = out["sh.agentscore.payment.x402"][0]
+    assert binding.config is not None
+    assert binding.config["networks"][0]["network"] == "base-84532"
+
+
+def test_x402_factory_recipient_omitted() -> None:
+    async def factory() -> str:
+        return "0xdynamic"
+
+    out = x402_payment_handler(networks=[X402BaseRailSpec(recipient=factory)])
+    binding = out["sh.agentscore.payment.x402"][0]
+    assert binding.config is not None
+    assert "recipient" not in binding.config["networks"][0]
+
+
+def test_x402_unknown_network_passes_through_verbatim() -> None:
+    """A non-standard CAIP-2 (e.g. an unsupported chain) ships through unchanged."""
+    out = x402_payment_handler(
+        networks=[X402BaseRailSpec(recipient="0xbase", network="custom-rail-id")],
+    )
+    binding = out["sh.agentscore.payment.x402"][0]
+    assert binding.config is not None
+    assert binding.config["networks"][0]["network"] == "custom-rail-id"
+
+
+def test_stripe_spt_handler_emits_profile_id() -> None:
+    out = stripe_spt_payment_handler(spec=StripeRailSpec(profile_id="profile_5xKvNqM9BaH"))
+    binding = out["sh.agentscore.payment.stripe_spt"][0]
+    assert binding.config == {"rail": "stripe-spt", "profile_id": "profile_5xKvNqM9BaH"}
+
+
+def test_handler_metadata_versioning() -> None:
+    """All three handlers share the same handler-version constant and spec URL prefix."""
+    mpp = mpp_payment_handler(networks=[TempoRailSpec(recipient="0xt")])
+    x402 = x402_payment_handler(networks=[X402BaseRailSpec(recipient="0xb")])
+    stripe = stripe_spt_payment_handler(spec=StripeRailSpec(profile_id="profile_x"))
+    mpp_binding = mpp["sh.agentscore.payment.mpp"][0]
+    x402_binding = x402["sh.agentscore.payment.x402"][0]
+    stripe_binding = stripe["sh.agentscore.payment.stripe_spt"][0]
+    # All same version + spec/schema base.
+    assert mpp_binding.version == x402_binding.version == stripe_binding.version
+    assert all(
+        b.spec.startswith("https://agentscore.sh/specification/payment-handlers/")
+        for b in [mpp_binding, x402_binding, stripe_binding]
+    )