From f038cf3be779074027f05e89aa9a40a3ba38560a Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Tue, 21 Jul 2026 12:27:08 -0400
Subject: [PATCH] Add machine-to-machine client authentication (#4583)
* Add M2M client credentials auth providers
Wrap the SDK's client_credentials and private_key_jwt OAuth providers as
FastMCP-idiomatic ClientCredentialsOAuthProvider and PrivateKeyJWTOAuthProvider,
enabling browser-free client authentication via Client(auth=...).
* Fix M2M token cache collision and explicit-scope drop
Namespace the token cache by client_id so distinct clients sharing one store don't overwrite each other's tokens; pin caller-supplied scopes so the token request keeps them; fix CodeQL URL-substring check in tests; drop unused logger.
* Preserve step-up scope union, scope-aware token cache, restore token expiry
Only pin the caller's explicit scopes on initial authorization, leaving the SDK's step-up scope union intact; namespace the token cache by requested scopes as well as client_id; restore persisted absolute expiry on init so an expired stored token is re-fetched.
* Skip expiry restore for non-expiring reloaded tokens
* Distinguish expires_in=0 from omitted when restoring expiry
* Scope step-up flag to the flow via ContextVar; runnable JWT signing example
---
docs/clients/auth/client-credentials.mdx | 89 +++
docs/development/v4-notes/change-register.mdx | 15 +
docs/docs.json | 1 +
fastmcp_slim/fastmcp/client/__init__.py | 9 +-
fastmcp_slim/fastmcp/client/auth/__init__.py | 15 +-
.../fastmcp/client/auth/client_credentials.py | 404 +++++++++++
fastmcp_slim/fastmcp/client/auth/oauth.py | 24 +-
.../fastmcp/client/transports/http.py | 9 +
fastmcp_slim/fastmcp/client/transports/sse.py | 9 +
tests/client/auth/test_client_credentials.py | 685 ++++++++++++++++++
10 files changed, 1254 insertions(+), 6 deletions(-)
create mode 100644 docs/clients/auth/client-credentials.mdx
create mode 100644 fastmcp_slim/fastmcp/client/auth/client_credentials.py
create mode 100644 tests/client/auth/test_client_credentials.py
diff --git a/docs/clients/auth/client-credentials.mdx b/docs/clients/auth/client-credentials.mdx
new file mode 100644
index 000000000..382a5d8e9
--- /dev/null
+++ b/docs/clients/auth/client-credentials.mdx
@@ -0,0 +1,89 @@
+---
+title: Machine-to-Machine Authentication
+sidebarTitle: Client Credentials
+description: Authenticate your FastMCP client to a protected server without a browser.
+icon: robot
+---
+
+import { VersionBadge } from "/snippets/version-badge.mdx"
+
+
+
+
+Machine-to-machine authentication is only relevant for HTTP-based transports.
+
+
+When a FastMCP client runs without a human present — a backend service, a scheduled job, a CI pipeline, one MCP server calling another — it cannot complete the browser-based [OAuth](/clients/auth/oauth) flow. Instead it authenticates as itself using the OAuth 2.0 **client credentials** grant: the client presents its own credentials directly to the authorization server, receives an access token, and attaches that token to every request. There is no redirect, no consent screen, and no user.
+
+FastMCP provides two providers for this, both implementing the `httpx2.Auth` interface so they drop into the same `auth=` parameter as every other client auth option. You pass the **MCP server URL**, not a token endpoint — the token endpoint is discovered from the server's OAuth metadata, exactly as the interactive `OAuth` helper does. As with `OAuth`, you can omit the URL entirely and let the transport supply it.
+
+## Client ID and Secret
+
+The common case is a pre-registered client with an ID and a secret. Use `ClientCredentialsOAuthProvider` and pass it to the `auth` parameter of your `Client` or transport:
+
+```python {2, 4-8, 10}
+from fastmcp import Client
+from fastmcp.client.auth import ClientCredentialsOAuthProvider
+
+auth = ClientCredentialsOAuthProvider(
+ client_id="my-client-id",
+ client_secret="my-client-secret",
+ scopes=["read", "write"],
+)
+
+async with Client("https://example.com/mcp", auth=auth) as client:
+ await client.list_tools()
+```
+
+The provider discovers the authorization server, exchanges the credentials for an access token, and caches the token in memory for the life of the client. When the token expires it is re-acquired automatically on the next request. Because re-acquiring a token is a single non-interactive request, tokens are held in memory by default with no warning — unlike the interactive `OAuth` flow, losing the cache on restart costs nothing.
+
+### `ClientCredentialsOAuthProvider` Parameters
+
+- **`mcp_url`** (`str`, optional): Full URL to the MCP endpoint. Omit it when passing the provider to `Client(auth=...)` — the transport supplies the URL automatically.
+- **`client_id`** (`str`, required): The pre-registered OAuth client ID.
+- **`client_secret`** (`str`, required): The OAuth client secret.
+- **`scopes`** (`str | list[str]`, optional): Scopes to request, as a space-separated string or a list.
+- **`token_endpoint_auth_method`** (`"client_secret_basic" | "client_secret_post"`, optional): How the credentials are presented to the token endpoint. Defaults to `"client_secret_basic"` (an HTTP Basic `Authorization` header); use `"client_secret_post"` to send them in the request body instead.
+- **`token_storage`** (`AsyncKeyValue`, optional): A key-value store for the acquired token. Defaults to in-memory storage.
+
+## Private Key JWT
+
+Some authorization servers require the client to prove its identity with a signed JWT assertion (RFC 7523 `private_key_jwt`) instead of a shared secret. This is common with workload identity federation, where the assertion comes from a cloud identity provider. Use `PrivateKeyJWTOAuthProvider` and supply an `assertion_provider` — an async callback that receives the authorization server's issuer identifier (the required JWT audience) and returns the assertion.
+
+For a locally signed assertion, build the callback with `SignedJWTParameters`:
+
+```python {4-7, 9, 11-15, 17-20, 22}
+from pathlib import Path
+
+from fastmcp import Client
+from fastmcp.client.auth import (
+ PrivateKeyJWTOAuthProvider,
+ SignedJWTParameters,
+)
+
+private_key_pem = Path("client-signing-key.pem").read_text()
+
+jwt_params = SignedJWTParameters(
+ issuer="my-client-id",
+ subject="my-client-id",
+ signing_key=private_key_pem,
+)
+
+auth = PrivateKeyJWTOAuthProvider(
+ client_id="my-client-id",
+ assertion_provider=jwt_params.create_assertion_provider(),
+)
+
+async with Client("https://example.com/mcp", auth=auth) as client:
+ await client.list_tools()
+```
+
+If you already have a JWT from an identity provider, wrap it with `static_assertion_provider`, or pass your own `async def provider(audience: str) -> str` callback to fetch one on demand.
+
+### `PrivateKeyJWTOAuthProvider` Parameters
+
+- **`mcp_url`** (`str`, optional): Full URL to the MCP endpoint. Omit it when passing the provider to `Client(auth=...)`.
+- **`client_id`** (`str`, required): The OAuth client ID.
+- **`assertion_provider`** (`Callable[[str], Awaitable[str]]`, required): Async callback that receives the authorization server's issuer identifier and returns a signed JWT assertion.
+- **`scopes`** (`str | list[str]`, optional): Scopes to request, as a space-separated string or a list.
+- **`token_storage`** (`AsyncKeyValue`, optional): A key-value store for the acquired token. Defaults to in-memory storage.
diff --git a/docs/development/v4-notes/change-register.mdx b/docs/development/v4-notes/change-register.mdx
index a2ba5ce45..22c0ce33b 100644
--- a/docs/development/v4-notes/change-register.mdx
+++ b/docs/development/v4-notes/change-register.mdx
@@ -360,6 +360,21 @@ config = CacheConfig(store=store, partition="tenant-a", target_id="weather-api")
*Verify:* `fastmcp_slim/fastmcp/client/caching.py`, `tests/client/client/test_kv_response_cache.py`.
+### Machine-to-machine client auth — New (feature)
+
+`fastmcp.client.auth` gains two browser-free auth providers for the OAuth 2.0 `client_credentials` grant, closing the most common client-auth gap (previously only interactive `OAuth` and static `BearerAuth` were available). `ClientCredentialsOAuthProvider(client_id=..., client_secret=...)` authenticates with a client ID and secret; `PrivateKeyJWTOAuthProvider(client_id=..., assertion_provider=...)` uses an RFC 7523 `private_key_jwt` assertion (workload identity federation or a locally signed JWT via the re-exported `SignedJWTParameters` / `static_assertion_provider` helpers). Both are thin wrappers over the SDK's `mcp.client.auth.extensions.client_credentials` providers and implement `httpx2.Auth`, so they slot into the same `Client(auth=...)` path as every other provider. Like interactive `OAuth`, they take the MCP server URL (the token endpoint is discovered from OAuth metadata) and bind to it lazily — omit `mcp_url` and the transport supplies it. In-memory token storage is the default with no warning, since a lost M2M token is re-acquired in one non-interactive request.
+
+```python
+from fastmcp import Client
+from fastmcp.client.auth import ClientCredentialsOAuthProvider
+
+auth = ClientCredentialsOAuthProvider(client_id="id", client_secret="secret")
+async with Client("https://example.com/mcp", auth=auth) as client:
+ await client.list_tools()
+```
+
+*Verify:* `fastmcp_slim/fastmcp/client/auth/client_credentials.py`, `fastmcp_slim/fastmcp/client/transports/{http,sse}.py`, `tests/client/auth/test_client_credentials.py`.
+
## HTTP
The maintainer asked whether FastMCP can now delete its custom HTTP app and let the SDK's `Server.streamable_http_app()` handle everything. The answer for this PR is **no** — every override earns its keep. Convergence is a v4 project gated on three upstream additions (see [Feature Program](/development/v4-notes/feature-program)).
diff --git a/docs/docs.json b/docs/docs.json
index 55bff75f4..cf7801774 100644
--- a/docs/docs.json
+++ b/docs/docs.json
@@ -265,6 +265,7 @@
"icon": "key",
"pages": [
"clients/auth/oauth",
+ "clients/auth/client-credentials",
"clients/auth/cimd",
"clients/auth/bearer"
],
diff --git a/fastmcp_slim/fastmcp/client/__init__.py b/fastmcp_slim/fastmcp/client/__init__.py
index d5e65e229..9fc2f350b 100644
--- a/fastmcp_slim/fastmcp/client/__init__.py
+++ b/fastmcp_slim/fastmcp/client/__init__.py
@@ -1,7 +1,12 @@
from fastmcp import _install_hints
try:
- from .auth import OAuth, BearerAuth
+ from .auth import (
+ BearerAuth,
+ ClientCredentialsOAuthProvider,
+ OAuth,
+ PrivateKeyJWTOAuthProvider,
+ )
from .client import Client
from .transports import (
ClientTransport,
@@ -21,11 +26,13 @@ except ImportError as exc:
__all__ = [
"BearerAuth",
"Client",
+ "ClientCredentialsOAuthProvider",
"ClientTransport",
"FastMCPTransport",
"NodeStdioTransport",
"NpxStdioTransport",
"OAuth",
+ "PrivateKeyJWTOAuthProvider",
"PythonStdioTransport",
"SSETransport",
"StdioTransport",
diff --git a/fastmcp_slim/fastmcp/client/auth/__init__.py b/fastmcp_slim/fastmcp/client/auth/__init__.py
index 6ec3ecf4b..e706c7f18 100644
--- a/fastmcp_slim/fastmcp/client/auth/__init__.py
+++ b/fastmcp_slim/fastmcp/client/auth/__init__.py
@@ -1,4 +1,17 @@
from .bearer import BearerAuth
+from .client_credentials import (
+ ClientCredentialsOAuthProvider,
+ PrivateKeyJWTOAuthProvider,
+ SignedJWTParameters,
+ static_assertion_provider,
+)
from .oauth import OAuth
-__all__ = ["BearerAuth", "OAuth"]
+__all__ = [
+ "BearerAuth",
+ "ClientCredentialsOAuthProvider",
+ "OAuth",
+ "PrivateKeyJWTOAuthProvider",
+ "SignedJWTParameters",
+ "static_assertion_provider",
+]
diff --git a/fastmcp_slim/fastmcp/client/auth/client_credentials.py b/fastmcp_slim/fastmcp/client/auth/client_credentials.py
new file mode 100644
index 000000000..b6e8a6588
--- /dev/null
+++ b/fastmcp_slim/fastmcp/client/auth/client_credentials.py
@@ -0,0 +1,404 @@
+"""Machine-to-machine (M2M) OAuth client authentication for FastMCP.
+
+These providers authenticate a FastMCP client to a protected MCP server without
+a browser, using the OAuth 2.0 ``client_credentials`` grant:
+
+- `ClientCredentialsOAuthProvider` authenticates with a ``client_id`` and
+ ``client_secret`` (the common M2M case).
+- `PrivateKeyJWTOAuthProvider` authenticates with an RFC 7523 ``private_key_jwt``
+ client assertion (workload identity federation, or a locally signed JWT).
+
+Both are thin wrappers over the MCP SDK's client-credentials providers. Like the
+interactive `OAuth` provider, they can be constructed without an ``mcp_url`` and
+bound to the server URL automatically when passed to `Client(auth=...)`.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import AsyncGenerator, Awaitable, Callable
+from contextvars import ContextVar
+from typing import Literal
+
+import httpx2
+from key_value.aio.protocols import AsyncKeyValue
+from key_value.aio.stores.memory import MemoryStore
+from mcp.client.auth.extensions.client_credentials import (
+ ClientCredentialsOAuthProvider as _SDKClientCredentialsOAuthProvider,
+)
+from mcp.client.auth.extensions.client_credentials import (
+ PrivateKeyJWTOAuthProvider as _SDKPrivateKeyJWTOAuthProvider,
+)
+from mcp.client.auth.extensions.client_credentials import (
+ SignedJWTParameters,
+ static_assertion_provider,
+)
+from mcp.client.auth.oauth2 import OAuthContext
+from mcp.client.auth.utils import extract_field_from_www_auth
+from typing_extensions import override
+
+from fastmcp.client.auth.oauth import TokenStorageAdapter
+
+__all__ = [
+ "ClientCredentialsOAuthProvider",
+ "PrivateKeyJWTOAuthProvider",
+ "SignedJWTParameters",
+ "static_assertion_provider",
+]
+
+# Whether the auth flow currently being driven is a 403 step-up rather than an
+# initial authorization. A ContextVar (not instance state) so it stays scoped to
+# the single flow driving it: concurrent flows run in separate tasks and never
+# see each other's value, and each flow resets it on exit.
+_in_step_up: ContextVar[bool] = ContextVar("fastmcp_m2m_in_step_up", default=False)
+
+
+def _normalize_scopes(scopes: str | list[str] | None) -> str | None:
+ """Normalize scopes to a space-separated string (or None)."""
+ if isinstance(scopes, list):
+ return " ".join(scopes)
+ return scopes
+
+
+def _cache_namespace(client_id: str, scopes: str | None) -> str:
+ """Namespace cached tokens by both client identity and requested scopes.
+
+ Two providers that differ in either their ``client_id`` or their requested
+ scopes must not share cached tokens: a token issued for one client or one
+ scope set is not interchangeable with another. Hashing a canonical
+ ``(client_id, scopes)`` pair keeps the namespace unambiguous regardless of the
+ characters either value contains.
+ """
+ identity = json.dumps([client_id, scopes], separators=(",", ":"))
+ return hashlib.sha256(identity.encode()).hexdigest()
+
+
+def _resolve_token_storage(
+ token_storage: AsyncKeyValue | None,
+ mcp_url: str,
+ client_id: str,
+ scopes: str | None,
+) -> TokenStorageAdapter:
+ """Wrap a token store in the FastMCP adapter, defaulting to in-memory.
+
+ Unlike the interactive `OAuth` provider, M2M providers do not warn when using
+ in-memory storage: re-acquiring a token is a single non-interactive request,
+ so losing the cache on restart is cheap rather than disruptive.
+
+ The cache is namespaced by client identity and requested scopes so that
+ providers with different credentials or scope sets can share one store against
+ the same MCP endpoint without overwriting each other's tokens.
+ """
+ store = token_storage or MemoryStore()
+ return TokenStorageAdapter(
+ async_key_value=store,
+ server_url=mcp_url,
+ cache_namespace=_cache_namespace(client_id, scopes),
+ )
+
+
+def _is_insufficient_scope_challenge(response: httpx2.Response) -> bool:
+ """True when a response is an RFC 6750 ``insufficient_scope`` step-up challenge."""
+ if response.status_code != 403:
+ return False
+ return extract_field_from_www_auth(response, "error") == "insufficient_scope"
+
+
+async def _restore_token_expiry(context: OAuthContext) -> None:
+ """Restore the persisted absolute token expiry after a token is reloaded.
+
+ The inherited initializer reloads the stored token but not its expiry, so a
+ provider recreated with persistent storage would treat an already-expired
+ token as still valid. Reading the absolute expiry back keeps `is_token_valid`
+ honest, prompting a fresh token request when the stored one has expired.
+
+ The restore is skipped unless the reloaded token itself declares an
+ `expires_in`. A token whose response omitted `expires_in` (``None``) is
+ non-expiring, and the store may still hold a stale expiry from a previous
+ token it replaced; applying that would wrongly force a re-exchange. A token
+ that declares `expires_in=0` is immediately expired and keeps its recorded
+ expiry, so it is distinguished from an omitted one.
+ """
+ storage = context.storage
+ tokens = context.current_tokens
+ if tokens is None or tokens.expires_in is None:
+ return
+ if not isinstance(storage, TokenStorageAdapter):
+ return
+ expiry = await storage.get_token_expiry()
+ if expiry is not None:
+ context.token_expiry_time = expiry
+
+
+async def _drive_flow_tracking_step_up(
+ flow: AsyncGenerator[httpx2.Request, httpx2.Response],
+) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
+ """Delegate to the inherited auth flow, flagging step-up challenges.
+
+ On a 403 ``insufficient_scope`` the inherited flow unions the challenged scope
+ with the current one before re-requesting the token. Setting `_in_step_up`
+ lets `_perform_authorization` leave the accumulated scope in place instead of
+ re-pinning the caller's explicit scopes over it. The flag is set and reset
+ inside this generator, so it is scoped to exactly this flow.
+ """
+ token = _in_step_up.set(False)
+ try:
+ try:
+ outgoing = await anext(flow)
+ except StopAsyncIteration:
+ return
+ while True:
+ response = yield outgoing
+ if _is_insufficient_scope_challenge(response):
+ _in_step_up.set(True)
+ try:
+ outgoing = await flow.asend(response)
+ except StopAsyncIteration:
+ return
+ finally:
+ _in_step_up.reset(token)
+ await flow.aclose()
+
+
+class ClientCredentialsOAuthProvider(_SDKClientCredentialsOAuthProvider):
+ """OAuth ``client_credentials`` provider using a client ID and secret.
+
+ This is the standard machine-to-machine flow: the client exchanges its
+ ``client_id`` and ``client_secret`` at the authorization server's token
+ endpoint for an access token, which is then attached to every request. The
+ token endpoint is discovered from the MCP server's OAuth metadata, so callers
+ provide the MCP server URL rather than a raw token endpoint.
+
+ Example:
+ ```python
+ from fastmcp import Client
+ from fastmcp.client.auth import ClientCredentialsOAuthProvider
+
+ auth = ClientCredentialsOAuthProvider(
+ client_id="my-client-id",
+ client_secret="my-client-secret",
+ scopes=["read", "write"],
+ )
+
+ async with Client("https://example.com/mcp", auth=auth) as client:
+ await client.list_tools()
+ ```
+ """
+
+ _bound: bool
+
+ def __init__(
+ self,
+ mcp_url: str | None = None,
+ *,
+ client_id: str,
+ client_secret: str,
+ scopes: str | list[str] | None = None,
+ token_endpoint_auth_method: Literal[
+ "client_secret_basic", "client_secret_post"
+ ] = "client_secret_basic",
+ token_storage: AsyncKeyValue | None = None,
+ ) -> None:
+ """Initialize a client_credentials OAuth provider.
+
+ Args:
+ mcp_url: Full URL to the MCP endpoint (e.g. "https://host/mcp").
+ Optional when the provider is passed to `Client(auth=...)`, which
+ supplies the URL automatically from the transport.
+ client_id: The pre-registered OAuth client ID.
+ client_secret: The OAuth client secret.
+ scopes: OAuth scopes to request, as a space-separated string or a list
+ of strings.
+ token_endpoint_auth_method: How client credentials are presented to the
+ token endpoint. "client_secret_basic" (default) sends them in an
+ HTTP Basic ``Authorization`` header; "client_secret_post" sends them
+ in the request body.
+ token_storage: An AsyncKeyValue-compatible token store. Tokens are kept
+ in memory if not provided.
+ """
+ self._client_id = client_id
+ self._client_secret = client_secret
+ self._scopes = _normalize_scopes(scopes)
+ self._token_endpoint_auth_method = token_endpoint_auth_method
+ self._token_storage = token_storage
+ self._bound = False
+
+ if mcp_url is not None:
+ self._bind(mcp_url)
+
+ def _bind(self, mcp_url: str) -> None:
+ """Bind this provider to a specific MCP server URL.
+
+ Called automatically when ``mcp_url`` is provided to ``__init__``, or by the
+ transport when the provider is used without an explicit URL.
+ """
+ if self._bound:
+ return
+
+ mcp_url = mcp_url.rstrip("/")
+ super().__init__(
+ server_url=mcp_url,
+ storage=_resolve_token_storage(
+ self._token_storage, mcp_url, self._client_id, self._scopes
+ ),
+ client_id=self._client_id,
+ client_secret=self._client_secret,
+ token_endpoint_auth_method=self._token_endpoint_auth_method,
+ scopes=self._scopes,
+ )
+ self._bound = True
+
+ @override
+ async def _initialize(self) -> None:
+ await super()._initialize()
+ await _restore_token_expiry(self.context)
+
+ @override
+ def async_auth_flow(
+ self, request: httpx2.Request
+ ) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
+ if not self._bound:
+ raise RuntimeError(
+ "ClientCredentialsOAuthProvider has no server URL. Either pass "
+ "mcp_url to the constructor or use it with Client(auth=...), which "
+ "provides the URL automatically from the transport."
+ )
+ return _drive_flow_tracking_step_up(super().async_auth_flow(request))
+
+ @override
+ async def _perform_authorization(self) -> httpx2.Request:
+ # The inherited flow overwrites client_metadata.scope with the
+ # server-advertised scopes during 401 handling. Restore the caller's
+ # explicit scopes so the token request carries what the caller asked for.
+ # On a step-up the SDK unions the challenged scope with the current one;
+ # leave that accumulated scope in place instead of clobbering it.
+ if self._scopes is not None and not _in_step_up.get():
+ self.context.client_metadata.scope = self._scopes
+ return await super()._perform_authorization()
+
+
+class PrivateKeyJWTOAuthProvider(_SDKPrivateKeyJWTOAuthProvider):
+ """OAuth ``client_credentials`` provider using ``private_key_jwt`` (RFC 7523).
+
+ Instead of a shared client secret, the client authenticates to the token
+ endpoint with a signed JWT assertion. The ``assertion_provider`` callback
+ receives the authorization server's issuer identifier (the required JWT
+ audience) and returns the assertion. Use
+ `SignedJWTParameters.create_assertion_provider()` to sign locally with a
+ private key, `static_assertion_provider()` for a pre-built JWT, or supply your
+ own callback for workload identity federation.
+
+ Example:
+ ```python
+ from pathlib import Path
+
+ from fastmcp import Client
+ from fastmcp.client.auth import (
+ PrivateKeyJWTOAuthProvider,
+ SignedJWTParameters,
+ )
+
+ private_key_pem = Path("client-signing-key.pem").read_text()
+
+ jwt_params = SignedJWTParameters(
+ issuer="my-client-id",
+ subject="my-client-id",
+ signing_key=private_key_pem,
+ )
+ auth = PrivateKeyJWTOAuthProvider(
+ client_id="my-client-id",
+ assertion_provider=jwt_params.create_assertion_provider(),
+ )
+
+ async with Client("https://example.com/mcp", auth=auth) as client:
+ await client.list_tools()
+ ```
+ """
+
+ _bound: bool
+
+ def __init__(
+ self,
+ mcp_url: str | None = None,
+ *,
+ client_id: str,
+ assertion_provider: Callable[[str], Awaitable[str]],
+ scopes: str | list[str] | None = None,
+ token_storage: AsyncKeyValue | None = None,
+ ) -> None:
+ """Initialize a private_key_jwt OAuth provider.
+
+ Args:
+ mcp_url: Full URL to the MCP endpoint (e.g. "https://host/mcp").
+ Optional when the provider is passed to `Client(auth=...)`, which
+ supplies the URL automatically from the transport.
+ client_id: The OAuth client ID.
+ assertion_provider: Async callback that receives the authorization
+ server's issuer identifier (the JWT audience) and returns a signed
+ JWT assertion. Use `SignedJWTParameters.create_assertion_provider()`
+ for locally signed JWTs, `static_assertion_provider()` for a
+ pre-built JWT, or provide your own callback for workload identity
+ federation.
+ scopes: OAuth scopes to request, as a space-separated string or a list
+ of strings.
+ token_storage: An AsyncKeyValue-compatible token store. Tokens are kept
+ in memory if not provided.
+ """
+ self._client_id = client_id
+ self._assertion_provider = assertion_provider
+ self._scopes = _normalize_scopes(scopes)
+ self._token_storage = token_storage
+ self._bound = False
+
+ if mcp_url is not None:
+ self._bind(mcp_url)
+
+ def _bind(self, mcp_url: str) -> None:
+ """Bind this provider to a specific MCP server URL.
+
+ Called automatically when ``mcp_url`` is provided to ``__init__``, or by the
+ transport when the provider is used without an explicit URL.
+ """
+ if self._bound:
+ return
+
+ mcp_url = mcp_url.rstrip("/")
+ super().__init__(
+ server_url=mcp_url,
+ storage=_resolve_token_storage(
+ self._token_storage, mcp_url, self._client_id, self._scopes
+ ),
+ client_id=self._client_id,
+ assertion_provider=self._assertion_provider,
+ scopes=self._scopes,
+ )
+ self._bound = True
+
+ @override
+ async def _initialize(self) -> None:
+ await super()._initialize()
+ await _restore_token_expiry(self.context)
+
+ @override
+ def async_auth_flow(
+ self, request: httpx2.Request
+ ) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
+ if not self._bound:
+ raise RuntimeError(
+ "PrivateKeyJWTOAuthProvider has no server URL. Either pass mcp_url "
+ "to the constructor or use it with Client(auth=...), which provides "
+ "the URL automatically from the transport."
+ )
+ return _drive_flow_tracking_step_up(super().async_auth_flow(request))
+
+ @override
+ async def _perform_authorization(self) -> httpx2.Request:
+ # The inherited flow overwrites client_metadata.scope with the
+ # server-advertised scopes during 401 handling. Restore the caller's
+ # explicit scopes so the token request carries what the caller asked for.
+ # On a step-up the SDK unions the challenged scope with the current one;
+ # leave that accumulated scope in place instead of clobbering it.
+ if self._scopes is not None and not _in_step_up.get():
+ self.context.client_metadata.scope = self._scopes
+ return await super()._perform_authorization()
diff --git a/fastmcp_slim/fastmcp/client/auth/oauth.py b/fastmcp_slim/fastmcp/client/auth/oauth.py
index a854501e6..a73ea14aa 100644
--- a/fastmcp_slim/fastmcp/client/auth/oauth.py
+++ b/fastmcp_slim/fastmcp/client/auth/oauth.py
@@ -87,12 +87,19 @@ async def check_if_auth_required(
class TokenStorageAdapter(TokenStorage):
_server_url: str
+ _cache_namespace: str | None
_key_value_store: AsyncKeyValue
_storage_oauth_token: PydanticAdapter[OAuthToken]
_storage_client_info: PydanticAdapter[OAuthClientInformationFull]
- def __init__(self, async_key_value: AsyncKeyValue, server_url: str):
+ def __init__(
+ self,
+ async_key_value: AsyncKeyValue,
+ server_url: str,
+ cache_namespace: str | None = None,
+ ):
self._server_url = server_url
+ self._cache_namespace = cache_namespace
self._key_value_store = async_key_value
self._storage_oauth_token = PydanticAdapter[OAuthToken](
default_collection="mcp-oauth-token",
@@ -107,14 +114,23 @@ class TokenStorageAdapter(TokenStorage):
raise_on_validation_error=True,
)
+ def _cache_key_prefix(self) -> str:
+ # When set, the namespace distinguishes clients that share one store
+ # against the same server URL (e.g. M2M providers with different
+ # client_ids). Without it, the prefix is the bare server URL, preserving
+ # the existing keys used by the interactive OAuth flow.
+ if self._cache_namespace is not None:
+ return f"{self._server_url}/{self._cache_namespace}"
+ return self._server_url
+
def _get_token_cache_key(self) -> str:
- return f"{self._server_url}/tokens"
+ return f"{self._cache_key_prefix()}/tokens"
def _get_client_info_cache_key(self) -> str:
- return f"{self._server_url}/client_info"
+ return f"{self._cache_key_prefix()}/client_info"
def _get_token_expiry_cache_key(self) -> str:
- return f"{self._server_url}/token_expiry"
+ return f"{self._cache_key_prefix()}/token_expiry"
async def clear(self) -> None:
await self._storage_oauth_token.delete(key=self._get_token_cache_key())
diff --git a/fastmcp_slim/fastmcp/client/transports/http.py b/fastmcp_slim/fastmcp/client/transports/http.py
index 51b6c595c..3ba827931 100644
--- a/fastmcp_slim/fastmcp/client/transports/http.py
+++ b/fastmcp_slim/fastmcp/client/transports/http.py
@@ -15,6 +15,10 @@ from pydantic import AnyUrl
from typing_extensions import Unpack
from fastmcp.client.auth.bearer import BearerAuth
+from fastmcp.client.auth.client_credentials import (
+ ClientCredentialsOAuthProvider,
+ PrivateKeyJWTOAuthProvider,
+)
from fastmcp.client.auth.oauth import OAuth
from fastmcp.client.dependencies import get_http_headers
from fastmcp.client.transports.base import (
@@ -112,6 +116,11 @@ class StreamableHttpTransport(ClientTransport):
if factory is not None:
auth.httpx_client_factory = factory
resolved = auth
+ elif isinstance(
+ auth, (ClientCredentialsOAuthProvider, PrivateKeyJWTOAuthProvider)
+ ):
+ auth._bind(self.url)
+ resolved = auth
elif isinstance(auth, str):
resolved = BearerAuth(auth)
else:
diff --git a/fastmcp_slim/fastmcp/client/transports/sse.py b/fastmcp_slim/fastmcp/client/transports/sse.py
index 84dbdadda..ed5602444 100644
--- a/fastmcp_slim/fastmcp/client/transports/sse.py
+++ b/fastmcp_slim/fastmcp/client/transports/sse.py
@@ -16,6 +16,10 @@ from pydantic import AnyUrl
from typing_extensions import Unpack
from fastmcp.client.auth.bearer import BearerAuth
+from fastmcp.client.auth.client_credentials import (
+ ClientCredentialsOAuthProvider,
+ PrivateKeyJWTOAuthProvider,
+)
from fastmcp.client.auth.oauth import OAuth
from fastmcp.client.dependencies import get_http_headers
from fastmcp.client.transports.base import (
@@ -88,6 +92,11 @@ class SSETransport(ClientTransport):
if factory is not None:
auth.httpx_client_factory = factory
resolved = auth
+ elif isinstance(
+ auth, (ClientCredentialsOAuthProvider, PrivateKeyJWTOAuthProvider)
+ ):
+ auth._bind(self.url)
+ resolved = auth
elif isinstance(auth, str):
resolved = BearerAuth(auth)
else:
diff --git a/tests/client/auth/test_client_credentials.py b/tests/client/auth/test_client_credentials.py
new file mode 100644
index 000000000..910870fb4
--- /dev/null
+++ b/tests/client/auth/test_client_credentials.py
@@ -0,0 +1,685 @@
+"""Tests for machine-to-machine (M2M) client authentication.
+
+These cover the ``client_credentials`` grant (client_id + client_secret) and the
+RFC 7523 ``private_key_jwt`` variant. Rather than standing up a real
+authorization server (the in-memory server does not implement the
+client_credentials grant), the tests drive the provider's ``async_auth_flow``
+directly with a mock responder that answers OAuth discovery and the token
+endpoint, exactly as httpx would while running the auth flow.
+"""
+
+import base64
+import warnings
+from collections.abc import Callable
+from contextlib import aclosing
+from urllib.parse import urlparse
+
+import httpx2
+import jwt
+import pytest
+from key_value.aio.stores.memory import MemoryStore
+from mcp.client.auth import OAuthTokenError
+from mcp.shared.auth import OAuthToken
+
+from fastmcp.client import Client
+from fastmcp.client.auth import (
+ ClientCredentialsOAuthProvider,
+ PrivateKeyJWTOAuthProvider,
+ SignedJWTParameters,
+ static_assertion_provider,
+)
+from fastmcp.client.transports import SSETransport, StreamableHttpTransport
+
+SERVER_URL = "https://mcp.example.com/mcp"
+AUTH_SERVER_URL = "https://auth.example.com"
+# 32+ bytes so PyJWT does not warn about weak HMAC keys under -W error.
+SIGNING_KEY = "unit-test-signing-key-padded-to-32b"
+
+
+def make_m2m_responder(
+ *,
+ token_response: dict,
+ token_status: int = 200,
+ server_url: str = SERVER_URL,
+ auth_server_url: str = AUTH_SERVER_URL,
+ prm_scopes_supported: list[str] | None = None,
+) -> tuple[Callable[[httpx2.Request], httpx2.Response], dict[str, httpx2.Request]]:
+ """Build a responder for the standard M2M discovery + token exchange flow.
+
+ Returns the responder and a dict that captures the token request and the
+ final (retried) resource request for assertions. When ``prm_scopes_supported``
+ is set, the protected-resource metadata advertises those scopes, which the SDK
+ flow would otherwise apply to the token request.
+ """
+ captured: dict[str, httpx2.Request] = {}
+
+ def responder(request: httpx2.Request) -> httpx2.Response:
+ url = str(request.url)
+ path = urlparse(url).path
+ header_keys = {key.lower() for key in request.headers}
+
+ if url.startswith(server_url):
+ if "authorization" in header_keys:
+ captured["final_request"] = request
+ return httpx2.Response(200, text="ok")
+ return httpx2.Response(401, headers={"WWW-Authenticate": "Bearer"})
+
+ if path.startswith("/.well-known/oauth-protected-resource"):
+ prm: dict = {
+ "resource": server_url,
+ "authorization_servers": [auth_server_url],
+ }
+ if prm_scopes_supported is not None:
+ prm["scopes_supported"] = prm_scopes_supported
+ return httpx2.Response(200, json=prm)
+
+ if path.startswith(
+ "/.well-known/oauth-authorization-server"
+ ) or path.startswith("/.well-known/openid-configuration"):
+ return httpx2.Response(
+ 200,
+ json={
+ "issuer": auth_server_url,
+ "authorization_endpoint": f"{auth_server_url}/authorize",
+ "token_endpoint": f"{auth_server_url}/token",
+ "response_types_supported": ["code"],
+ },
+ )
+
+ if url == f"{auth_server_url}/token":
+ captured["token_request"] = request
+ return httpx2.Response(token_status, json=token_response)
+
+ raise AssertionError(f"unexpected request: {request.method} {url}")
+
+ return responder, captured
+
+
+async def drive_auth_flow(
+ provider: httpx2.Auth,
+ responder: Callable[[httpx2.Request], httpx2.Response],
+ *,
+ server_url: str = SERVER_URL,
+) -> list[httpx2.Request]:
+ """Drive an httpx auth flow to completion, feeding each yield to responder."""
+ requests: list[httpx2.Request] = []
+ async with aclosing(
+ provider.async_auth_flow(httpx2.Request("POST", server_url))
+ ) as flow:
+ sent: httpx2.Response | None = None
+ while True:
+ try:
+ request = await flow.asend(sent) # ty: ignore[invalid-argument-type]
+ except StopAsyncIteration:
+ break
+ requests.append(request)
+ sent = responder(request)
+ return requests
+
+
+def form_body(request: httpx2.Request) -> dict[str, str]:
+ """Parse an x-www-form-urlencoded request body into a dict."""
+ return dict(httpx2.QueryParams(request.content.decode()))
+
+
+class TestClientCredentialsConstruction:
+ """Constructor ergonomics and deferred binding."""
+
+ def test_deferred_binding(self):
+ provider = ClientCredentialsOAuthProvider(
+ client_id="cid", client_secret="secret"
+ )
+ assert provider._bound is False
+
+ provider._bind(f"{SERVER_URL}/")
+ assert provider._bound is True
+ # Trailing slash is normalized away.
+ assert provider.context.server_url == SERVER_URL
+
+ def test_binding_at_construction(self):
+ provider = ClientCredentialsOAuthProvider(
+ SERVER_URL, client_id="cid", client_secret="secret"
+ )
+ assert provider._bound is True
+ assert provider.context.server_url == SERVER_URL
+
+ def test_bind_is_idempotent(self):
+ provider = ClientCredentialsOAuthProvider(
+ SERVER_URL, client_id="cid", client_secret="secret"
+ )
+ provider._bind("https://other.example.com/mcp")
+ assert provider.context.server_url == SERVER_URL
+
+ @pytest.mark.parametrize(
+ "scopes, expected",
+ [
+ (["read", "write"], "read write"),
+ ("read write", "read write"),
+ (None, None),
+ ],
+ )
+ def test_scope_normalization(self, scopes, expected):
+ provider = ClientCredentialsOAuthProvider(
+ SERVER_URL, client_id="cid", client_secret="secret", scopes=scopes
+ )
+ assert provider.context.client_metadata.scope == expected
+
+ def test_default_token_endpoint_auth_method(self):
+ provider = ClientCredentialsOAuthProvider(
+ SERVER_URL, client_id="cid", client_secret="secret"
+ )
+ assert (
+ provider._fixed_client_info.token_endpoint_auth_method
+ == "client_secret_basic"
+ )
+
+ def test_token_endpoint_auth_method_override(self):
+ provider = ClientCredentialsOAuthProvider(
+ SERVER_URL,
+ client_id="cid",
+ client_secret="secret",
+ token_endpoint_auth_method="client_secret_post",
+ )
+ assert (
+ provider._fixed_client_info.token_endpoint_auth_method
+ == "client_secret_post"
+ )
+
+ def test_in_memory_storage_does_not_warn(self):
+ """M2M re-acquires tokens cheaply, so no in-memory storage warning."""
+ with warnings.catch_warnings():
+ warnings.simplefilter("error")
+ ClientCredentialsOAuthProvider(
+ SERVER_URL, client_id="cid", client_secret="secret"
+ )
+
+ async def test_unbound_provider_raises(self):
+ provider = ClientCredentialsOAuthProvider(
+ client_id="cid", client_secret="secret"
+ )
+ with pytest.raises(RuntimeError, match="has no server URL"):
+ provider.async_auth_flow(httpx2.Request("POST", SERVER_URL))
+
+
+class TestClientCredentialsFlow:
+ """The provider discovers the token endpoint, acquires and attaches a token."""
+
+ async def test_acquires_and_attaches_token(self):
+ provider = ClientCredentialsOAuthProvider(
+ SERVER_URL, client_id="cid", client_secret="secret"
+ )
+ responder, captured = make_m2m_responder(
+ token_response={
+ "access_token": "ACCESS123",
+ "token_type": "Bearer",
+ "expires_in": 3600,
+ }
+ )
+
+ requests = await drive_auth_flow(provider, responder)
+
+ # The token exchange used the client_credentials grant.
+ token_body = form_body(captured["token_request"])
+ assert token_body["grant_type"] == "client_credentials"
+
+ # The retried request carries the acquired bearer token.
+ assert requests[-1].headers["Authorization"] == "Bearer ACCESS123"
+ assert captured["final_request"].headers["Authorization"] == "Bearer ACCESS123"
+
+ async def test_client_secret_basic_uses_authorization_header(self):
+ provider = ClientCredentialsOAuthProvider(
+ SERVER_URL,
+ client_id="cid",
+ client_secret="secret",
+ token_endpoint_auth_method="client_secret_basic",
+ )
+ responder, captured = make_m2m_responder(
+ token_response={"access_token": "T", "token_type": "Bearer"}
+ )
+
+ await drive_auth_flow(provider, responder)
+
+ token_request = captured["token_request"]
+ expected = base64.b64encode(b"cid:secret").decode()
+ assert token_request.headers["Authorization"] == f"Basic {expected}"
+ # Credentials are not duplicated in the body.
+ assert "client_secret" not in form_body(token_request)
+
+ async def test_client_secret_post_uses_body(self):
+ provider = ClientCredentialsOAuthProvider(
+ SERVER_URL,
+ client_id="cid",
+ client_secret="secret",
+ token_endpoint_auth_method="client_secret_post",
+ )
+ responder, captured = make_m2m_responder(
+ token_response={"access_token": "T", "token_type": "Bearer"}
+ )
+
+ await drive_auth_flow(provider, responder)
+
+ token_body = form_body(captured["token_request"])
+ assert token_body["client_id"] == "cid"
+ assert token_body["client_secret"] == "secret"
+ assert "Authorization" not in captured["token_request"].headers
+
+ async def test_token_error_surfaces(self):
+ provider = ClientCredentialsOAuthProvider(
+ SERVER_URL, client_id="cid", client_secret="wrong"
+ )
+ responder, _ = make_m2m_responder(
+ token_response={"error": "invalid_client"},
+ token_status=401,
+ )
+
+ with pytest.raises(OAuthTokenError, match="Token exchange failed"):
+ await drive_auth_flow(provider, responder)
+
+ async def test_explicit_scopes_win_over_server_advertised(self):
+ """A caller's explicit scopes reach the token request even when the
+ server advertises a different set (the inherited flow would otherwise
+ overwrite them during 401 handling)."""
+ provider = ClientCredentialsOAuthProvider(
+ SERVER_URL,
+ client_id="cid",
+ client_secret="secret",
+ scopes=["read", "write"],
+ )
+ responder, captured = make_m2m_responder(
+ token_response={"access_token": "T", "token_type": "Bearer"},
+ prm_scopes_supported=["admin", "superuser"],
+ )
+
+ await drive_auth_flow(provider, responder)
+
+ token_body = form_body(captured["token_request"])
+ assert token_body["scope"] == "read write"
+
+
+class TestTokenCacheIsolation:
+ """Cached tokens are namespaced by client identity, not just server URL."""
+
+ async def test_distinct_client_ids_do_not_share_cached_tokens(self):
+ """Two providers with different client_ids sharing one store against the
+ same endpoint each retain their own token instead of clobbering one
+ another."""
+ store = MemoryStore()
+
+ provider_a = ClientCredentialsOAuthProvider(
+ SERVER_URL,
+ client_id="client-a",
+ client_secret="secret-a",
+ token_storage=store,
+ )
+ provider_b = ClientCredentialsOAuthProvider(
+ SERVER_URL,
+ client_id="client-b",
+ client_secret="secret-b",
+ token_storage=store,
+ )
+
+ responder_a, _ = make_m2m_responder(
+ token_response={
+ "access_token": "TOKEN-A",
+ "token_type": "Bearer",
+ "expires_in": 3600,
+ }
+ )
+ responder_b, _ = make_m2m_responder(
+ token_response={
+ "access_token": "TOKEN-B",
+ "token_type": "Bearer",
+ "expires_in": 3600,
+ }
+ )
+
+ await drive_auth_flow(provider_a, responder_a)
+ requests_b = await drive_auth_flow(provider_b, responder_b)
+
+ # provider_b acquires and uses its own token rather than reloading the
+ # token provider_a wrote to the shared store.
+ assert requests_b[-1].headers["Authorization"] == "Bearer TOKEN-B"
+
+ # Each client's token is preserved under its own namespace.
+ tokens_a = await provider_a.context.storage.get_tokens()
+ tokens_b = await provider_b.context.storage.get_tokens()
+ assert tokens_a is not None and tokens_a.access_token == "TOKEN-A"
+ assert tokens_b is not None and tokens_b.access_token == "TOKEN-B"
+
+ async def test_distinct_scopes_do_not_share_cached_tokens(self):
+ """Two providers with the same client_id but different requested scopes
+ sharing one store each retain their own token: a token issued for one
+ scope set must not be reused for another."""
+ store = MemoryStore()
+
+ provider_read = ClientCredentialsOAuthProvider(
+ SERVER_URL,
+ client_id="cid",
+ client_secret="secret",
+ scopes=["read"],
+ token_storage=store,
+ )
+ provider_write = ClientCredentialsOAuthProvider(
+ SERVER_URL,
+ client_id="cid",
+ client_secret="secret",
+ scopes=["write"],
+ token_storage=store,
+ )
+
+ responder_read, _ = make_m2m_responder(
+ token_response={
+ "access_token": "TOKEN-READ",
+ "token_type": "Bearer",
+ "expires_in": 3600,
+ }
+ )
+ responder_write, _ = make_m2m_responder(
+ token_response={
+ "access_token": "TOKEN-WRITE",
+ "token_type": "Bearer",
+ "expires_in": 3600,
+ }
+ )
+
+ await drive_auth_flow(provider_read, responder_read)
+ requests_write = await drive_auth_flow(provider_write, responder_write)
+
+ # The write-scoped provider acquires its own token instead of reloading
+ # the read-scoped token from the shared store.
+ assert requests_write[-1].headers["Authorization"] == "Bearer TOKEN-WRITE"
+
+ tokens_read = await provider_read.context.storage.get_tokens()
+ tokens_write = await provider_write.context.storage.get_tokens()
+ assert tokens_read is not None and tokens_read.access_token == "TOKEN-READ"
+ assert tokens_write is not None and tokens_write.access_token == "TOKEN-WRITE"
+
+
+class TestPersistentTokenExpiry:
+ """A token reloaded from persistent storage honors its stored expiry."""
+
+ @pytest.mark.parametrize("expires_in", [-100, 0])
+ async def test_expired_stored_token_is_refetched(self, expires_in):
+ """Recreating a provider against a store holding an already-expired token
+ re-fetches instead of trusting the stale token as if it never expires.
+
+ `expires_in=0` is the boundary: an immediately-expiring token still
+ declares an expiry, so it must not be mistaken for a non-expiring one.
+ """
+ store = MemoryStore()
+
+ def make_provider() -> ClientCredentialsOAuthProvider:
+ return ClientCredentialsOAuthProvider(
+ SERVER_URL,
+ client_id="cid",
+ client_secret="secret",
+ scopes=["read"],
+ token_storage=store,
+ )
+
+ # Seed the shared store with a token whose absolute expiry is in the past.
+ seed_provider = make_provider()
+ await seed_provider.context.storage.set_tokens(
+ OAuthToken(
+ access_token="STALE-TOKEN",
+ token_type="Bearer",
+ expires_in=expires_in,
+ )
+ )
+
+ provider = make_provider()
+ responder, _ = make_m2m_responder(
+ token_response={
+ "access_token": "FRESH-TOKEN",
+ "token_type": "Bearer",
+ "expires_in": 3600,
+ }
+ )
+
+ requests = await drive_auth_flow(provider, responder)
+
+ assert requests[-1].headers["Authorization"] == "Bearer FRESH-TOKEN"
+
+ async def test_nonexpiring_token_ignores_stale_stored_expiry(self):
+ """A reloaded token without `expires_in` is non-expiring and must not
+ inherit a stale expiry left by a previous token it replaced."""
+ store = MemoryStore()
+
+ def make_provider() -> ClientCredentialsOAuthProvider:
+ return ClientCredentialsOAuthProvider(
+ SERVER_URL,
+ client_id="cid",
+ client_secret="secret",
+ scopes=["read"],
+ token_storage=store,
+ )
+
+ # Record a stale past expiry, then replace the token with a non-expiring
+ # one — set_tokens leaves the earlier expiry record in place.
+ seed = make_provider()
+ await seed.context.storage.set_tokens(
+ OAuthToken(access_token="OLD", token_type="Bearer", expires_in=-100)
+ )
+ await seed.context.storage.set_tokens(
+ OAuthToken(access_token="NONEXPIRING", token_type="Bearer")
+ )
+
+ provider = make_provider()
+ responder, _ = make_m2m_responder(
+ token_response={
+ "access_token": "FRESH-TOKEN",
+ "token_type": "Bearer",
+ "expires_in": 3600,
+ }
+ )
+
+ requests = await drive_auth_flow(provider, responder)
+
+ # The non-expiring token is used as-is; the stale expiry does not force
+ # a needless re-exchange to FRESH-TOKEN.
+ assert requests[-1].headers["Authorization"] == "Bearer NONEXPIRING"
+
+
+class TestStepUpScopeAccumulation:
+ """A 403 insufficient_scope step-up unions the challenged scope with the
+ caller's scopes instead of dropping the accumulated grant."""
+
+ async def test_step_up_requests_union_of_scopes(self):
+ token_scopes: list[str] = []
+ require_write = False
+
+ def responder(request: httpx2.Request) -> httpx2.Response:
+ url = str(request.url)
+ path = urlparse(url).path
+
+ if url.startswith(SERVER_URL):
+ auth = request.headers.get("Authorization", "")
+ if not auth:
+ return httpx2.Response(401, headers={"WWW-Authenticate": "Bearer"})
+ granted = auth.removeprefix("Bearer ").split()
+ # Once the server begins demanding "write", a token lacking it is
+ # challenged for step-up rather than accepted.
+ if require_write and "write" not in granted:
+ return httpx2.Response(
+ 403,
+ headers={
+ "WWW-Authenticate": (
+ 'Bearer error="insufficient_scope", scope="write"'
+ )
+ },
+ )
+ return httpx2.Response(200, text="ok")
+
+ if path.startswith("/.well-known/oauth-protected-resource"):
+ return httpx2.Response(
+ 200,
+ json={
+ "resource": SERVER_URL,
+ "authorization_servers": [AUTH_SERVER_URL],
+ },
+ )
+
+ if path.startswith(
+ "/.well-known/oauth-authorization-server"
+ ) or path.startswith("/.well-known/openid-configuration"):
+ return httpx2.Response(
+ 200,
+ json={
+ "issuer": AUTH_SERVER_URL,
+ "authorization_endpoint": f"{AUTH_SERVER_URL}/authorize",
+ "token_endpoint": f"{AUTH_SERVER_URL}/token",
+ "response_types_supported": ["code"],
+ },
+ )
+
+ if url == f"{AUTH_SERVER_URL}/token":
+ scope = form_body(request).get("scope", "")
+ token_scopes.append(scope)
+ return httpx2.Response(
+ 200,
+ json={
+ "access_token": scope or "noscope",
+ "token_type": "Bearer",
+ "expires_in": 3600,
+ },
+ )
+
+ raise AssertionError(f"unexpected request: {request.method} {url}")
+
+ provider = ClientCredentialsOAuthProvider(
+ SERVER_URL, client_id="cid", client_secret="secret", scopes=["read"]
+ )
+
+ # Initial acquisition requests exactly the caller's scopes.
+ await drive_auth_flow(provider, responder)
+ assert token_scopes[0] == "read"
+
+ # The server now requires an additional scope for the operation.
+ require_write = True
+ await drive_auth_flow(provider, responder)
+
+ # The step-up token request carries the union, not just the caller's scope.
+ assert set(token_scopes[-1].split()) == {"read", "write"}
+
+
+class TestPrivateKeyJWTFlow:
+ """private_key_jwt builds a client assertion and attaches the token."""
+
+ async def test_signed_assertion_flow(self):
+ jwt_params = SignedJWTParameters(
+ issuer="cid",
+ subject="cid",
+ signing_key=SIGNING_KEY,
+ signing_algorithm="HS256",
+ )
+ provider = PrivateKeyJWTOAuthProvider(
+ SERVER_URL,
+ client_id="cid",
+ assertion_provider=jwt_params.create_assertion_provider(),
+ )
+ responder, captured = make_m2m_responder(
+ token_response={
+ "access_token": "JWT-ACCESS",
+ "token_type": "Bearer",
+ "expires_in": 3600,
+ }
+ )
+
+ requests = await drive_auth_flow(provider, responder)
+
+ token_body = form_body(captured["token_request"])
+ assert token_body["grant_type"] == "client_credentials"
+ assert (
+ token_body["client_assertion_type"]
+ == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
+ )
+
+ # The assertion is a JWT whose audience is the authorization server issuer.
+ claims = jwt.decode(
+ token_body["client_assertion"], options={"verify_signature": False}
+ )
+ assert claims["iss"] == "cid"
+ assert claims["sub"] == "cid"
+ # RFC 7523bis: the assertion audience is the auth server's issuer.
+ assert claims["aud"] == AUTH_SERVER_URL
+
+ assert requests[-1].headers["Authorization"] == "Bearer JWT-ACCESS"
+
+ async def test_static_assertion_flow(self):
+ prebuilt = jwt.encode(
+ {"iss": "cid", "sub": "cid", "aud": "anything"},
+ SIGNING_KEY,
+ algorithm="HS256",
+ )
+ provider = PrivateKeyJWTOAuthProvider(
+ SERVER_URL,
+ client_id="cid",
+ assertion_provider=static_assertion_provider(prebuilt),
+ )
+ responder, captured = make_m2m_responder(
+ token_response={"access_token": "S", "token_type": "Bearer"}
+ )
+
+ await drive_auth_flow(provider, responder)
+
+ token_body = form_body(captured["token_request"])
+ assert token_body["client_assertion"] == prebuilt
+
+ async def test_unbound_provider_raises(self):
+ provider = PrivateKeyJWTOAuthProvider(
+ client_id="cid",
+ assertion_provider=static_assertion_provider("token"),
+ )
+ with pytest.raises(RuntimeError, match="has no server URL"):
+ provider.async_auth_flow(httpx2.Request("POST", SERVER_URL))
+
+
+class TestTransportIntegration:
+ """Providers slot into a transport's ``auth=`` and bind to the URL."""
+
+ def test_streamable_http_transport_binds_client_credentials(self):
+ provider = ClientCredentialsOAuthProvider(
+ client_id="cid", client_secret="secret"
+ )
+ transport = StreamableHttpTransport(SERVER_URL, auth=provider)
+ assert transport.auth is provider
+ assert provider._bound is True
+ assert provider.context.server_url == SERVER_URL
+
+ def test_sse_transport_binds_private_key_jwt(self):
+ provider = PrivateKeyJWTOAuthProvider(
+ client_id="cid",
+ assertion_provider=static_assertion_provider("token"),
+ )
+ transport = SSETransport(SERVER_URL, auth=provider)
+ assert transport.auth is provider
+ assert provider._bound is True
+
+ def test_client_binds_provider_from_url(self):
+ provider = ClientCredentialsOAuthProvider(
+ client_id="cid", client_secret="secret"
+ )
+ Client(SERVER_URL, auth=provider)
+ assert provider._bound is True
+ assert provider.context.server_url == SERVER_URL
+
+
+def test_assertion_provider_signs_expected_audience():
+ """SignedJWTParameters produces an assertion bound to the given audience."""
+ jwt_params = SignedJWTParameters(
+ issuer="cid",
+ subject="cid",
+ signing_key=SIGNING_KEY,
+ signing_algorithm="HS256",
+ )
+ provider = jwt_params.create_assertion_provider()
+
+ async def _run():
+ return await provider("https://issuer.example.com")
+
+ import anyio
+
+ assertion = anyio.run(_run)
+ # The assertion is a JWT whose audience is exactly the requested issuer.
+ claims = jwt.decode(assertion, options={"verify_signature": False})
+ assert claims["aud"] == "https://issuer.example.com"