From bb5977c8e465766705fa44a919aae39986b02cc3 Mon Sep 17 00:00:00 2001 From: Brendan Ryan Date: Mon, 23 Mar 2026 13:02:18 -0700 Subject: [PATCH 1/8] feat: add Stripe payment method (SPT charge intent) Adds Stripe payment method support to pympp, mirroring the TypeScript mppx implementation: - StripeMethod client with create_token callback for SPT creation - ChargeIntent server with Stripe SDK client or raw secret_key paths - Pydantic schemas for ChargeRequest and StripeCredentialPayload - transform_request hook for injecting networkId/paymentMethodTypes - MPP analytics metadata on every PaymentIntent (mpp_version, etc.) - Idempotency key format matching mppx: mppx_{challengeId}_{spt} - 29 tests covering schemas, client, server, and factory --- pyproject.toml | 3 + src/mpp/methods/stripe/__init__.py | 42 +++ src/mpp/methods/stripe/_defaults.py | 3 + src/mpp/methods/stripe/client.py | 243 +++++++++++++ src/mpp/methods/stripe/intents.py | 269 +++++++++++++++ src/mpp/methods/stripe/schemas.py | 35 ++ tests/test_stripe.py | 506 ++++++++++++++++++++++++++++ 7 files changed, 1101 insertions(+) create mode 100644 src/mpp/methods/stripe/__init__.py create mode 100644 src/mpp/methods/stripe/_defaults.py create mode 100644 src/mpp/methods/stripe/client.py create mode 100644 src/mpp/methods/stripe/intents.py create mode 100644 src/mpp/methods/stripe/schemas.py create mode 100644 tests/test_stripe.py diff --git a/pyproject.toml b/pyproject.toml index 275151f..f66e07c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,9 @@ tempo = [ "pytempo>=0.2.1", "pydantic>=2.0", ] +stripe = [ + "pydantic>=2.0", +] server = ["pydantic>=2.0", "python-dotenv>=1.0"] mcp = ["mcp>=1.1.0"] dev = [ diff --git a/src/mpp/methods/stripe/__init__.py b/src/mpp/methods/stripe/__init__.py new file mode 100644 index 0000000..58d5e11 --- /dev/null +++ b/src/mpp/methods/stripe/__init__.py @@ -0,0 +1,42 @@ +"""Stripe payment method for HTTP 402 authentication. + +Uses Stripe's Shared Payment Token (SPT) flow for one-time charges. + +Example: + # Client-side + from mpp.client import get + from mpp.methods.stripe import stripe, ChargeIntent + + async def create_spt(params): + # Proxy to your server endpoint that creates an SPT + ... + return spt_token + + response = await get( + "https://api.example.com/resource", + methods=[stripe( + create_token=create_spt, + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_...")}, + )], + ) + + # Server-side + from mpp.server import Mpp + from mpp.methods.stripe import stripe, ChargeIntent + + server = Mpp.create( + method=stripe( + secret_key="sk_...", + network_id="bn_...", + payment_method_types=["card"], + currency="usd", + decimals=2, + intents={"charge": ChargeIntent(secret_key="sk_...")}, + ), + ) +""" + +from mpp.methods.stripe.client import StripeMethod, stripe +from mpp.methods.stripe.intents import ChargeIntent +from mpp.methods.stripe.schemas import ChargeRequest, StripeCredentialPayload diff --git a/src/mpp/methods/stripe/_defaults.py b/src/mpp/methods/stripe/_defaults.py new file mode 100644 index 0000000..588fe39 --- /dev/null +++ b/src/mpp/methods/stripe/_defaults.py @@ -0,0 +1,3 @@ +"""Default constants for Stripe payment method.""" + +STRIPE_API_BASE = "https://api.stripe.com/v1" diff --git a/src/mpp/methods/stripe/client.py b/src/mpp/methods/stripe/client.py new file mode 100644 index 0000000..222163e --- /dev/null +++ b/src/mpp/methods/stripe/client.py @@ -0,0 +1,243 @@ +"""Stripe payment method for client-side credential creation. + +Implements the charge client method using Stripe's Shared Payment Token (SPT) flow. +""" + +from __future__ import annotations + +import math +import time +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any + +from mpp import Challenge, Credential + +if TYPE_CHECKING: + from mpp.server.intent import Intent + + from mpp import Credential as CredentialType + + +class OnChallengeParameters: + """Parameters passed to the ``create_token`` callback. + + Attributes: + amount: Payment amount in smallest currency unit (e.g. ``"150"`` for $1.50). + challenge: The full payment challenge from the server. + currency: Three-letter ISO currency code (e.g. ``"usd"``). + expires_at: SPT expiration as a Unix timestamp (seconds). + metadata: Optional metadata from the challenge's methodDetails. + network_id: Stripe Business Network profile ID. + payment_method: Stripe payment method ID (e.g. ``"pm_card_visa"``). + """ + + __slots__ = ( + "amount", + "challenge", + "currency", + "expires_at", + "metadata", + "network_id", + "payment_method", + ) + + def __init__( + self, + *, + amount: str, + challenge: Challenge, + currency: str, + expires_at: int, + metadata: dict[str, str] | None, + network_id: str | None, + payment_method: str | None, + ) -> None: + self.amount = amount + self.challenge = challenge + self.currency = currency + self.expires_at = expires_at + self.metadata = metadata + self.network_id = network_id + self.payment_method = payment_method + + +CreateTokenFn = Callable[[OnChallengeParameters], Awaitable[str]] + + +@dataclass +class StripeMethod: + """Stripe payment method implementation. + + Handles client-side credential creation for Stripe SPT payments. + """ + + name: str = "stripe" + create_token: CreateTokenFn | None = None + payment_method: str | None = None + external_id: str | None = None + currency: str | None = None + decimals: int = 2 + recipient: str | None = None + network_id: str | None = None + payment_method_types: list[str] = field(default_factory=lambda: ["card"]) + _intents: dict[str, Intent] = field(default_factory=dict) + + @property + def intents(self) -> dict[str, Intent]: + """Available intents for this method.""" + return self._intents + + def transform_request( + self, request: dict[str, Any], credential: CredentialType | None + ) -> dict[str, Any]: + """Inject Stripe-specific methodDetails into the challenge request. + + Called by ``Mpp`` before challenge creation to add ``networkId`` + and ``paymentMethodTypes`` to the request's ``methodDetails``. + """ + method_details = dict(request.get("methodDetails", {})) + if self.network_id and "networkId" not in method_details: + method_details["networkId"] = self.network_id + if self.payment_method_types and "paymentMethodTypes" not in method_details: + method_details["paymentMethodTypes"] = self.payment_method_types + request = {**request, "methodDetails": method_details} + return request + + async def create_credential(self, challenge: Challenge) -> Credential: + """Create a credential to satisfy the given challenge. + + Calls the user-supplied ``create_token`` callback to obtain an SPT, + then wraps it in a credential for the Authorization header. + + Args: + challenge: The payment challenge from the server. + + Returns: + A credential with the SPT payload. + + Raises: + ValueError: If no ``create_token`` callback or ``payment_method`` is configured. + """ + if self.create_token is None: + raise ValueError("No create_token callback configured") + + request = challenge.request + method_details = request.get("methodDetails", {}) + + payment_method = self.payment_method + if not payment_method: + raise ValueError("payment_method is required (pass to stripe() or via context)") + + amount = str(request.get("amount", "")) + currency = str(request.get("currency", "")) + network_id = method_details.get("networkId") if isinstance(method_details, dict) else None + metadata = method_details.get("metadata") if isinstance(method_details, dict) else None + + if challenge.expires: + expires_at = math.floor( + _parse_iso_timestamp(challenge.expires) + ) + else: + expires_at = math.floor(time.time()) + 3600 + + spt = await self.create_token( + OnChallengeParameters( + amount=amount, + challenge=challenge, + currency=currency, + expires_at=expires_at, + metadata=metadata, + network_id=network_id, + payment_method=payment_method, + ) + ) + + payload: dict[str, Any] = {"spt": spt} + if self.external_id: + payload["externalId"] = self.external_id + + return Credential( + challenge=challenge.to_echo(), + payload=payload, + ) + + +def _parse_iso_timestamp(iso_str: str) -> float: + """Parse an ISO 8601 timestamp to a Unix timestamp (seconds).""" + from datetime import datetime + + dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00")) + return dt.timestamp() + + +# ────────────────────────────────────────────────────────────────── +# Factory +# ────────────────────────────────────────────────────────────────── + + +def stripe( + intents: dict[str, Intent], + create_token: CreateTokenFn | None = None, + payment_method: str | None = None, + external_id: str | None = None, + currency: str | None = None, + decimals: int = 2, + recipient: str | None = None, + network_id: str | None = None, + payment_method_types: list[str] | None = None, + secret_key: str | None = None, +) -> StripeMethod: + """Create a Stripe payment method. + + Args: + intents: Intents to register (e.g. ``{"charge": ChargeIntent(...)}``) + create_token: Callback to create an SPT (client-side). + Receives :class:`OnChallengeParameters` and returns the SPT string. + payment_method: Default Stripe payment method ID (e.g. ``"pm_card_visa"``). + external_id: Optional client-side external reference ID. + currency: Default ISO currency code (e.g. ``"usd"``). + decimals: Decimal places for the currency (default: 2 for USD cents). + recipient: Optional default recipient. + network_id: Stripe Business Network profile ID. Included in + challenge ``methodDetails.networkId``. + payment_method_types: Stripe payment method types (default: ``["card"]``). + Included in challenge ``methodDetails.paymentMethodTypes``. + secret_key: Stripe secret API key. Passed through to intents if needed; + not used directly by the method. + + Returns: + A configured :class:`StripeMethod` instance. + + Example: + from mpp.methods.stripe import stripe, ChargeIntent + + # Server + method = stripe( + secret_key="sk_...", + network_id="bn_...", + payment_method_types=["card"], + currency="usd", + decimals=2, + intents={"charge": ChargeIntent(secret_key="sk_...")}, + ) + + # Client + method = stripe( + create_token=my_spt_proxy, + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_...")}, + ) + """ + method = StripeMethod( + create_token=create_token, + payment_method=payment_method, + external_id=external_id, + currency=currency, + decimals=decimals, + recipient=recipient, + network_id=network_id, + payment_method_types=payment_method_types or ["card"], + ) + method._intents = dict(intents) + return method diff --git a/src/mpp/methods/stripe/intents.py b/src/mpp/methods/stripe/intents.py new file mode 100644 index 0000000..ddadf5e --- /dev/null +++ b/src/mpp/methods/stripe/intents.py @@ -0,0 +1,269 @@ +"""Stripe payment intents (server-side verification). + +Implements the charge intent using Stripe's Shared Payment Token (SPT) flow. +""" + +from __future__ import annotations + +import base64 +from datetime import UTC, datetime +from typing import TYPE_CHECKING, Any + +from mpp import Credential, Receipt +from mpp.errors import ( + PaymentActionRequiredError, + PaymentExpiredError, + VerificationError, +) +from mpp.methods.stripe._defaults import STRIPE_API_BASE +from mpp.methods.stripe.schemas import StripeCredentialPayload + +if TYPE_CHECKING: + from typing import Protocol + + class StripePaymentIntents(Protocol): + def create(self, *args: Any, **kwargs: Any) -> Any: ... + + class StripeClient(Protocol): + """Duck-typed interface for the ``stripe`` Python SDK. + + Matches the subset of the API used for server-side payment verification. + Any object with a ``payment_intents.create()`` method is accepted. + """ + + payment_intents: StripePaymentIntents + + +DEFAULT_TIMEOUT = 30.0 + + +def _build_analytics(credential: Credential) -> dict[str, str]: + """Build MPP analytics metadata for the Stripe PaymentIntent.""" + challenge = credential.challenge + analytics: dict[str, str] = { + "mpp_challenge_id": challenge.id, + "mpp_intent": challenge.intent, + "mpp_is_mpp": "true", + "mpp_server_id": challenge.realm, + "mpp_version": "1", + } + if credential.source: + analytics["mpp_client_id"] = credential.source + return analytics + + +class ChargeIntent: + """Stripe charge intent for one-time payments via SPTs. + + Verifies payment by creating a Stripe PaymentIntent with the + client-supplied Shared Payment Token (SPT). + + Accepts either a ``client`` (a pre-configured Stripe SDK instance) + or a raw ``secret_key``. Using ``client`` is recommended. + + Example: + import stripe as stripe_sdk + from mpp.methods.stripe import ChargeIntent + + client = stripe_sdk.StripeClient("sk_...") + intent = ChargeIntent(client=client) + + # Or with a raw secret key (no Stripe SDK needed): + intent = ChargeIntent(secret_key="sk_...") + """ + + name = "charge" + + def __init__( + self, + client: StripeClient | None = None, + secret_key: str | None = None, + http_client: Any | None = None, + timeout: float = DEFAULT_TIMEOUT, + ) -> None: + """Initialize the charge intent. + + Args: + client: Pre-configured Stripe SDK instance (duck-typed). + Any object with ``payment_intents.create()`` works. + secret_key: Stripe secret API key for raw HTTP verification. + Used only when ``client`` is not provided. + http_client: Optional httpx client for raw HTTP calls. + If provided, the caller is responsible for closing it. + timeout: Request timeout in seconds (default: 30). + + Raises: + ValueError: If neither ``client`` nor ``secret_key`` is provided. + """ + if client is None and secret_key is None: + raise ValueError("Either client or secret_key is required") + self._client = client + self._secret_key = secret_key + self._http_client = http_client + self._owns_client = http_client is None + self._timeout = timeout + + async def __aenter__(self) -> ChargeIntent: + """Enter async context, creating HTTP client if needed.""" + if self._client is None: + await self._get_http_client() + return self + + async def __aexit__(self, *args: Any) -> None: + """Exit async context, closing owned HTTP client.""" + await self.aclose() + + async def aclose(self) -> None: + """Close the HTTP client if we own it.""" + if self._owns_client and self._http_client is not None: + await self._http_client.aclose() + self._http_client = None + + async def _get_http_client(self) -> Any: + """Get or create an httpx async client.""" + if self._http_client is None: + import httpx + + self._http_client = httpx.AsyncClient(timeout=self._timeout) + return self._http_client + + async def verify( + self, + credential: Credential, + request: dict[str, Any], + ) -> Receipt: + """Verify a Stripe charge credential. + + Creates a Stripe PaymentIntent using the SPT from the credential + payload, then checks that payment succeeded. + + Args: + credential: The payment credential from the client. + request: The original payment request parameters. + + Returns: + A receipt indicating success. + + Raises: + VerificationError: If the SPT is missing or PaymentIntent fails. + PaymentExpiredError: If the challenge has expired. + PaymentActionRequiredError: If 3DS or other action is needed. + """ + challenge = credential.challenge + + if challenge.expires: + expires = datetime.fromisoformat(challenge.expires.replace("Z", "+00:00")) + if expires < datetime.now(UTC): + raise PaymentExpiredError(challenge.expires) + + try: + parsed = StripeCredentialPayload.model_validate(credential.payload) + except Exception as err: + raise VerificationError( + "Invalid credential payload: missing or malformed spt" + ) from err + + spt = parsed.spt + credential_external_id = parsed.externalId + + user_metadata = request.get("methodDetails", {}).get("metadata") + resolved_metadata = {**_build_analytics(credential), **(user_metadata or {})} + + if self._client is not None: + pi = await self._create_with_client( + client=self._client, + challenge_id=challenge.id, + request=request, + spt=spt, + metadata=resolved_metadata, + ) + else: + pi = await self._create_with_secret_key( + secret_key=self._secret_key, # type: ignore[arg-type] + challenge_id=challenge.id, + request=request, + spt=spt, + metadata=resolved_metadata, + ) + + if pi["status"] == "requires_action": + raise PaymentActionRequiredError("Stripe PaymentIntent requires action") + if pi["status"] != "succeeded": + raise VerificationError(f"Stripe PaymentIntent status: {pi['status']}") + + return Receipt.success( + reference=pi["id"], + method="stripe", + external_id=credential_external_id, + ) + + async def _create_with_client( + self, + client: StripeClient, + challenge_id: str, + request: dict[str, Any], + spt: str, + metadata: dict[str, str], + ) -> dict[str, str]: + """Create a PaymentIntent using the Stripe SDK client.""" + try: + result = client.payment_intents.create( + params={ + "amount": int(request["amount"]), + "automatic_payment_methods": { + "allow_redirects": "never", + "enabled": True, + }, + "confirm": True, + "currency": request["currency"], + "metadata": metadata, + "shared_payment_granted_token": spt, + }, + options={"idempotency_key": f"mppx_{challenge_id}_{spt}"}, + ) + # Handle both sync and async Stripe clients + if hasattr(result, "__await__"): + result = await result + return {"id": result.id, "status": result.status} + except Exception as err: + raise VerificationError("Stripe PaymentIntent failed") from err + + async def _create_with_secret_key( + self, + secret_key: str, + challenge_id: str, + request: dict[str, Any], + spt: str, + metadata: dict[str, str], + ) -> dict[str, str]: + """Create a PaymentIntent using raw HTTP with a secret key.""" + http_client = await self._get_http_client() + + auth_value = base64.b64encode(f"{secret_key}:".encode()).decode() + + body: dict[str, str] = { + "amount": str(request["amount"]), + "automatic_payment_methods[allow_redirects]": "never", + "automatic_payment_methods[enabled]": "true", + "confirm": "true", + "currency": str(request["currency"]), + "shared_payment_granted_token": spt, + } + for key, value in metadata.items(): + body[f"metadata[{key}]"] = value + + response = await http_client.post( + f"{STRIPE_API_BASE}/payment_intents", + headers={ + "Authorization": f"Basic {auth_value}", + "Content-Type": "application/x-www-form-urlencoded", + "Idempotency-Key": f"mppx_{challenge_id}_{spt}", + }, + data=body, + ) + + if not response.is_success: + raise VerificationError("Stripe PaymentIntent failed") + + result = response.json() + return {"id": result["id"], "status": result["status"]} diff --git a/src/mpp/methods/stripe/schemas.py b/src/mpp/methods/stripe/schemas.py new file mode 100644 index 0000000..4502fcb --- /dev/null +++ b/src/mpp/methods/stripe/schemas.py @@ -0,0 +1,35 @@ +"""Pydantic schemas for Stripe payment requests and credentials.""" + +from __future__ import annotations + +from pydantic import BaseModel, Field + + +class StripeMethodDetails(BaseModel): + """Method-specific details for Stripe charge requests.""" + + metadata: dict[str, str] | None = None + networkId: str + paymentMethodTypes: list[str] = Field(min_length=1) + + +class ChargeRequest(BaseModel): + """Request schema for the Stripe charge intent. + + After the transform in ``stripe()``, ``amount`` is in the smallest + currency unit (e.g. cents for USD) and ``decimals`` is removed. + """ + + amount: str + currency: str + description: str | None = None + externalId: str | None = None + methodDetails: StripeMethodDetails + recipient: str | None = None + + +class StripeCredentialPayload(BaseModel): + """Credential payload for Stripe SPT-based payments.""" + + externalId: str | None = None + spt: str diff --git a/tests/test_stripe.py b/tests/test_stripe.py new file mode 100644 index 0000000..76531bb --- /dev/null +++ b/tests/test_stripe.py @@ -0,0 +1,506 @@ +"""Tests for the Stripe payment method.""" + +from __future__ import annotations + +import math +import time +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from mpp import Challenge, Credential, ChallengeEcho, Receipt +from mpp.errors import ( + PaymentActionRequiredError, + PaymentExpiredError, + VerificationError, +) +from mpp.methods.stripe import ChargeIntent, StripeMethod, stripe +from mpp.methods.stripe.client import OnChallengeParameters +from mpp.methods.stripe.schemas import ChargeRequest, StripeCredentialPayload, StripeMethodDetails + + +# ────────────────────────────────────────────────────────────────── +# Schema tests +# ────────────────────────────────────────────────────────────────── + + +class TestStripeCredentialPayload: + def test_valid_payload(self): + payload = StripeCredentialPayload.model_validate({"spt": "spt_test_123"}) + assert payload.spt == "spt_test_123" + assert payload.externalId is None + + def test_with_external_id(self): + payload = StripeCredentialPayload.model_validate( + {"spt": "spt_test_123", "externalId": "order-42"} + ) + assert payload.spt == "spt_test_123" + assert payload.externalId == "order-42" + + def test_missing_spt(self): + with pytest.raises(Exception): + StripeCredentialPayload.model_validate({"externalId": "order-42"}) + + +class TestChargeRequest: + def test_valid_request(self): + req = ChargeRequest.model_validate({ + "amount": "150", + "currency": "usd", + "methodDetails": { + "networkId": "bn_test", + "paymentMethodTypes": ["card"], + }, + }) + assert req.amount == "150" + assert req.currency == "usd" + assert req.methodDetails.networkId == "bn_test" + assert req.methodDetails.paymentMethodTypes == ["card"] + + def test_missing_method_details(self): + with pytest.raises(Exception): + ChargeRequest.model_validate({ + "amount": "150", + "currency": "usd", + }) + + def test_empty_payment_method_types(self): + with pytest.raises(Exception): + ChargeRequest.model_validate({ + "amount": "150", + "currency": "usd", + "methodDetails": { + "networkId": "bn_test", + "paymentMethodTypes": [], + }, + }) + + +# ────────────────────────────────────────────────────────────────── +# Client tests +# ────────────────────────────────────────────────────────────────── + + +def _make_challenge(**overrides: Any) -> Challenge: + defaults = { + "id": "test-challenge-id", + "method": "stripe", + "intent": "charge", + "request": { + "amount": "150", + "currency": "usd", + "methodDetails": { + "networkId": "bn_test", + "paymentMethodTypes": ["card"], + }, + }, + "realm": "api.example.com", + "request_b64": "eyJ0ZXN0IjoidHJ1ZSJ9", + "expires": (datetime.now(UTC) + timedelta(hours=1)).isoformat(), + } + defaults.update(overrides) + return Challenge(**defaults) + + +class TestStripeMethod: + @pytest.mark.asyncio + async def test_create_credential(self): + async def fake_create_token(params: OnChallengeParameters) -> str: + assert params.amount == "150" + assert params.currency == "usd" + assert params.network_id == "bn_test" + assert params.payment_method == "pm_card_visa" + return "spt_test_abc" + + method = stripe( + create_token=fake_create_token, + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge() + cred = await method.create_credential(challenge) + + assert cred.payload["spt"] == "spt_test_abc" + assert cred.challenge.method == "stripe" + assert cred.challenge.intent == "charge" + + @pytest.mark.asyncio + async def test_create_credential_with_external_id(self): + async def fake_create_token(params: OnChallengeParameters) -> str: + return "spt_test_abc" + + method = stripe( + create_token=fake_create_token, + payment_method="pm_card_visa", + external_id="order-42", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge() + cred = await method.create_credential(challenge) + + assert cred.payload["spt"] == "spt_test_abc" + assert cred.payload["externalId"] == "order-42" + + @pytest.mark.asyncio + async def test_create_credential_no_create_token_raises(self): + method = stripe( + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge() + with pytest.raises(ValueError, match="create_token"): + await method.create_credential(challenge) + + @pytest.mark.asyncio + async def test_create_credential_no_payment_method_raises(self): + async def fake_create_token(params: OnChallengeParameters) -> str: + return "spt_test_abc" + + method = stripe( + create_token=fake_create_token, + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge() + with pytest.raises(ValueError, match="payment_method"): + await method.create_credential(challenge) + + def test_transform_request(self): + method = stripe( + network_id="bn_test", + payment_method_types=["card"], + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + request = {"amount": "150", "currency": "usd"} + result = method.transform_request(request, None) + + assert result["methodDetails"]["networkId"] == "bn_test" + assert result["methodDetails"]["paymentMethodTypes"] == ["card"] + + def test_transform_request_does_not_overwrite(self): + method = stripe( + network_id="bn_default", + payment_method_types=["card"], + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + request = { + "amount": "150", + "currency": "usd", + "methodDetails": {"networkId": "bn_override"}, + } + result = method.transform_request(request, None) + + assert result["methodDetails"]["networkId"] == "bn_override" + assert result["methodDetails"]["paymentMethodTypes"] == ["card"] + + def test_method_name(self): + method = stripe( + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + assert method.name == "stripe" + + def test_intents(self): + intent = ChargeIntent(secret_key="sk_test_123") + method = stripe(intents={"charge": intent}) + assert method.intents["charge"] is intent + + @pytest.mark.asyncio + async def test_expires_at_from_challenge(self): + """Verify expires_at is computed from the challenge's expires field.""" + recorded_params: list[OnChallengeParameters] = [] + + async def fake_create_token(params: OnChallengeParameters) -> str: + recorded_params.append(params) + return "spt_test" + + method = stripe( + create_token=fake_create_token, + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + expires = (datetime.now(UTC) + timedelta(hours=2)).isoformat() + challenge = _make_challenge(expires=expires) + await method.create_credential(challenge) + + expected = math.floor( + datetime.fromisoformat(expires.replace("Z", "+00:00")).timestamp() + ) + assert recorded_params[0].expires_at == expected + + @pytest.mark.asyncio + async def test_expires_at_fallback_when_no_expires(self): + """When challenge has no expires, default to now + 1 hour.""" + recorded_params: list[OnChallengeParameters] = [] + + async def fake_create_token(params: OnChallengeParameters) -> str: + recorded_params.append(params) + return "spt_test" + + method = stripe( + create_token=fake_create_token, + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge(expires=None) + before = math.floor(time.time()) + 3600 + await method.create_credential(challenge) + after = math.floor(time.time()) + 3600 + + assert before <= recorded_params[0].expires_at <= after + + +# ────────────────────────────────────────────────────────────────── +# Server intent tests +# ────────────────────────────────────────────────────────────────── + + +def _make_credential( + spt: str = "spt_test_abc", + external_id: str | None = None, + expires: str | None = None, +) -> Credential: + payload: dict[str, Any] = {"spt": spt} + if external_id: + payload["externalId"] = external_id + if expires is None: + expires = (datetime.now(UTC) + timedelta(hours=1)).isoformat() + return Credential( + challenge=ChallengeEcho( + id="test-challenge-id", + realm="api.example.com", + method="stripe", + intent="charge", + request="eyJ0ZXN0IjoidHJ1ZSJ9", + expires=expires, + ), + payload=payload, + source="stripe:test", + ) + + +SAMPLE_REQUEST: dict[str, Any] = { + "amount": "150", + "currency": "usd", + "methodDetails": { + "networkId": "bn_test", + "paymentMethodTypes": ["card"], + }, +} + + +@dataclass +class FakePaymentIntent: + id: str = "pi_test_123" + status: str = "succeeded" + + +class FakePaymentIntents: + def __init__(self, result: FakePaymentIntent | None = None): + self._result = result or FakePaymentIntent() + + def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: + return self._result + + +class FakeStripeClient: + def __init__(self, result: FakePaymentIntent | None = None): + self.payment_intents = FakePaymentIntents(result) + + +class TestChargeIntent: + @pytest.mark.asyncio + async def test_verify_with_client_success(self): + intent = ChargeIntent(client=FakeStripeClient()) + credential = _make_credential() + receipt = await intent.verify(credential, SAMPLE_REQUEST) + + assert receipt.status == "success" + assert receipt.reference == "pi_test_123" + assert receipt.method == "stripe" + + @pytest.mark.asyncio + async def test_verify_with_external_id(self): + intent = ChargeIntent(client=FakeStripeClient()) + credential = _make_credential(external_id="order-42") + receipt = await intent.verify(credential, SAMPLE_REQUEST) + + assert receipt.external_id == "order-42" + + @pytest.mark.asyncio + async def test_verify_expired_challenge(self): + intent = ChargeIntent(client=FakeStripeClient()) + expired = (datetime.now(UTC) - timedelta(hours=1)).isoformat() + credential = _make_credential(expires=expired) + + with pytest.raises(PaymentExpiredError): + await intent.verify(credential, SAMPLE_REQUEST) + + @pytest.mark.asyncio + async def test_verify_requires_action(self): + pi = FakePaymentIntent(status="requires_action") + intent = ChargeIntent(client=FakeStripeClient(result=pi)) + credential = _make_credential() + + with pytest.raises(PaymentActionRequiredError): + await intent.verify(credential, SAMPLE_REQUEST) + + @pytest.mark.asyncio + async def test_verify_failed_status(self): + pi = FakePaymentIntent(status="requires_payment_method") + intent = ChargeIntent(client=FakeStripeClient(result=pi)) + credential = _make_credential() + + with pytest.raises(VerificationError, match="requires_payment_method"): + await intent.verify(credential, SAMPLE_REQUEST) + + @pytest.mark.asyncio + async def test_verify_missing_spt(self): + intent = ChargeIntent(client=FakeStripeClient()) + credential = Credential( + challenge=ChallengeEcho( + id="test-challenge-id", + realm="api.example.com", + method="stripe", + intent="charge", + request="eyJ0ZXN0IjoidHJ1ZSJ9", + expires=(datetime.now(UTC) + timedelta(hours=1)).isoformat(), + ), + payload={"not_spt": "bad"}, + ) + + with pytest.raises(VerificationError, match="spt"): + await intent.verify(credential, SAMPLE_REQUEST) + + @pytest.mark.asyncio + async def test_verify_client_exception(self): + class FailingIntents: + def create(self, *args: Any, **kwargs: Any) -> Any: + raise RuntimeError("Stripe API error") + + class FailingClient: + payment_intents = FailingIntents() + + intent = ChargeIntent(client=FailingClient()) + credential = _make_credential() + + with pytest.raises(VerificationError, match="PaymentIntent failed"): + await intent.verify(credential, SAMPLE_REQUEST) + + def test_no_client_or_secret_key_raises(self): + with pytest.raises(ValueError, match="Either client or secret_key"): + ChargeIntent() + + @pytest.mark.asyncio + async def test_analytics_metadata(self): + """Verify analytics metadata is passed to PaymentIntent creation.""" + captured_kwargs: list[dict] = [] + + class CapturingIntents: + def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: + captured_kwargs.append(kwargs) + return FakePaymentIntent() + + class CapturingClient: + payment_intents = CapturingIntents() + + intent = ChargeIntent(client=CapturingClient()) + credential = _make_credential() + await intent.verify(credential, SAMPLE_REQUEST) + + params = captured_kwargs[0]["params"] + metadata = params["metadata"] + assert metadata["mpp_version"] == "1" + assert metadata["mpp_is_mpp"] == "true" + assert metadata["mpp_intent"] == "charge" + assert metadata["mpp_challenge_id"] == "test-challenge-id" + assert metadata["mpp_server_id"] == "api.example.com" + assert metadata["mpp_client_id"] == "stripe:test" + + @pytest.mark.asyncio + async def test_idempotency_key(self): + """Verify idempotency key format matches mppx.""" + captured_kwargs: list[dict] = [] + + class CapturingIntents: + def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: + captured_kwargs.append(kwargs) + return FakePaymentIntent() + + class CapturingClient: + payment_intents = CapturingIntents() + + intent = ChargeIntent(client=CapturingClient()) + credential = _make_credential(spt="spt_test_xyz") + await intent.verify(credential, SAMPLE_REQUEST) + + options = captured_kwargs[0]["options"] + assert options["idempotency_key"] == "mppx_test-challenge-id_spt_test_xyz" + + @pytest.mark.asyncio + async def test_user_metadata_overrides_analytics(self): + """User-supplied metadata should override analytics keys.""" + captured_kwargs: list[dict] = [] + + class CapturingIntents: + def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: + captured_kwargs.append(kwargs) + return FakePaymentIntent() + + class CapturingClient: + payment_intents = CapturingIntents() + + intent = ChargeIntent(client=CapturingClient()) + credential = _make_credential() + request_with_metadata = { + **SAMPLE_REQUEST, + "methodDetails": { + **SAMPLE_REQUEST["methodDetails"], + "metadata": {"mpp_version": "custom", "user_key": "user_val"}, + }, + } + await intent.verify(credential, request_with_metadata) + + metadata = captured_kwargs[0]["params"]["metadata"] + assert metadata["mpp_version"] == "custom" + assert metadata["user_key"] == "user_val" + + +# ────────────────────────────────────────────────────────────────── +# Integration: stripe() factory +# ────────────────────────────────────────────────────────────────── + + +class TestStripeFactory: + def test_defaults(self): + method = stripe( + intents={"charge": ChargeIntent(secret_key="sk_test")}, + ) + assert method.name == "stripe" + assert method.decimals == 2 + assert method.payment_method_types == ["card"] + assert method.currency is None + + def test_custom_params(self): + method = stripe( + intents={"charge": ChargeIntent(secret_key="sk_test")}, + currency="eur", + decimals=0, + network_id="bn_custom", + payment_method_types=["card", "sepa_debit"], + recipient="acct_123", + ) + assert method.currency == "eur" + assert method.decimals == 0 + assert method.network_id == "bn_custom" + assert method.payment_method_types == ["card", "sepa_debit"] + assert method.recipient == "acct_123" From 72e7e944acc9e71ed478fc92110af4914bd92a1a Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Mon, 23 Mar 2026 20:03:05 +0000 Subject: [PATCH 2/8] chore: add changelog --- .changelog/quiet-cows-laugh.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changelog/quiet-cows-laugh.md diff --git a/.changelog/quiet-cows-laugh.md b/.changelog/quiet-cows-laugh.md new file mode 100644 index 0000000..6be59af --- /dev/null +++ b/.changelog/quiet-cows-laugh.md @@ -0,0 +1,5 @@ +--- +pympp: minor +--- + +Added Stripe payment method (`mpp.methods.stripe`) supporting the Shared Payment Token (SPT) flow for HTTP 402 authentication. Includes client-side `StripeMethod` and `stripe()` factory, server-side `ChargeIntent` for PaymentIntent verification via Stripe SDK or raw HTTP, Pydantic schemas, and a `stripe` optional dependency group. From 45366e725043f789264447c4eae2b2a65c305c41 Mon Sep 17 00:00:00 2001 From: Brendan Ryan Date: Mon, 23 Mar 2026 14:37:06 -0700 Subject: [PATCH 3/8] fix: address code review findings for Stripe payment method - Fix Stripe SDK client shape: support both client.v1.payment_intents (stripe-python v8+) and client.payment_intents (legacy/custom) - Add metadata.externalId rejection in create_credential (mppx parity) - Require networkId in challenge.methodDetails (mppx parity) - Use VerificationFailedError instead of VerificationError for proper RFC 9457 Problem Details support - Surface Stripe error details in raw HTTP path instead of generic msg - Remove unused secret_key param from stripe() factory - Fix docstring examples: clients don't need secret keys - Convert OnChallengeParameters to frozen dataclass for consistency - Remove fragile hasattr(__await__) sync/async detection - Pass request body as first positional arg (Stripe SDK convention) - Add tests: raw HTTP path, modern v1 client, async lifecycle, networkId guard, metadata.externalId rejection, factory param removal (44 tests, up from 29) --- src/mpp/methods/stripe/__init__.py | 3 +- src/mpp/methods/stripe/client.py | 47 ++--- src/mpp/methods/stripe/intents.py | 56 ++++-- tests/test_stripe.py | 286 +++++++++++++++++++++++++++-- 4 files changed, 330 insertions(+), 62 deletions(-) diff --git a/src/mpp/methods/stripe/__init__.py b/src/mpp/methods/stripe/__init__.py index 58d5e11..11d5b26 100644 --- a/src/mpp/methods/stripe/__init__.py +++ b/src/mpp/methods/stripe/__init__.py @@ -17,7 +17,7 @@ async def create_spt(params): methods=[stripe( create_token=create_spt, payment_method="pm_card_visa", - intents={"charge": ChargeIntent(secret_key="sk_...")}, + intents={}, )], ) @@ -27,7 +27,6 @@ async def create_spt(params): server = Mpp.create( method=stripe( - secret_key="sk_...", network_id="bn_...", payment_method_types=["card"], currency="usd", diff --git a/src/mpp/methods/stripe/client.py b/src/mpp/methods/stripe/client.py index 222163e..3317250 100644 --- a/src/mpp/methods/stripe/client.py +++ b/src/mpp/methods/stripe/client.py @@ -19,6 +19,7 @@ from mpp import Credential as CredentialType +@dataclass(frozen=True) class OnChallengeParameters: """Parameters passed to the ``create_token`` callback. @@ -32,34 +33,13 @@ class OnChallengeParameters: payment_method: Stripe payment method ID (e.g. ``"pm_card_visa"``). """ - __slots__ = ( - "amount", - "challenge", - "currency", - "expires_at", - "metadata", - "network_id", - "payment_method", - ) - - def __init__( - self, - *, - amount: str, - challenge: Challenge, - currency: str, - expires_at: int, - metadata: dict[str, str] | None, - network_id: str | None, - payment_method: str | None, - ) -> None: - self.amount = amount - self.challenge = challenge - self.currency = currency - self.expires_at = expires_at - self.metadata = metadata - self.network_id = network_id - self.payment_method = payment_method + amount: str + challenge: Challenge + currency: str + expires_at: int + metadata: dict[str, str] | None + network_id: str + payment_method: str CreateTokenFn = Callable[[OnChallengeParameters], Awaitable[str]] @@ -132,7 +112,14 @@ async def create_credential(self, challenge: Challenge) -> Credential: amount = str(request.get("amount", "")) currency = str(request.get("currency", "")) network_id = method_details.get("networkId") if isinstance(method_details, dict) else None + if not network_id: + raise ValueError("networkId is required in challenge.methodDetails") metadata = method_details.get("metadata") if isinstance(method_details, dict) else None + if isinstance(metadata, dict) and "externalId" in metadata: + raise ValueError( + "methodDetails.metadata.externalId is reserved; " + "use credential externalId instead" + ) if challenge.expires: expires_at = math.floor( @@ -186,7 +173,6 @@ def stripe( recipient: str | None = None, network_id: str | None = None, payment_method_types: list[str] | None = None, - secret_key: str | None = None, ) -> StripeMethod: """Create a Stripe payment method. @@ -203,8 +189,6 @@ def stripe( challenge ``methodDetails.networkId``. payment_method_types: Stripe payment method types (default: ``["card"]``). Included in challenge ``methodDetails.paymentMethodTypes``. - secret_key: Stripe secret API key. Passed through to intents if needed; - not used directly by the method. Returns: A configured :class:`StripeMethod` instance. @@ -214,7 +198,6 @@ def stripe( # Server method = stripe( - secret_key="sk_...", network_id="bn_...", payment_method_types=["card"], currency="usd", diff --git a/src/mpp/methods/stripe/intents.py b/src/mpp/methods/stripe/intents.py index ddadf5e..b3a14d4 100644 --- a/src/mpp/methods/stripe/intents.py +++ b/src/mpp/methods/stripe/intents.py @@ -13,7 +13,7 @@ from mpp.errors import ( PaymentActionRequiredError, PaymentExpiredError, - VerificationError, + VerificationFailedError, ) from mpp.methods.stripe._defaults import STRIPE_API_BASE from mpp.methods.stripe.schemas import StripeCredentialPayload @@ -27,8 +27,8 @@ def create(self, *args: Any, **kwargs: Any) -> Any: ... class StripeClient(Protocol): """Duck-typed interface for the ``stripe`` Python SDK. - Matches the subset of the API used for server-side payment verification. - Any object with a ``payment_intents.create()`` method is accepted. + Accepts either the modern ``StripeClient`` (``client.v1.payment_intents``) + or legacy/custom clients (``client.payment_intents``). """ payment_intents: StripePaymentIntents @@ -52,6 +52,25 @@ def _build_analytics(credential: Credential) -> dict[str, str]: return analytics +def _resolve_payment_intents(client: Any) -> Any: + """Resolve the payment_intents accessor from a Stripe client. + + Supports both the modern ``StripeClient`` (``client.v1.payment_intents``) + and legacy/custom clients (``client.payment_intents``). + """ + v1 = getattr(client, "v1", None) + if v1 is not None: + pi = getattr(v1, "payment_intents", None) + if pi is not None: + return pi + pi = getattr(client, "payment_intents", None) + if pi is not None: + return pi + raise TypeError( + "Unsupported Stripe client: expected .v1.payment_intents or .payment_intents" + ) + + class ChargeIntent: """Stripe charge intent for one-time payments via SPTs. @@ -85,7 +104,8 @@ def __init__( Args: client: Pre-configured Stripe SDK instance (duck-typed). - Any object with ``payment_intents.create()`` works. + Supports both ``StripeClient`` (v8+, ``client.v1.payment_intents``) + and legacy clients (``client.payment_intents``). secret_key: Stripe secret API key for raw HTTP verification. Used only when ``client`` is not provided. http_client: Optional httpx client for raw HTTP calls. @@ -145,7 +165,7 @@ async def verify( A receipt indicating success. Raises: - VerificationError: If the SPT is missing or PaymentIntent fails. + VerificationFailedError: If the SPT is missing or PaymentIntent fails. PaymentExpiredError: If the challenge has expired. PaymentActionRequiredError: If 3DS or other action is needed. """ @@ -159,7 +179,7 @@ async def verify( try: parsed = StripeCredentialPayload.model_validate(credential.payload) except Exception as err: - raise VerificationError( + raise VerificationFailedError( "Invalid credential payload: missing or malformed spt" ) from err @@ -189,7 +209,7 @@ async def verify( if pi["status"] == "requires_action": raise PaymentActionRequiredError("Stripe PaymentIntent requires action") if pi["status"] != "succeeded": - raise VerificationError(f"Stripe PaymentIntent status: {pi['status']}") + raise VerificationFailedError(f"Stripe PaymentIntent status: {pi['status']}") return Receipt.success( reference=pi["id"], @@ -207,8 +227,9 @@ async def _create_with_client( ) -> dict[str, str]: """Create a PaymentIntent using the Stripe SDK client.""" try: - result = client.payment_intents.create( - params={ + payment_intents = _resolve_payment_intents(client) + result = payment_intents.create( + { "amount": int(request["amount"]), "automatic_payment_methods": { "allow_redirects": "never", @@ -221,12 +242,11 @@ async def _create_with_client( }, options={"idempotency_key": f"mppx_{challenge_id}_{spt}"}, ) - # Handle both sync and async Stripe clients - if hasattr(result, "__await__"): - result = await result return {"id": result.id, "status": result.status} + except (VerificationFailedError, TypeError): + raise except Exception as err: - raise VerificationError("Stripe PaymentIntent failed") from err + raise VerificationFailedError("Stripe PaymentIntent failed") from err async def _create_with_secret_key( self, @@ -263,7 +283,15 @@ async def _create_with_secret_key( ) if not response.is_success: - raise VerificationError("Stripe PaymentIntent failed") + detail = None + try: + err = response.json().get("error", {}) + detail = err.get("message") or err.get("code") + except Exception: + detail = response.text[:200] if response.text else None + raise VerificationFailedError( + detail or f"Stripe PaymentIntent failed (HTTP {response.status_code})" + ) result = response.json() return {"id": result["id"], "status": result["status"]} diff --git a/tests/test_stripe.py b/tests/test_stripe.py index 76531bb..f34ca74 100644 --- a/tests/test_stripe.py +++ b/tests/test_stripe.py @@ -2,23 +2,26 @@ from __future__ import annotations +import base64 import math import time from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, MagicMock +import httpx import pytest from mpp import Challenge, Credential, ChallengeEcho, Receipt from mpp.errors import ( PaymentActionRequiredError, PaymentExpiredError, - VerificationError, + VerificationFailedError, ) from mpp.methods.stripe import ChargeIntent, StripeMethod, stripe from mpp.methods.stripe.client import OnChallengeParameters +from mpp.methods.stripe.intents import _resolve_payment_intents from mpp.methods.stripe.schemas import ChargeRequest, StripeCredentialPayload, StripeMethodDetails @@ -171,6 +174,52 @@ async def fake_create_token(params: OnChallengeParameters) -> str: with pytest.raises(ValueError, match="payment_method"): await method.create_credential(challenge) + @pytest.mark.asyncio + async def test_create_credential_missing_network_id_raises(self): + """networkId is required in challenge.methodDetails (mppx parity).""" + async def fake_create_token(params: OnChallengeParameters) -> str: + return "spt_test_abc" + + method = stripe( + create_token=fake_create_token, + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge(request={ + "amount": "150", + "currency": "usd", + "methodDetails": { + "paymentMethodTypes": ["card"], + }, + }) + with pytest.raises(ValueError, match="networkId is required"): + await method.create_credential(challenge) + + @pytest.mark.asyncio + async def test_create_credential_rejects_metadata_external_id(self): + """metadata.externalId is reserved (mppx parity).""" + async def fake_create_token(params: OnChallengeParameters) -> str: + return "spt_test_abc" + + method = stripe( + create_token=fake_create_token, + payment_method="pm_card_visa", + intents={"charge": ChargeIntent(secret_key="sk_test_123")}, + ) + + challenge = _make_challenge(request={ + "amount": "150", + "currency": "usd", + "methodDetails": { + "networkId": "bn_test", + "paymentMethodTypes": ["card"], + "metadata": {"externalId": "should-fail"}, + }, + }) + with pytest.raises(ValueError, match="externalId is reserved"): + await method.create_credential(challenge) + def test_transform_request(self): method = stripe( network_id="bn_test", @@ -313,10 +362,40 @@ def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: class FakeStripeClient: + """Legacy-style client with client.payment_intents.""" + + def __init__(self, result: FakePaymentIntent | None = None): + self.payment_intents = FakePaymentIntents(result) + + +class FakeV1: def __init__(self, result: FakePaymentIntent | None = None): self.payment_intents = FakePaymentIntents(result) +class FakeModernStripeClient: + """Modern-style client with client.v1.payment_intents (stripe-python v8+).""" + + def __init__(self, result: FakePaymentIntent | None = None): + self.v1 = FakeV1(result) + + +class TestResolvePaymentIntents: + def test_modern_client_v1(self): + client = FakeModernStripeClient() + pi = _resolve_payment_intents(client) + assert pi is client.v1.payment_intents + + def test_legacy_client(self): + client = FakeStripeClient() + pi = _resolve_payment_intents(client) + assert pi is client.payment_intents + + def test_unsupported_client_raises(self): + with pytest.raises(TypeError, match="Unsupported Stripe client"): + _resolve_payment_intents(object()) + + class TestChargeIntent: @pytest.mark.asyncio async def test_verify_with_client_success(self): @@ -328,6 +407,17 @@ async def test_verify_with_client_success(self): assert receipt.reference == "pi_test_123" assert receipt.method == "stripe" + @pytest.mark.asyncio + async def test_verify_with_modern_client_success(self): + """Verify works with client.v1.payment_intents (stripe-python v8+).""" + intent = ChargeIntent(client=FakeModernStripeClient()) + credential = _make_credential() + receipt = await intent.verify(credential, SAMPLE_REQUEST) + + assert receipt.status == "success" + assert receipt.reference == "pi_test_123" + assert receipt.method == "stripe" + @pytest.mark.asyncio async def test_verify_with_external_id(self): intent = ChargeIntent(client=FakeStripeClient()) @@ -360,7 +450,7 @@ async def test_verify_failed_status(self): intent = ChargeIntent(client=FakeStripeClient(result=pi)) credential = _make_credential() - with pytest.raises(VerificationError, match="requires_payment_method"): + with pytest.raises(VerificationFailedError, match="requires_payment_method"): await intent.verify(credential, SAMPLE_REQUEST) @pytest.mark.asyncio @@ -378,7 +468,7 @@ async def test_verify_missing_spt(self): payload={"not_spt": "bad"}, ) - with pytest.raises(VerificationError, match="spt"): + with pytest.raises(VerificationFailedError, match="spt"): await intent.verify(credential, SAMPLE_REQUEST) @pytest.mark.asyncio @@ -393,7 +483,7 @@ class FailingClient: intent = ChargeIntent(client=FailingClient()) credential = _make_credential() - with pytest.raises(VerificationError, match="PaymentIntent failed"): + with pytest.raises(VerificationFailedError, match="PaymentIntent failed"): await intent.verify(credential, SAMPLE_REQUEST) def test_no_client_or_secret_key_raises(self): @@ -403,11 +493,11 @@ def test_no_client_or_secret_key_raises(self): @pytest.mark.asyncio async def test_analytics_metadata(self): """Verify analytics metadata is passed to PaymentIntent creation.""" - captured_kwargs: list[dict] = [] + captured: list[tuple[tuple, dict]] = [] class CapturingIntents: def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: - captured_kwargs.append(kwargs) + captured.append((args, kwargs)) return FakePaymentIntent() class CapturingClient: @@ -417,7 +507,7 @@ class CapturingClient: credential = _make_credential() await intent.verify(credential, SAMPLE_REQUEST) - params = captured_kwargs[0]["params"] + params = captured[0][0][0] metadata = params["metadata"] assert metadata["mpp_version"] == "1" assert metadata["mpp_is_mpp"] == "true" @@ -429,11 +519,11 @@ class CapturingClient: @pytest.mark.asyncio async def test_idempotency_key(self): """Verify idempotency key format matches mppx.""" - captured_kwargs: list[dict] = [] + captured: list[tuple[tuple, dict]] = [] class CapturingIntents: def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: - captured_kwargs.append(kwargs) + captured.append((args, kwargs)) return FakePaymentIntent() class CapturingClient: @@ -443,17 +533,42 @@ class CapturingClient: credential = _make_credential(spt="spt_test_xyz") await intent.verify(credential, SAMPLE_REQUEST) - options = captured_kwargs[0]["options"] + options = captured[0][1]["options"] assert options["idempotency_key"] == "mppx_test-challenge-id_spt_test_xyz" + @pytest.mark.asyncio + async def test_client_request_body_is_first_positional_arg(self): + """Verify request body is passed as first positional arg (Stripe SDK convention).""" + captured: list[tuple[tuple, dict]] = [] + + class CapturingIntents: + def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: + captured.append((args, kwargs)) + return FakePaymentIntent() + + class CapturingClient: + payment_intents = CapturingIntents() + + intent = ChargeIntent(client=CapturingClient()) + credential = _make_credential() + await intent.verify(credential, SAMPLE_REQUEST) + + assert len(captured[0][0]) == 1 + body = captured[0][0][0] + assert body["amount"] == 150 + assert body["currency"] == "usd" + assert body["confirm"] is True + assert body["shared_payment_granted_token"] == "spt_test_abc" + assert "options" in captured[0][1] + @pytest.mark.asyncio async def test_user_metadata_overrides_analytics(self): """User-supplied metadata should override analytics keys.""" - captured_kwargs: list[dict] = [] + captured: list[tuple[tuple, dict]] = [] class CapturingIntents: def create(self, *args: Any, **kwargs: Any) -> FakePaymentIntent: - captured_kwargs.append(kwargs) + captured.append((args, kwargs)) return FakePaymentIntent() class CapturingClient: @@ -470,11 +585,146 @@ class CapturingClient: } await intent.verify(credential, request_with_metadata) - metadata = captured_kwargs[0]["params"]["metadata"] + metadata = captured[0][0][0]["metadata"] assert metadata["mpp_version"] == "custom" assert metadata["user_key"] == "user_val" +# ────────────────────────────────────────────────────────────────── +# Raw HTTP path tests (_create_with_secret_key) +# ────────────────────────────────────────────────────────────────── + + +class TestChargeIntentRawHttp: + @pytest.mark.asyncio + async def test_verify_with_secret_key_success(self): + """Verify the raw HTTP path creates a PaymentIntent successfully.""" + mock_response = httpx.Response( + 200, + json={"id": "pi_http_123", "status": "succeeded"}, + request=httpx.Request("POST", "https://api.stripe.com/v1/payment_intents"), + ) + mock_client = AsyncMock(spec=httpx.AsyncClient) + mock_client.post.return_value = mock_response + + intent = ChargeIntent(secret_key="sk_test_raw", http_client=mock_client) + credential = _make_credential() + receipt = await intent.verify(credential, SAMPLE_REQUEST) + + assert receipt.status == "success" + assert receipt.reference == "pi_http_123" + assert receipt.method == "stripe" + + call_kwargs = mock_client.post.call_args + assert "api.stripe.com/v1/payment_intents" in call_kwargs.args[0] + + headers = call_kwargs.kwargs["headers"] + expected_auth = base64.b64encode(b"sk_test_raw:").decode() + assert headers["Authorization"] == f"Basic {expected_auth}" + assert headers["Idempotency-Key"] == "mppx_test-challenge-id_spt_test_abc" + + data = call_kwargs.kwargs["data"] + assert data["amount"] == "150" + assert data["currency"] == "usd" + assert data["shared_payment_granted_token"] == "spt_test_abc" + + @pytest.mark.asyncio + async def test_verify_with_secret_key_stripe_error(self): + """Verify Stripe error details are surfaced in the exception.""" + mock_response = httpx.Response( + 400, + json={"error": {"message": "Invalid card number", "code": "card_declined"}}, + request=httpx.Request("POST", "https://api.stripe.com/v1/payment_intents"), + ) + mock_client = AsyncMock(spec=httpx.AsyncClient) + mock_client.post.return_value = mock_response + + intent = ChargeIntent(secret_key="sk_test_raw", http_client=mock_client) + credential = _make_credential() + + with pytest.raises(VerificationFailedError, match="Invalid card number"): + await intent.verify(credential, SAMPLE_REQUEST) + + @pytest.mark.asyncio + async def test_verify_with_secret_key_non_json_error(self): + """Verify non-JSON error responses are handled gracefully.""" + mock_response = httpx.Response( + 500, + text="Internal Server Error", + request=httpx.Request("POST", "https://api.stripe.com/v1/payment_intents"), + ) + mock_client = AsyncMock(spec=httpx.AsyncClient) + mock_client.post.return_value = mock_response + + intent = ChargeIntent(secret_key="sk_test_raw", http_client=mock_client) + credential = _make_credential() + + with pytest.raises(VerificationFailedError, match="Internal Server Error"): + await intent.verify(credential, SAMPLE_REQUEST) + + @pytest.mark.asyncio + async def test_verify_with_secret_key_metadata_in_form(self): + """Verify metadata is encoded as form fields.""" + mock_response = httpx.Response( + 200, + json={"id": "pi_meta", "status": "succeeded"}, + request=httpx.Request("POST", "https://api.stripe.com/v1/payment_intents"), + ) + mock_client = AsyncMock(spec=httpx.AsyncClient) + mock_client.post.return_value = mock_response + + intent = ChargeIntent(secret_key="sk_test_raw", http_client=mock_client) + credential = _make_credential() + await intent.verify(credential, SAMPLE_REQUEST) + + data = mock_client.post.call_args.kwargs["data"] + assert data["metadata[mpp_is_mpp]"] == "true" + assert data["metadata[mpp_version]"] == "1" + + +# ────────────────────────────────────────────────────────────────── +# Async context manager tests +# ────────────────────────────────────────────────────────────────── + + +class TestChargeIntentLifecycle: + @pytest.mark.asyncio + async def test_aclose_closes_owned_client(self): + """Owned HTTP client is closed on aclose().""" + intent = ChargeIntent(secret_key="sk_test") + mock_client = AsyncMock(spec=httpx.AsyncClient) + intent._http_client = mock_client + intent._owns_client = True + + await intent.aclose() + + mock_client.aclose.assert_awaited_once() + assert intent._http_client is None + + @pytest.mark.asyncio + async def test_aclose_does_not_close_injected_client(self): + """Injected HTTP client is NOT closed on aclose().""" + mock_client = AsyncMock(spec=httpx.AsyncClient) + intent = ChargeIntent(secret_key="sk_test", http_client=mock_client) + + await intent.aclose() + + mock_client.aclose.assert_not_awaited() + + @pytest.mark.asyncio + async def test_context_manager_closes_owned_client(self): + """__aexit__ closes the owned HTTP client.""" + intent = ChargeIntent(secret_key="sk_test") + mock_client = AsyncMock(spec=httpx.AsyncClient) + intent._http_client = mock_client + intent._owns_client = True + + async with intent: + pass + + mock_client.aclose.assert_awaited_once() + + # ────────────────────────────────────────────────────────────────── # Integration: stripe() factory # ────────────────────────────────────────────────────────────────── @@ -504,3 +754,11 @@ def test_custom_params(self): assert method.network_id == "bn_custom" assert method.payment_method_types == ["card", "sepa_debit"] assert method.recipient == "acct_123" + + def test_no_secret_key_param(self): + """Factory no longer accepts secret_key (removed per review).""" + with pytest.raises(TypeError): + stripe( + intents={"charge": ChargeIntent(secret_key="sk_test")}, + secret_key="sk_test", # type: ignore[call-arg] + ) From 3acfde5494bebe67cb9a91de4de7c7c0ac3a922b Mon Sep 17 00:00:00 2001 From: Brendan Ryan Date: Mon, 23 Mar 2026 14:39:14 -0700 Subject: [PATCH 4/8] fix: resolve lint errors (import sorting, unused imports, blind exceptions) --- src/mpp/methods/stripe/client.py | 10 +-- src/mpp/methods/stripe/intents.py | 4 +- tests/test_stripe.py | 104 ++++++++++++++++-------------- 3 files changed, 61 insertions(+), 57 deletions(-) diff --git a/src/mpp/methods/stripe/client.py b/src/mpp/methods/stripe/client.py index 3317250..c3e22eb 100644 --- a/src/mpp/methods/stripe/client.py +++ b/src/mpp/methods/stripe/client.py @@ -14,9 +14,8 @@ from mpp import Challenge, Credential if TYPE_CHECKING: - from mpp.server.intent import Intent - from mpp import Credential as CredentialType + from mpp.server.intent import Intent @dataclass(frozen=True) @@ -117,14 +116,11 @@ async def create_credential(self, challenge: Challenge) -> Credential: metadata = method_details.get("metadata") if isinstance(method_details, dict) else None if isinstance(metadata, dict) and "externalId" in metadata: raise ValueError( - "methodDetails.metadata.externalId is reserved; " - "use credential externalId instead" + "methodDetails.metadata.externalId is reserved; use credential externalId instead" ) if challenge.expires: - expires_at = math.floor( - _parse_iso_timestamp(challenge.expires) - ) + expires_at = math.floor(_parse_iso_timestamp(challenge.expires)) else: expires_at = math.floor(time.time()) + 3600 diff --git a/src/mpp/methods/stripe/intents.py b/src/mpp/methods/stripe/intents.py index b3a14d4..9c3715e 100644 --- a/src/mpp/methods/stripe/intents.py +++ b/src/mpp/methods/stripe/intents.py @@ -66,9 +66,7 @@ def _resolve_payment_intents(client: Any) -> Any: pi = getattr(client, "payment_intents", None) if pi is not None: return pi - raise TypeError( - "Unsupported Stripe client: expected .v1.payment_intents or .payment_intents" - ) + raise TypeError("Unsupported Stripe client: expected .v1.payment_intents or .payment_intents") class ChargeIntent: diff --git a/tests/test_stripe.py b/tests/test_stripe.py index f34ca74..2f03725 100644 --- a/tests/test_stripe.py +++ b/tests/test_stripe.py @@ -8,22 +8,22 @@ from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock import httpx import pytest +from pydantic import ValidationError -from mpp import Challenge, Credential, ChallengeEcho, Receipt +from mpp import Challenge, ChallengeEcho, Credential from mpp.errors import ( PaymentActionRequiredError, PaymentExpiredError, VerificationFailedError, ) -from mpp.methods.stripe import ChargeIntent, StripeMethod, stripe +from mpp.methods.stripe import ChargeIntent, stripe from mpp.methods.stripe.client import OnChallengeParameters from mpp.methods.stripe.intents import _resolve_payment_intents -from mpp.methods.stripe.schemas import ChargeRequest, StripeCredentialPayload, StripeMethodDetails - +from mpp.methods.stripe.schemas import ChargeRequest, StripeCredentialPayload # ────────────────────────────────────────────────────────────────── # Schema tests @@ -44,42 +44,48 @@ def test_with_external_id(self): assert payload.externalId == "order-42" def test_missing_spt(self): - with pytest.raises(Exception): + with pytest.raises(ValidationError): StripeCredentialPayload.model_validate({"externalId": "order-42"}) class TestChargeRequest: def test_valid_request(self): - req = ChargeRequest.model_validate({ - "amount": "150", - "currency": "usd", - "methodDetails": { - "networkId": "bn_test", - "paymentMethodTypes": ["card"], - }, - }) + req = ChargeRequest.model_validate( + { + "amount": "150", + "currency": "usd", + "methodDetails": { + "networkId": "bn_test", + "paymentMethodTypes": ["card"], + }, + } + ) assert req.amount == "150" assert req.currency == "usd" assert req.methodDetails.networkId == "bn_test" assert req.methodDetails.paymentMethodTypes == ["card"] def test_missing_method_details(self): - with pytest.raises(Exception): - ChargeRequest.model_validate({ - "amount": "150", - "currency": "usd", - }) + with pytest.raises(ValidationError): + ChargeRequest.model_validate( + { + "amount": "150", + "currency": "usd", + } + ) def test_empty_payment_method_types(self): - with pytest.raises(Exception): - ChargeRequest.model_validate({ - "amount": "150", - "currency": "usd", - "methodDetails": { - "networkId": "bn_test", - "paymentMethodTypes": [], - }, - }) + with pytest.raises(ValidationError): + ChargeRequest.model_validate( + { + "amount": "150", + "currency": "usd", + "methodDetails": { + "networkId": "bn_test", + "paymentMethodTypes": [], + }, + } + ) # ────────────────────────────────────────────────────────────────── @@ -177,6 +183,7 @@ async def fake_create_token(params: OnChallengeParameters) -> str: @pytest.mark.asyncio async def test_create_credential_missing_network_id_raises(self): """networkId is required in challenge.methodDetails (mppx parity).""" + async def fake_create_token(params: OnChallengeParameters) -> str: return "spt_test_abc" @@ -186,19 +193,22 @@ async def fake_create_token(params: OnChallengeParameters) -> str: intents={"charge": ChargeIntent(secret_key="sk_test_123")}, ) - challenge = _make_challenge(request={ - "amount": "150", - "currency": "usd", - "methodDetails": { - "paymentMethodTypes": ["card"], - }, - }) + challenge = _make_challenge( + request={ + "amount": "150", + "currency": "usd", + "methodDetails": { + "paymentMethodTypes": ["card"], + }, + } + ) with pytest.raises(ValueError, match="networkId is required"): await method.create_credential(challenge) @pytest.mark.asyncio async def test_create_credential_rejects_metadata_external_id(self): """metadata.externalId is reserved (mppx parity).""" + async def fake_create_token(params: OnChallengeParameters) -> str: return "spt_test_abc" @@ -208,15 +218,17 @@ async def fake_create_token(params: OnChallengeParameters) -> str: intents={"charge": ChargeIntent(secret_key="sk_test_123")}, ) - challenge = _make_challenge(request={ - "amount": "150", - "currency": "usd", - "methodDetails": { - "networkId": "bn_test", - "paymentMethodTypes": ["card"], - "metadata": {"externalId": "should-fail"}, - }, - }) + challenge = _make_challenge( + request={ + "amount": "150", + "currency": "usd", + "methodDetails": { + "networkId": "bn_test", + "paymentMethodTypes": ["card"], + "metadata": {"externalId": "should-fail"}, + }, + } + ) with pytest.raises(ValueError, match="externalId is reserved"): await method.create_credential(challenge) @@ -280,9 +292,7 @@ async def fake_create_token(params: OnChallengeParameters) -> str: challenge = _make_challenge(expires=expires) await method.create_credential(challenge) - expected = math.floor( - datetime.fromisoformat(expires.replace("Z", "+00:00")).timestamp() - ) + expected = math.floor(datetime.fromisoformat(expires.replace("Z", "+00:00")).timestamp()) assert recorded_params[0].expires_at == expected @pytest.mark.asyncio From 058b9596fb638a9e4c3bac56445e0d136b7c0848 Mon Sep 17 00:00:00 2001 From: Brendan Ryan Date: Mon, 23 Mar 2026 14:40:43 -0700 Subject: [PATCH 5/8] fix: use Any for client param to satisfy pyright type checks The StripeClient protocol was too strict for pyright's invariance checking on mutable attributes. Since _resolve_payment_intents() handles duck-type dispatch at runtime, the type annotation can safely be Any. --- src/mpp/methods/stripe/intents.py | 22 +++------------------- 1 file changed, 3 insertions(+), 19 deletions(-) diff --git a/src/mpp/methods/stripe/intents.py b/src/mpp/methods/stripe/intents.py index 9c3715e..8cd7a70 100644 --- a/src/mpp/methods/stripe/intents.py +++ b/src/mpp/methods/stripe/intents.py @@ -7,7 +7,7 @@ import base64 from datetime import UTC, datetime -from typing import TYPE_CHECKING, Any +from typing import Any from mpp import Credential, Receipt from mpp.errors import ( @@ -18,22 +18,6 @@ from mpp.methods.stripe._defaults import STRIPE_API_BASE from mpp.methods.stripe.schemas import StripeCredentialPayload -if TYPE_CHECKING: - from typing import Protocol - - class StripePaymentIntents(Protocol): - def create(self, *args: Any, **kwargs: Any) -> Any: ... - - class StripeClient(Protocol): - """Duck-typed interface for the ``stripe`` Python SDK. - - Accepts either the modern ``StripeClient`` (``client.v1.payment_intents``) - or legacy/custom clients (``client.payment_intents``). - """ - - payment_intents: StripePaymentIntents - - DEFAULT_TIMEOUT = 30.0 @@ -93,7 +77,7 @@ class ChargeIntent: def __init__( self, - client: StripeClient | None = None, + client: Any | None = None, secret_key: str | None = None, http_client: Any | None = None, timeout: float = DEFAULT_TIMEOUT, @@ -217,7 +201,7 @@ async def verify( async def _create_with_client( self, - client: StripeClient, + client: Any, challenge_id: str, request: dict[str, Any], spt: str, From 2e156a89baf6e25971931d88d3dea6f765b1a506 Mon Sep 17 00:00:00 2001 From: Brendan Ryan Date: Mon, 23 Mar 2026 15:00:27 -0700 Subject: [PATCH 6/8] refactor: extract DEFAULT_TIMEOUT to shared mpp._defaults Consolidates the 3 duplicate DEFAULT_TIMEOUT = 30.0 constants from tempo/intents.py, tempo/_rpc.py, and stripe/intents.py into a single shared mpp._defaults module. --- src/mpp/_defaults.py | 4 ++++ src/mpp/methods/stripe/intents.py | 3 +-- src/mpp/methods/tempo/_rpc.py | 2 +- src/mpp/methods/tempo/intents.py | 3 +-- 4 files changed, 7 insertions(+), 5 deletions(-) create mode 100644 src/mpp/_defaults.py diff --git a/src/mpp/_defaults.py b/src/mpp/_defaults.py new file mode 100644 index 0000000..5960e37 --- /dev/null +++ b/src/mpp/_defaults.py @@ -0,0 +1,4 @@ +"""Shared defaults for all payment methods.""" + +DEFAULT_TIMEOUT = 30.0 +"""HTTP request timeout in seconds.""" diff --git a/src/mpp/methods/stripe/intents.py b/src/mpp/methods/stripe/intents.py index 8cd7a70..ceec2c4 100644 --- a/src/mpp/methods/stripe/intents.py +++ b/src/mpp/methods/stripe/intents.py @@ -10,6 +10,7 @@ from typing import Any from mpp import Credential, Receipt +from mpp._defaults import DEFAULT_TIMEOUT from mpp.errors import ( PaymentActionRequiredError, PaymentExpiredError, @@ -18,8 +19,6 @@ from mpp.methods.stripe._defaults import STRIPE_API_BASE from mpp.methods.stripe.schemas import StripeCredentialPayload -DEFAULT_TIMEOUT = 30.0 - def _build_analytics(credential: Credential) -> dict[str, str]: """Build MPP analytics metadata for the Stripe PaymentIntent.""" diff --git a/src/mpp/methods/tempo/_rpc.py b/src/mpp/methods/tempo/_rpc.py index 5b51836..f48119a 100644 --- a/src/mpp/methods/tempo/_rpc.py +++ b/src/mpp/methods/tempo/_rpc.py @@ -5,7 +5,7 @@ import asyncio from typing import Any -DEFAULT_TIMEOUT = 30.0 +from mpp._defaults import DEFAULT_TIMEOUT async def _rpc_call( diff --git a/src/mpp/methods/tempo/intents.py b/src/mpp/methods/tempo/intents.py index 8d582ab..cafd433 100644 --- a/src/mpp/methods/tempo/intents.py +++ b/src/mpp/methods/tempo/intents.py @@ -13,6 +13,7 @@ import attrs from mpp import Credential, Receipt +from mpp._defaults import DEFAULT_TIMEOUT from mpp.errors import VerificationError from mpp.methods.tempo._defaults import DEFAULT_FEE_PAYER_URL, PATH_USD, rpc_url_for_chain from mpp.methods.tempo.schemas import ( @@ -29,8 +30,6 @@ from mpp.methods.tempo.account import TempoAccount -DEFAULT_TIMEOUT = 30.0 - # Receipt polling: 20 * 0.5s = ~10s, enough for testnet block times (~2-4s). MAX_RECEIPT_RETRY_ATTEMPTS = 20 RECEIPT_RETRY_DELAY_SECONDS = 0.5 From a868d28d36e060fc60e4aaea5460f0504bb94d19 Mon Sep 17 00:00:00 2001 From: Brendan Ryan Date: Mon, 23 Mar 2026 15:12:10 -0700 Subject: [PATCH 7/8] feat: add Stripe example (server + headless client) Pay-per-fortune example demonstrating the full SPT flow: - FastAPI server with /api/create-spt proxy and /api/fortune gated endpoint - Headless CLI client using pm_card_visa test card - Mirrors the mppx examples/stripe/ structure --- examples/README.md | 1 + examples/stripe/README.md | 84 +++++++++++++++++++++++++ examples/stripe/client.py | 78 +++++++++++++++++++++++ examples/stripe/pyproject.toml | 14 +++++ examples/stripe/server.py | 109 +++++++++++++++++++++++++++++++++ 5 files changed, 286 insertions(+) create mode 100644 examples/stripe/README.md create mode 100644 examples/stripe/client.py create mode 100644 examples/stripe/pyproject.toml create mode 100644 examples/stripe/server.py diff --git a/examples/README.md b/examples/README.md index efed72b..c59a5fa 100644 --- a/examples/README.md +++ b/examples/README.md @@ -8,6 +8,7 @@ Code examples for using the Machine Payments Protocol (pympp). |---------|-------------| | [fetch/](fetch/) | CLI tool for fetching URLs with automatic payment handling | | [mcp-server/](mcp-server/) | MCP server with payment-protected tools | +| [stripe/](stripe/) | Stripe SPT payment flow (server + headless client) | ## Documentation Examples diff --git a/examples/stripe/README.md b/examples/stripe/README.md new file mode 100644 index 0000000..369f1c1 --- /dev/null +++ b/examples/stripe/README.md @@ -0,0 +1,84 @@ +# Stripe Example + +A pay-per-fortune API using Stripe's Shared Payment Token (SPT) flow. + +## What This Demonstrates + +- Server-side payment protection with `Mpp.create()` and the Stripe method +- SPT proxy endpoint (secret key stays server-side) +- Headless client using a test card (`pm_card_visa`) +- Full 402 → challenge → credential → retry flow + +## Prerequisites + +- Python 3.12+ +- [uv](https://docs.astral.sh/uv/) (recommended) or pip +- A Stripe test-mode secret key (`sk_test_...`) + +## Installation + +```bash +cd examples/stripe +uv sync +``` + +## Running + +**Start the server:** + +```bash +export STRIPE_SECRET_KEY=sk_test_... +uv run python server.py +``` + +The server starts at http://localhost:8000. + +**Run the client** (in another terminal): + +```bash +uv run python client.py +# 🥠 A smooth long journey! Great expectations. +# 📝 Receipt: pi_3Q... +``` + +## Testing Manually + +**Without payment** (returns 402): + +```bash +curl -i http://localhost:8000/api/fortune +# HTTP/1.1 402 Payment Required +# WWW-Authenticate: Payment ... +``` + +## How It Works + +``` +Client Server Stripe + │ │ │ + │ GET /api/fortune │ │ + ├──────────────────────────────> │ │ + │ │ │ + │ 402 + WWW-Authenticate │ │ + │<────────────────────────────── │ │ + │ │ │ + │ POST /api/create-spt │ │ + ├──────────────────────────────> │ Create SPT (test helper) │ + │ ├─────────────────────────────> │ + │ spt_... │ │ + │<────────────────────────────── │<───────────────────────────── │ + │ │ │ + │ GET /api/fortune │ │ + │ Authorization: Payment │ │ + ├──────────────────────────────> │ PaymentIntent (SPT + confirm)│ + │ ├─────────────────────────────> │ + │ │ pi_... succeeded │ + │ 200 + fortune + receipt │<───────────────────────────── │ + │<────────────────────────────── │ │ +``` + +1. Client requests the fortune → server returns 402 with a payment challenge +2. pympp client calls `create_token` → POSTs to `/api/create-spt` → server creates SPT via Stripe +3. Client retries with a credential containing the SPT +4. Server creates a PaymentIntent with `shared_payment_granted_token` and `confirm=True` +5. On success, returns the fortune with a receipt diff --git a/examples/stripe/client.py b/examples/stripe/client.py new file mode 100644 index 0000000..c45b5ac --- /dev/null +++ b/examples/stripe/client.py @@ -0,0 +1,78 @@ +#!/usr/bin/env python3 +"""CLI client that pays for a fortune using Stripe SPTs. + +Uses a test card (pm_card_visa) for headless operation — no browser needed. + +Usage: + export STRIPE_SECRET_KEY=sk_test_... + python client.py [--server http://localhost:8000] +""" + +import argparse +import asyncio +import sys + +import httpx + +from mpp.client import Client +from mpp.methods.stripe import stripe + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + prog="stripe-fortune", + description="Fetch a fortune with automatic Stripe payment", + ) + parser.add_argument( + "--server", + default="http://localhost:8000", + help="Server base URL (default: http://localhost:8000)", + ) + return parser.parse_args() + + +async def run(args: argparse.Namespace) -> int: + server_url = args.server.rstrip("/") + + async def create_token(params): + """Proxy SPT creation through the server.""" + async with httpx.AsyncClient() as http: + response = await http.post( + f"{server_url}/api/create-spt", + json={ + "paymentMethod": params.payment_method, + "amount": params.amount, + "currency": params.currency, + "expiresAt": params.expires_at, + "networkId": params.network_id, + "metadata": params.metadata, + }, + ) + response.raise_for_status() + return response.json()["spt"] + + method = stripe( + create_token=create_token, + payment_method="pm_card_visa", + intents={}, + ) + + async with Client(methods=[method]) as client: + response = await client.get(f"{server_url}/api/fortune") + + if response.status_code >= 400: + print(f"Error {response.status_code}: {response.text}", file=sys.stderr) + return 1 + + data = response.json() + print(f"🥠 {data['fortune']}") + print(f"📝 Receipt: {data['receipt']}") + return 0 + + +def main() -> None: + sys.exit(asyncio.run(run(parse_args()))) + + +if __name__ == "__main__": + main() diff --git a/examples/stripe/pyproject.toml b/examples/stripe/pyproject.toml new file mode 100644 index 0000000..9931ce7 --- /dev/null +++ b/examples/stripe/pyproject.toml @@ -0,0 +1,14 @@ +[project] +name = "stripe-example" +version = "0.1.0" +description = "Stripe payment example using the Machine Payments Protocol" +requires-python = ">=3.12" +dependencies = [ + "pympp[stripe,server]", + "fastapi", + "uvicorn", + "httpx", +] + +[tool.uv.sources] +pympp = { path = "../.." } diff --git a/examples/stripe/server.py b/examples/stripe/server.py new file mode 100644 index 0000000..d8a37d6 --- /dev/null +++ b/examples/stripe/server.py @@ -0,0 +1,109 @@ +"""Stripe payment-protected API server. + +Demonstrates the Machine Payments Protocol with Stripe's Shared Payment +Token (SPT) flow. Two endpoints: + +- POST /api/create-spt — proxy for SPT creation (requires secret key) +- GET /api/fortune — paid endpoint ($1.00 per fortune) +""" + +import base64 +import os +import random + +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse + +from mpp import Challenge +from mpp.methods.stripe import ChargeIntent, stripe +from mpp.server import Mpp + +app = FastAPI(title="Stripe Fortune Server") + +SECRET_KEY = os.environ["STRIPE_SECRET_KEY"] + +server = Mpp.create( + method=stripe( + network_id=os.environ.get("STRIPE_NETWORK_ID", "internal"), + payment_method_types=["card"], + currency="usd", + decimals=2, + recipient=os.environ.get("STRIPE_ACCOUNT", "acct_default"), + intents={"charge": ChargeIntent(secret_key=SECRET_KEY)}, + ), +) + +FORTUNES = [ + "A beautiful, smart, and loving person will come into your life.", + "A dubious friend may be an enemy in camouflage.", + "A faithful friend is a strong defense.", + "A fresh start will put you on your way.", + "A golden egg of opportunity falls into your lap this month.", + "A good time to finish up old tasks.", + "A light heart carries you through all the hard times ahead.", + "A smooth long journey! Great expectations.", +] + + +@app.post("/api/create-spt") +async def create_spt(request: Request): + """Proxy endpoint for SPT creation. + + The client calls this with a payment method ID and challenge details. + We call Stripe's test SPT endpoint using our secret key. + """ + body = await request.json() + + params = { + "payment_method": body["paymentMethod"], + "usage_limits[currency]": body["currency"], + "usage_limits[max_amount]": body["amount"], + "usage_limits[expires_at]": str(body["expiresAt"]), + } + if body.get("networkId"): + params["seller_details[network_id]"] = body["networkId"] + + import httpx + + auth_value = base64.b64encode(f"{SECRET_KEY}:".encode()).decode() + async with httpx.AsyncClient() as client: + response = await client.post( + "https://api.stripe.com/v1/test_helpers/shared_payment/granted_tokens", + headers={ + "Authorization": f"Basic {auth_value}", + "Content-Type": "application/x-www-form-urlencoded", + }, + data=params, + ) + response.raise_for_status() + result = response.json() + + return {"spt": result["id"]} + + +@app.get("/api/fortune") +async def fortune(request: Request): + """Paid endpoint — returns a fortune for $1.00.""" + result = await server.charge( + authorization=request.headers.get("Authorization"), + amount="1", + ) + + if isinstance(result, Challenge): + return JSONResponse( + status_code=402, + content={"error": "Payment required"}, + headers={"WWW-Authenticate": result.to_www_authenticate(server.realm)}, + ) + + credential, receipt = result + return { + "fortune": random.choice(FORTUNES), + "receipt": receipt.reference, + } + + +if __name__ == "__main__": + import uvicorn + + uvicorn.run(app, host="0.0.0.0", port=8000) From 2e713e94648cf4cdd3faa750f6d310a822ebc910 Mon Sep 17 00:00:00 2001 From: Brendan Ryan Date: Mon, 23 Mar 2026 15:23:25 -0700 Subject: [PATCH 8/8] fix: stripe example - remove seller_details param, add error handling seller_details[network_id] is only for Stripe Business Network accounts. Skip it for standard accounts so the SPT endpoint works universally. --- examples/stripe/server.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/examples/stripe/server.py b/examples/stripe/server.py index d8a37d6..3adb16a 100644 --- a/examples/stripe/server.py +++ b/examples/stripe/server.py @@ -60,8 +60,6 @@ async def create_spt(request: Request): "usage_limits[max_amount]": body["amount"], "usage_limits[expires_at]": str(body["expiresAt"]), } - if body.get("networkId"): - params["seller_details[network_id]"] = body["networkId"] import httpx @@ -75,7 +73,8 @@ async def create_spt(request: Request): }, data=params, ) - response.raise_for_status() + if not response.is_success: + return JSONResponse(status_code=502, content=response.json()) result = response.json() return {"spt": result["id"]}