feat: implement x402 buyer payment workflows

README.md

@@ -452,6 +452,125 @@ wire dictionaries from upstream models. Upstream fills omitted requirement
 are preserved. These typed models do not preserve arbitrary unknown top-level
 fields. Parsing a model or creating an identifier does not verify or settle a payment.
 
+## x402 Buyer
+
+Install `inflowpay[x402]` to pay with an InFlow account. `Buyer.create()` loads
+the account's supported payment methods before returning. Supply your buyer API
+key or access-token provider through `ClientOptions`; those credentials are sent
+to InFlow, not to the merchant.
+
+```python
+import asyncio
+import os
+
+import httpx
+from x402.http.clients.httpx import x402AsyncTransport
+
+from inflowpay import ClientOptions
+from inflowpay.x402.buyer import Buyer
+
+
+async def main():
+    async with (
+        await Buyer.create(
+            ClientOptions(api_key=os.environ["INFLOW_API_KEY"], environment="sandbox")
+        ) as buyer,
+        httpx.AsyncClient(transport=x402AsyncTransport(buyer), follow_redirects=False) as http,
+    ):
+        response = await http.get(os.environ["PAID_RESOURCE_URL"])
+        response.raise_for_status()
+        print(response.text)
+
+
+asyncio.run(main())
+```
+
+The upstream transport reads the merchant's payment requirements, asks the Buyer
+for a payment payload, and retries the resource request with that payload. An
+InFlow-managed payment can wait for the account owner to approve it. The default
+approval wait is 15 minutes, polling every 5 seconds; configure `pending_timeout`
+and `poll_interval` in seconds on `Buyer.create()`.
+
+### Show an approval before waiting
+
+Use `await buyer.prepare(requirement, resource)` when your application needs the
+`approval_id` and `transaction_id` before waiting. Pass a selected upstream
+`PaymentRequirements` and `ResourceInfo`. Then call `await payment.await_payload()`
+to obtain `encoded_payload`, `payment_payload`, and `transaction_id`, or
+`await payment.cancel()` to request approval cancellation.
+
+Repeated waits share the completed payload and run completion hooks once. A
+timeout before receiving a payload allows another wait on the same handle without
+creating a second approval. Cancelling an individual waiting task does not cancel
+the server approval; neither does closing the Buyer. The application owns that
+decision in the two-phase flow. The automatic `create_payment_payload()` flow
+requests cancellation if it fails before receiving a signed payload.
+
+### Payment selection and spending controls
+
+`inflowpay.x402.buyer.Buyer` supports two signing paths. It requests an
+InFlow-managed payment when InFlow supports the offered payment requirement.
+Otherwise, it passes the request to an external-wallet scheme registered with
+the upstream x402 client.
+
+Use `register_policy()` to filter payment requirements for either path. For
+InFlow-managed payments, the account's server-side policies and approval process
+also apply. The upstream `set_spend_controls()` settings apply only to
+external-wallet payments; they do not cap an InFlow-managed payment. This is the
+same separation used by the InFlow Node SDK. If your application needs a local
+amount limit for managed payments, enforce it in a registered payment policy.
+
+The explicit `prepare(requirement, resource)` method uses the requirement you
+provide, rather than selecting among offers or running selection policies. Apply
+your application's selection policy before calling it. InFlow's server-side
+policies and approvals still apply.
+
+### External wallets
+
+Install `inflowpay[evm]` or `inflowpay[svm]` and register an upstream signing scheme
+with `buyer.register(network, scheme)` to allow external-wallet payments. The
+Buyer first selects a supported InFlow payment; otherwise it delegates to the
+registered upstream schemes. `prefer` controls the managed scheme order and
+defaults to `("balance", "exact")`. Managed signing does not accept Permit2
+requirements; those use an external wallet.
+
+InFlow-managed payment requests use asynchronous network calls. External-wallet
+payments run through the upstream Python x402 signing implementation. Its Solana
+signer performs synchronous network requests for mint metadata and, when needed,
+a recent blockhash. Those requests block other tasks on the same event loop even
+when the caller awaits `create_payment_payload()`. The upstream TypeScript Solana
+signer awaits these network requests instead.
+
+The Python Buyer preserves upstream signing behavior; it does not move wallet
+signers into background threads. Applications using external wallets should
+account for this blocking work when sharing an event loop with other requests.
+See [upstream issue #3649](https://github.com/x402-foundation/x402/issues/3649)
+for the reproduction and comparison with the TypeScript implementation.
+
+### EIP-7702 sponsorship
+
+`inflowpay.x402.eip7702.SponsorshipExtension` supports external EVM wallets when
+the merchant advertises `inflowEip7702GasSponsoring` for an exact Permit2 payment.
+Install the `evm` extra. Construct the extension with anonymous `ClientOptions`,
+a `SponsorshipSigner`, and an asynchronous consent callback, then register it with
+`buyer.register_extension(extension)`. Keep the extension's asynchronous context
+manager open while creating payments; it owns a separate HTTP client.
+
+The signer provides its address and three asynchronous methods: `allowance()`
+reads the token allowance; `sign_message()` signs the supplied operation hash
+using Ethereum's personal-message prefix; and `sign_authorization()` signs the
+supplied EIP-7702 authorization. Both signing methods return 65-byte signatures
+with recovery ID 27 or 28. The signer must belong to the wallet registered with
+the upstream payment scheme.
+
+Sponsorship is skipped when the existing Permit2 allowance covers the payment.
+Otherwise, the extension requests preparation from InFlow and verifies the
+returned contracts, payment calls, operation hash, and expiry before signing.
+If account delegation is required, the consent callback must return `True` only
+after the owner agrees: delegation persists even if the payment fails. The
+extension returns signed data; it does not broadcast a transaction. Availability
+depends on the InFlow environment's sponsorship endpoint and supported networks.
+
 ## x402 facilitator capabilities
 
 Facilitator capabilities describe the payment schemes, networks, and extensions

scripts/verify_distribution.py

@@ -24,6 +24,7 @@
 from x402.http.middleware.fastapi import payment_middleware
 from x402.mechanisms.evm.exact import ExactEvmClientScheme
 from x402.mechanisms.svm.exact import ExactSvmClientScheme
+from inflowpay.x402.eip7702 import SponsorshipExtension, SponsorshipSigner
 import x402.mcp
 """
 
@@ -69,6 +70,15 @@ async def check_seller():
 assert PaymentRequirements is UpstreamRequirements
 entry = payment_identifier_entry(declare_payment_identifier(), generate_payment_id())
 assert entry is not None and entry['info']['required'] is False
+from inflowpay.x402.buyer import Buyer
+from inflowpay import ClientOptions
+import asyncio
+import httpx
+async def check_buyer():
+    transport = httpx.MockTransport(lambda request: httpx.Response(200, json={'kinds': []}))
+    async with await Buyer.create(ClientOptions(transport=transport)) as buyer:
+        assert (await buyer.get_supported()).kinds == []
+asyncio.run(check_buyer())
 for name in ('mpp', 'mcp', 'web3', 'solana', 'fastapi', 'rfc8785'):
     assert util.find_spec(name) is None, name
 """

src/inflowpay/x402/_payment.py

@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+import asyncio
+import math
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from typing import cast
+from urllib.parse import quote
+
+from x402.schemas import PaymentPayload
+
+from .._runtime import Client
+from ..errors import InflowApiError
+
+
+class X402PaymentError(Exception):
+    def __init__(self, code: str, message: str, *, status: str | None = None) -> None:
+        self.code = code
+        self.status = status
+        super().__init__(message)
+
+
+def response_object(value: object) -> dict[str, object]:
+    if not isinstance(value, dict):
+        raise X402PaymentError("invalid-response", "Expected an x402 response object")
+    return cast(dict[str, object], value)
+
+
+def response_string(value: object) -> str:
+    if not isinstance(value, str) or not value:
+        raise X402PaymentError("invalid-response", "Missing x402 response field")
+    return value
+
+
+def validate_wait(poll_interval: float, timeout: float) -> None:
+    if any(not math.isfinite(value) or value < 0 for value in (poll_interval, timeout)):
+        raise ValueError("Polling settings must be finite and nonnegative")
+
+
+@dataclass(frozen=True)
+class EncodedPayment:
+    encoded_payload: str
+    payment_payload: PaymentPayload
+    transaction_id: str
+
+
+def _observe_completion(task: asyncio.Task[EncodedPayment]) -> None:
+    # Hooks may finish after the last waiter cancels. Keep their error available for
+    # a later wait without emitting an unobserved-task exception in the meantime.
+    if not task.cancelled():
+        task.exception()
+
+
+class PreparedPayment:
+    def __init__(
+        self,
+        client: Client,
+        created: dict[str, object],
+        *,
+        poll_interval: float,
+        timeout: float,
+        after: Callable[[PaymentPayload], Awaitable[None]],
+    ) -> None:
+        self.transaction_id = response_string(created.get("transactionId"))
+        self.approval_id = response_string(created.get("approvalId"))
+        self._approved = created.get("approvalStatus") == "APPROVED"
+        self._client = client
+        self._poll_interval = poll_interval
+        self._timeout = timeout
+        self._after = after
+        self._completion: asyncio.Task[EncodedPayment] | None = None
+        self._received = False
+        self._cancelled = False
+        self._cancellation: asyncio.Task[None] | None = None
+        self._waiters = 0
+
+    async def status(self) -> str:
+        return response_string((await self._read()).get("status"))
+
+    async def _read(self) -> dict[str, object]:
+        return response_object(
+            await self._client.request(
+                "GET", f"/v1/transactions/{quote(self.transaction_id, safe='')}/x402"
+            )
+        )
+
+    async def cancel(self) -> None:
+        self._cancelled = True
+        if self._completion is not None and not self._completion.done():
+            self._completion.cancel()
+        if self._cancellation is None:
+            self._cancellation = asyncio.create_task(self._client.cancel_approval(self.approval_id))
+        await asyncio.shield(self._cancellation)
+
+    async def await_payload(
+        self, *, poll_interval: float | None = None, timeout: float | None = None
+    ) -> EncodedPayment:
+        if self._cancelled:
+            raise X402PaymentError("payment-cancelled", "x402 payment cancelled")
+        interval = self._poll_interval if poll_interval is None else poll_interval
+        budget = self._timeout if timeout is None else timeout
+        validate_wait(interval, budget)
+        if self._completion is None:
+            self._completion = asyncio.create_task(self._resolve(interval, budget))
+            self._completion.add_done_callback(_observe_completion)
+        completion = self._completion
+        self._waiters += 1
+        try:
+            # A cancelled waiter must not cancel another caller's wait for this same payment.
+            result = await asyncio.shield(completion)
+            return EncodedPayment(
+                result.encoded_payload,
+                result.payment_payload.model_copy(deep=True),
+                result.transaction_id,
+            )
+        except asyncio.CancelledError:
+            if self._cancelled:
+                raise X402PaymentError("payment-cancelled", "x402 payment cancelled") from None
+            raise
+        finally:
+            self._waiters -= 1
+            if self._waiters == 0 and not self._received and not completion.done():
+                completion.cancel()
+                await asyncio.gather(completion, return_exceptions=True)
+            if completion.done() and not self._received and self._completion is completion:
+                self._completion = None
+
+    async def _resolve(self, interval: float, timeout: float) -> EncodedPayment:
+        deadline = asyncio.get_running_loop().time() + timeout
+        budget = asyncio.timeout_at(deadline)
+        first = self._approved
+        try:
+            async with budget:
+                while True:
+                    if asyncio.get_running_loop().time() >= deadline:
+                        raise X402PaymentError("payment-timeout", "x402 payment wait timed out")
+                    try:
+                        response = await self._read()
+                    except InflowApiError as error:
+                        if error.http_status not in (0, 429) and error.http_status < 500:
+                            raise
+                        response = {}
+                    if asyncio.get_running_loop().time() >= deadline:
+                        raise X402PaymentError("payment-timeout", "x402 payment wait timed out")
+                    if (
+                        response.get("encodedPayload") is not None
+                        and response.get("paymentPayload") is not None
+                    ):
+                        result = EncodedPayment(
+                            response_string(response["encodedPayload"]),
+                            PaymentPayload.model_validate(response["paymentPayload"]),
+                            self.transaction_id,
+                        )
+                        if result.payment_payload.x402_version != 2:
+                            raise X402PaymentError("invalid-response", "Expected x402 version 2")
+                        self._received = True
+                        break
+                    status = response.get("status")
+                    if status in ("DECLINED", "EXPIRED", "GENERAL_ERROR", "INSUFFICIENT_FUNDS"):
+                        raise X402PaymentError(
+                            "payment-failed", "x402 payment failed", status=status
+                        )
+                    if first:
+                        first = False
+                    else:
+                        await asyncio.sleep(interval)
+        except TimeoutError as error:
+            if budget.expired():
+                raise X402PaymentError("payment-timeout", "x402 payment wait timed out") from error
+            raise
+        # Cache completion, including hook errors: repeated waits must not repeat hook side effects.
+        await self._after(result.payment_payload.model_copy(deep=True))
+        if self._cancelled:
+            raise X402PaymentError("payment-cancelled", "x402 payment cancelled")
+        return result
+
+    async def _close(self) -> None:
+        # Closing the client stops local waits; the caller still owns two-phase approvals.
+        self._cancelled = True
+        if self._completion is not None and not self._completion.done():
+            self._completion.cancel()
+        if self._completion is not None:
+            await asyncio.gather(self._completion, return_exceptions=True)
+        if self._cancellation is not None:
+            await asyncio.shield(self._cancellation)