feat: support per-request API-key providers

README.md

@@ -140,8 +140,9 @@ service details. TAP does not enable or modify any payment route automatically.
 
 `ClientOptions` selects `production` (the default, `https://api.inflowpay.ai`) or
 `sandbox` (`https://sandbox.inflowpay.ai`). `base_url` overrides that destination.
-Use `api_key` for API-key authentication or an asynchronous `access_token` callback
-for Bearer authentication, not both. Omit both for anonymous requests. The callback
+Choose one of `api_key`, an asynchronous `api_key_provider`, or an asynchronous
+`access_token` callback. Omit all three for anonymous requests. `api_key_provider`
+returns the API key to send in `X-API-KEY`; its result is not cached. Each callback
 runs for each HTTP attempt, allowing your application to refresh a token; its errors
 propagate without being retried. It must support concurrent calls and cancellation.
 

src/inflowpay/_runtime.py

@@ -125,10 +125,18 @@ def __init__(self, options: ClientOptions) -> None:
             raise ValueError(
                 "base URL must be HTTP or HTTPS without credentials, query, or fragment"
             )
+        if (
+            sum(
+                value is not None
+                for value in (options.api_key, options.api_key_provider, options.access_token)
+            )
+            > 1
+        ):
+            raise ValueError("authentication options are mutually exclusive")
+        if options.api_key_provider is not None and not callable(options.api_key_provider):
+            raise ValueError("API key provider must be callable")
         if options.api_key is not None:
             _credential(options.api_key)
-            if options.access_token is not None:
-                raise ValueError("API key and access token provider are mutually exclusive")
         _positive(options.timeout)
         self.base_url = base.rstrip("/")
         self._options = options
@@ -192,6 +200,10 @@ async def _attempt(
     ) -> object:
         await asyncio.sleep(0)
         token = self._options.api_key
+        if self._options.api_key_provider is not None:
+            token = await self._options.api_key_provider()
+            _credential(token)
+            await asyncio.sleep(0)
         # Provider failures are application errors, not retryable transport failures.
         if self._options.access_token is not None:
             token = await self._options.access_token()
@@ -204,7 +216,7 @@ async def _attempt(
         if content is not None:
             headers["Content-Type"] = "application/json"
         if token is not None:
-            if self._options.api_key is not None:
+            if self._options.api_key is not None or self._options.api_key_provider is not None:
                 headers["X-API-KEY"] = token
             else:
                 headers["Authorization"] = f"Bearer {token}"

src/inflowpay/mpp/seller.py

@@ -185,7 +185,7 @@ async def create(cls, options: ClientOptions, *, method: str = "inflow") -> Self
             raise MppSellerConfigurationError(
                 "Seller supports inflow, tempo, stripe and card charges"
             )
-        if options.api_key is None:
+        if options.api_key is None and options.api_key_provider is None:
             raise MppSellerConfigurationError("Seller setup requires an InFlow Seller API key")
         client = Client(options)
         try:

src/inflowpay/options.py

@@ -10,6 +10,7 @@ class ClientOptions:
     environment: Literal["production", "sandbox"] = "production"
     base_url: str | None = None
     api_key: str | None = field(default=None, repr=False)
+    api_key_provider: Callable[[], Awaitable[str]] | None = field(default=None, repr=False)
     access_token: Callable[[], Awaitable[str]] | None = field(default=None, repr=False)
     timeout: float = 30.0
     # Ownership transfers to the SDK client; do not share this transport between clients.

src/inflowpay/x402/eip7702.py

@@ -216,7 +216,11 @@ def __init__(
         consent: Callable[[Authorization], Awaitable[bool]],
     ) -> None:
         # This public endpoint is independent of the managed Buyer's account credentials.
-        if options.api_key is not None or options.access_token is not None:
+        if (
+            options.api_key is not None
+            or options.api_key_provider is not None
+            or options.access_token is not None
+        ):
             raise ValueError("EIP-7702 sponsorship uses anonymous ClientOptions")
         self._owner = _address(signer.address)
         self._signer = signer

src/inflowpay/x402/facilitator.py

@@ -65,9 +65,13 @@ def __init__(self, client: Client, supported: SupportedResponse) -> None:
     @classmethod
     async def create(cls, options: ClientOptions, *, anonymous: bool = False) -> Self:
         if anonymous:
-            if options.api_key is not None or options.access_token is not None:
+            if (
+                options.api_key is not None
+                or options.api_key_provider is not None
+                or options.access_token is not None
+            ):
                 raise ValueError("Anonymous facilitator must not receive credentials")
-        elif options.api_key is None:
+        elif options.api_key is None and options.api_key_provider is None:
             raise ValueError("Facilitator setup requires a Seller API key or anonymous=True")
         client = Client(options)
         try:

src/inflowpay/x402/seller.py

@@ -94,7 +94,7 @@ def __init__(self, client: Client) -> None:
 
     @classmethod
     async def create(cls, options: ClientOptions) -> Self:
-        if options.api_key is None:
+        if options.api_key is None and options.api_key_provider is None:
             raise ValueError("Seller setup requires an InFlow Seller API key")
         seller = cls(Client(options))
         try:

tests/test_api_key_provider.py

@@ -0,0 +1,146 @@
+import asyncio
+from collections.abc import Awaitable, Callable
+from typing import cast
+
+import httpx
+import pytest
+
+from inflowpay import ClientOptions
+from inflowpay._runtime import Client
+from inflowpay.mpp.seller import Seller as MppSeller
+from inflowpay.x402.facilitator import Facilitator
+from inflowpay.x402.seller import Seller as X402Seller
+from test_mpp_seller import Platform as MppPlatform
+from test_runtime_native import native_server
+from test_x402_seller import Platform as X402Platform
+
+
+@pytest.mark.parametrize("kind", ["mpp", "x402", "facilitator"])
+async def test_seller_factories_accept_provider(kind: str) -> None:
+    calls = 0
+
+    async def provider() -> str:
+        nonlocal calls
+        calls += 1
+        return "secret"
+
+    platform = MppPlatform() if kind == "mpp" else X402Platform()
+    options = ClientOptions(api_key_provider=provider, transport=platform)
+    factory = MppSeller if kind == "mpp" else X402Seller if kind == "x402" else Facilitator
+    async with await factory.create(options):
+        assert calls > 0
+        assert all(request.headers["X-API-KEY"] == "secret" for request in platform.requests)
+
+
+async def test_provider_resolves_per_attempt() -> None:
+    calls = 0
+    received: list[str] = []
+
+    async def provider() -> str:
+        nonlocal calls
+        calls += 1
+        return f"key-{calls}"
+
+    async def handle(
+        path: str, headers: dict[str, str], body: bytes, writer: asyncio.StreamWriter
+    ) -> None:
+        received.append(headers["x-api-key"])
+        assert "authorization" not in headers
+        status = 503 if len(received) == 1 else 200
+        writer.write(
+            f"HTTP/1.1 {status} Test\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{{}}".encode()
+        )
+
+    async with (
+        native_server(handle) as base,
+        Client(ClientOptions(api_key_provider=provider, base_url=base)) as client,
+    ):
+        assert calls == 0
+        await client.request("GET", "/example", retries=1)
+        await client.request("GET", "/example")
+    assert received == ["key-1", "key-2", "key-3"]
+
+
+@pytest.mark.parametrize("value", ["", " ", "bad\nkey", "é"])
+async def test_invalid_provider_result(value: str) -> None:
+    async def provider() -> str:
+        return value
+
+    async def handle(request: httpx.Request) -> httpx.Response:
+        pytest.fail("unexpected HTTP request")
+
+    async with Client(
+        ClientOptions(api_key_provider=provider, transport=httpx.MockTransport(handle))
+    ) as client:
+        with pytest.raises(ValueError):
+            await client.request("GET", "/example", retries=3)
+
+
+async def test_provider_error_is_not_retried() -> None:
+    calls = 0
+    failure = RuntimeError("provider unavailable")
+
+    async def provider() -> str:
+        nonlocal calls
+        calls += 1
+        raise failure
+
+    async with Client(ClientOptions(api_key_provider=provider)) as client:
+        with pytest.raises(RuntimeError) as raised:
+            await client.request("GET", "/example", retries=3)
+    assert raised.value is failure
+    assert calls == 1
+
+
+async def test_concurrent_requests_resolve_independently() -> None:
+    calls = 0
+
+    async def provider() -> str:
+        nonlocal calls
+        calls += 1
+        key = f"key-{calls}"
+        await asyncio.sleep(0)
+        return key
+
+    async def handle(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json=request.headers["X-API-KEY"])
+
+    async with Client(
+        ClientOptions(api_key_provider=provider, transport=httpx.MockTransport(handle))
+    ) as client:
+        one, two = await asyncio.gather(
+            client.request("GET", "/one"), client.request("GET", "/two")
+        )
+        assert (one, two) == ("key-1", "key-2")
+
+
+async def test_provider_cancellation_before_http() -> None:
+    async def provider() -> str:
+        task = asyncio.current_task()
+        assert task is not None
+        task.cancel()
+        return "key"
+
+    async def handle(request: httpx.Request) -> httpx.Response:
+        pytest.fail("unexpected HTTP request")
+
+    async with Client(
+        ClientOptions(api_key_provider=provider, transport=httpx.MockTransport(handle))
+    ) as client:
+        task = asyncio.create_task(client.request("GET", "/example"))
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+
+def test_conflicting_providers() -> None:
+    async def provider() -> str:
+        return "key"
+
+    for options in (
+        ClientOptions(api_key="key", api_key_provider=provider),
+        ClientOptions(api_key_provider=provider, access_token=provider),
+    ):
+        with pytest.raises(ValueError, match="mutually exclusive"):
+            Client(options)
+    with pytest.raises(ValueError, match="callable"):
+        Client(ClientOptions(api_key_provider=cast(Callable[[], Awaitable[str]], "not callable")))

tests/test_x402_seller.py

@@ -197,6 +197,8 @@ async def token() -> str:
 
     with pytest.raises(ValueError):
         await Facilitator.create(ClientOptions(access_token=token), anonymous=True)
+    with pytest.raises(ValueError):
+        await Facilitator.create(ClientOptions(api_key_provider=token), anonymous=True)
     with pytest.raises(ValueError):
         await Seller.create(ClientOptions())
 

tests/test_x402_sponsorship.py

@@ -269,3 +269,9 @@ async def consent(_: Authorization) -> bool:
 
     with pytest.raises(ValueError, match="anonymous"):
         SponsorshipExtension(ClientOptions(api_key="test-key"), signer, consent)
+
+    async def provider() -> str:
+        return "test-key"
+
+    with pytest.raises(ValueError, match="anonymous"):
+        SponsorshipExtension(ClientOptions(api_key_provider=provider), signer, consent)