feat: flatten stripe_multichain helpers to kwargs; delete 4 wrappers

agentscore_commerce/stripe_multichain/__init__.py

@@ -5,23 +5,16 @@
     create_mppx_stripe,
 )
 from agentscore_commerce.stripe_multichain.payment_intent import (
-    CreateMultichainPaymentIntentInput,
     MultichainPaymentIntentResult,
     StripeClientLike,
     create_multichain_payment_intent,
     get_deposit_address,
 )
-from agentscore_commerce.stripe_multichain.pi_cache import (
-    PiCache,
-    PiCacheOptions,
-    create_pi_cache,
-)
+from agentscore_commerce.stripe_multichain.pi_cache import PiCache, create_pi_cache
 from agentscore_commerce.stripe_multichain.simulate_deposit import (
     DEFAULT_BUYER_WALLET,
     STRIPE_TEST_TX_HASH_FAILED,
     STRIPE_TEST_TX_HASH_SUCCESS,
-    SimulateCryptoDepositInput,
-    SimulateDepositIfTestModeInput,
     simulate_crypto_deposit,
     simulate_deposit_if_test_mode,
 )
@@ -31,12 +24,8 @@
     "DEFAULT_PAYMENT_METHOD_TYPES",
     "STRIPE_TEST_TX_HASH_FAILED",
     "STRIPE_TEST_TX_HASH_SUCCESS",
-    "CreateMultichainPaymentIntentInput",
     "MultichainPaymentIntentResult",
     "PiCache",
-    "PiCacheOptions",
-    "SimulateCryptoDepositInput",
-    "SimulateDepositIfTestModeInput",
     "StripeClientLike",
     "create_mppx_stripe",
     "create_multichain_payment_intent",

agentscore_commerce/stripe_multichain/payment_intent.py

@@ -4,7 +4,7 @@
 chains, returning the PI id + deposit addresses per network. Distinct from the Stripe SPT flow.
 """
 
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import Any, Protocol
 
 
@@ -16,44 +16,46 @@ class StripeClientLike(Protocol):
     payment_intents: StripePaymentIntentsAPI
 
 
-@dataclass
-class CreateMultichainPaymentIntentInput:
-    stripe: Any  # StripeClientLike, but kept loose so vendors can pass their actual `stripe.StripeClient`
-    amount: int  # in cents (Stripe convention)
-    currency: str = "usd"
-    networks: list[str] = field(default_factory=lambda: ["tempo", "base", "solana"])
-    metadata: dict[str, str] | None = None
-    idempotency_key: str | None = None
-
-
 @dataclass
 class MultichainPaymentIntentResult:
     payment_intent_id: str
     deposit_addresses: dict[str, str]
 
 
-def create_multichain_payment_intent(input: CreateMultichainPaymentIntentInput) -> MultichainPaymentIntentResult:
+_DEFAULT_NETWORKS: tuple[str, ...] = ("tempo", "base", "solana")
+
+
+def create_multichain_payment_intent(
+    *,
+    stripe: Any,  # StripeClientLike, kept loose so vendors can pass their actual `stripe.StripeClient`
+    amount: int,  # in cents (Stripe convention)
+    currency: str = "usd",
+    networks: list[str] | None = None,
+    metadata: dict[str, str] | None = None,
+    idempotency_key: str | None = None,
+) -> MultichainPaymentIntentResult:
     """Create a Stripe PaymentIntent with multichain crypto deposit_options.
 
     Returns the PI id + per-network deposit addresses. Raises if Stripe doesn't return any addresses.
     """
+    resolved_networks = list(networks) if networks else list(_DEFAULT_NETWORKS)
     params: dict[str, Any] = {
-        "amount": input.amount,
-        "currency": input.currency,
+        "amount": amount,
+        "currency": currency,
         "payment_method_types": ["crypto"],
         "payment_method_data": {"type": "crypto"},
         "payment_method_options": {
-            "crypto": {"mode": "deposit", "deposit_options": {"networks": input.networks}},
+            "crypto": {"mode": "deposit", "deposit_options": {"networks": resolved_networks}},
         },
         "confirm": True,
     }
-    if input.metadata:
-        params["metadata"] = input.metadata
+    if metadata:
+        params["metadata"] = metadata
 
     pi = (
-        input.stripe.payment_intents.create(params, idempotency_key=input.idempotency_key)
-        if input.idempotency_key
-        else input.stripe.payment_intents.create(params)
+        stripe.payment_intents.create(params, idempotency_key=idempotency_key)
+        if idempotency_key
+        else stripe.payment_intents.create(params)
     )
     deposit_addresses: dict[str, str] = {}
     next_action = getattr(pi, "next_action", None) or (pi.get("next_action") if isinstance(pi, dict) else None)

agentscore_commerce/stripe_multichain/pi_cache.py

@@ -52,19 +52,6 @@ class _Entry(Generic[T]):
     expires_at: float
 
 
-@dataclass
-class PiCacheOptions:
-    """Optional configuration for :func:`create_pi_cache`."""
-
-    #: Redis connection URL (e.g. ``rediss://…cache.amazonaws.com:6379``). When omitted,
-    #: the cache falls back to in-process dicts with the same API.
-    redis_url: str | None = None
-    #: TTL for cached entries in seconds. Default 300.
-    ttl_seconds: int = 300
-    #: Prefix for Redis keys. Default ``'payto:'``.
-    key_prefix: str = "payto:"
-
-
 @dataclass
 class PiCache:
     """Stripe PI + deposit-address cache produced by :func:`create_pi_cache`."""
@@ -78,17 +65,25 @@ class PiCache:
     stop: Callable[[], None]
 
 
-def create_pi_cache(opts: PiCacheOptions | None = None) -> PiCache:
+def create_pi_cache(
+    *,
+    redis_url: str | None = None,
+    ttl_seconds: int = 300,
+    key_prefix: str = "payto:",
+) -> PiCache:
     """Construct a Stripe PI + deposit-address cache instance.
 
     Returns a ``PiCache`` with async ``cache_address`` / ``has_address`` (Redis-backed
     when ``redis_url`` is set) and sync helpers for PI-id and network-address lookup.
     A background task evicts expired in-memory entries every 60 seconds; call
     ``stop()`` from server shutdown handlers to cancel it.
+
+    ``redis_url`` — connection URL (e.g. ``rediss://…cache.amazonaws.com:6379``); when
+    omitted, the cache falls back to in-process dicts with the same API.
+    ``ttl_seconds`` — entry TTL (default 300).
+    ``key_prefix`` — Redis key prefix (default ``'payto:'``).
     """
-    options = opts or PiCacheOptions()
-    ttl = options.ttl_seconds
-    key_prefix = options.key_prefix
+    ttl = ttl_seconds
 
     redis_client: _RedisLike | None = None
     addr_mem_cache: dict[str, float] = {}
@@ -97,7 +92,7 @@ def create_pi_cache(opts: PiCacheOptions | None = None) -> PiCache:
 
     async def _get_redis() -> _RedisLike | None:
         nonlocal redis_client
-        if not options.redis_url:
+        if not redis_url:
             return None
         if redis_client is not None:
             return redis_client
@@ -112,7 +107,7 @@ async def _get_redis() -> _RedisLike | None:
                 "[pi-cache] redis_url set but `redis` is not installed. Run `pip install redis` or unset redis_url."
             )
             return None
-        redis_client = redis_asyncio.from_url(options.redis_url)
+        redis_client = redis_asyncio.from_url(redis_url)
         return redis_client
 
     async def cache_address(address: str) -> None:

agentscore_commerce/stripe_multichain/simulate_deposit.py

@@ -2,7 +2,6 @@
 
 import logging
 from collections.abc import Callable
-from dataclasses import dataclass, field
 from typing import Literal
 
 import httpx
@@ -28,57 +27,52 @@
 STRIPE_TEST_TX_HASH_FAILED = "0x000000000000000000000000000000000000000000000000000000testfailed"
 
 
-@dataclass
-class SimulateCryptoDepositInput:
-    payment_intent_id: str
-    network: Literal["tempo", "base", "solana"]
-    stripe_secret_key: str
-    buyer_wallet: str | None = None
-    token_currency: str | None = None
-    transaction_hash: str | None = None
-    stripe_version: str | None = None
-    stripe_api_base: str = "https://api.stripe.com"
-    extra: dict[str, str] = field(default_factory=dict)
-
-
-async def simulate_crypto_deposit(input: SimulateCryptoDepositInput) -> None:
+async def simulate_crypto_deposit(
+    *,
+    payment_intent_id: str,
+    network: Literal["tempo", "base", "solana"],
+    stripe_secret_key: str,
+    buyer_wallet: str | None = None,
+    token_currency: str | None = None,
+    transaction_hash: str | None = None,
+    stripe_version: str | None = None,
+    stripe_api_base: str = "https://api.stripe.com",
+    extra: dict[str, str] | None = None,
+) -> None:
     """Call Stripe's `test_helpers/payment_intents/{id}/simulate_crypto_deposit` endpoint."""
-    url = f"{input.stripe_api_base}/v1/test_helpers/payment_intents/{input.payment_intent_id}/simulate_crypto_deposit"
+    url = f"{stripe_api_base}/v1/test_helpers/payment_intents/{payment_intent_id}/simulate_crypto_deposit"
     params: dict[str, str] = {
-        "network": input.network,
-        "buyer_wallet": input.buyer_wallet or DEFAULT_BUYER_WALLET.get(input.network, ""),
+        "network": network,
+        "buyer_wallet": buyer_wallet or DEFAULT_BUYER_WALLET.get(network, ""),
     }
-    if input.token_currency:
-        params["token_currency"] = input.token_currency
-    if input.transaction_hash:
-        params["transaction_hash"] = input.transaction_hash
-    params.update(input.extra)
+    if token_currency:
+        params["token_currency"] = token_currency
+    if transaction_hash:
+        params["transaction_hash"] = transaction_hash
+    if extra:
+        params.update(extra)
     headers: dict[str, str] = {
-        "Authorization": f"Bearer {input.stripe_secret_key}",
+        "Authorization": f"Bearer {stripe_secret_key}",
         "Content-Type": "application/x-www-form-urlencoded",
     }
-    if input.stripe_version:
-        headers["Stripe-Version"] = input.stripe_version
+    if stripe_version:
+        headers["Stripe-Version"] = stripe_version
     async with httpx.AsyncClient() as client:
         res = await client.post(url, headers=headers, content="&".join(f"{k}={v}" for k, v in params.items()))
         if res.status_code >= 300:
             raise RuntimeError(f"Stripe simulate_crypto_deposit failed: {res.status_code} {res.text}")
 
 
-@dataclass
-class SimulateDepositIfTestModeInput:
-    """Input for :func:`simulate_deposit_if_test_mode`."""
-
-    get_payment_intent_id: Callable[[str], str | None]
-    deposit_address: str
-    network: Literal["tempo", "base", "solana"]
-    stripe_secret_key: str
-    buyer_wallet: str | None = None
-    token_currency: str = "usdc"
-    stripe_version: str | None = None
-
-
-async def simulate_deposit_if_test_mode(input: SimulateDepositIfTestModeInput) -> None:
+async def simulate_deposit_if_test_mode(
+    *,
+    get_payment_intent_id: Callable[[str], str | None],
+    deposit_address: str,
+    network: Literal["tempo", "base", "solana"],
+    stripe_secret_key: str,
+    buyer_wallet: str | None = None,
+    token_currency: str = "usdc",  # noqa: S107 — literal default, not a secret
+    stripe_version: str | None = None,
+) -> None:
     """Higher-level wrapper around :func:`simulate_crypto_deposit` for the testnet/dev path.
 
     Bundles the three steps every Stripe-multichain merchant repeats:
@@ -94,34 +88,32 @@ async def simulate_deposit_if_test_mode(input: SimulateDepositIfTestModeInput) -
 
     Use case is exclusively dev/testnet end-to-end — production servers (sk_live_) no-op.
     """
-    if not input.stripe_secret_key.startswith("sk_test_"):
+    if not stripe_secret_key.startswith("sk_test_"):
         return
-    pi_id = input.get_payment_intent_id(input.deposit_address)
+    pi_id = get_payment_intent_id(deposit_address)
     if not pi_id:
         logger.warning(
             "[stripe] Skipping deposit simulation — no PI cached for deposit address %s… (network=%s). "
             "The PI cache TTL may have expired between 402 emission and settlement.",
-            input.deposit_address[:10],
-            input.network,
+            deposit_address[:10],
+            network,
         )
         return
     try:
         await simulate_crypto_deposit(
-            SimulateCryptoDepositInput(
-                payment_intent_id=pi_id,
-                network=input.network,
-                stripe_secret_key=input.stripe_secret_key,
-                buyer_wallet=input.buyer_wallet,
-                token_currency=input.token_currency,
-                transaction_hash=STRIPE_TEST_TX_HASH_SUCCESS,
-                stripe_version=input.stripe_version,
-            )
+            payment_intent_id=pi_id,
+            network=network,
+            stripe_secret_key=stripe_secret_key,
+            buyer_wallet=buyer_wallet,
+            token_currency=token_currency,
+            transaction_hash=STRIPE_TEST_TX_HASH_SUCCESS,
+            stripe_version=stripe_version,
         )
-        logger.warning("[stripe] ✓ Simulated %s deposit for PI %s", input.network, pi_id)
+        logger.warning("[stripe] ✓ Simulated %s deposit for PI %s", network, pi_id)
     except Exception as err:
         logger.error(
             "[stripe] ✗ Failed to simulate %s deposit for PI %s: %s",
-            input.network,
+            network,
             pi_id,
             err,
         )

examples/multi_rail_merchant.py

@@ -58,8 +58,6 @@
     verify_x402_request,
 )
 from agentscore_commerce.stripe_multichain import (
-    PiCacheOptions,
-    SimulateDepositIfTestModeInput,
     create_pi_cache,
     simulate_deposit_if_test_mode,
 )
@@ -75,7 +73,7 @@
 # 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
 # settles it); falls back to in-process dict for single-instance dev.
-pi_cache = create_pi_cache(PiCacheOptions(redis_url=os.environ.get("REDIS_URL")))
+pi_cache = create_pi_cache(redis_url=os.environ.get("REDIS_URL"))
 
 app = FastAPI()
 _gate = AgentScoreGate(
@@ -168,12 +166,10 @@ async def purchase(request: Request, assess: dict = Depends(get_agentscore_data)
         # Fire Stripe testnet sim; no-ops on live keys. x402 settle only ever
         # lands on base in 1.4+ (Solana moved to MPP `solana/charge`).
         await simulate_deposit_if_test_mode(
-            SimulateDepositIfTestModeInput(
-                get_payment_intent_id=pi_cache.get_payment_intent_id,
-                deposit_address=verified.signed_pay_to,
-                network="base",
-                stripe_secret_key=os.environ["STRIPE_SECRET_KEY"],
-            )
+            get_payment_intent_id=pi_cache.get_payment_intent_id,
+            deposit_address=verified.signed_pay_to,
+            network="base",
+            stripe_secret_key=os.environ["STRIPE_SECRET_KEY"],
         )
 
         headers: dict[str, str] = {}