Let pre-provisioned OAuth clients name their authorization server

ClientCredentialsOAuthProvider and PrivateKeyJWTOAuthProvider take an
optional issuer keyword: the issuer identifier of the authorization
server the fixed client_id (and secret) were issued by. When set, token
requests, the client_credentials exchange and any refresh, are only built
from discovered authorization server metadata whose issuer matches it; if
discovery yields metadata for another server, or none at all, the flow
stops with OAuthFlowError before the secret is attached or an assertion
is minted, and the metadata and tokens held are dropped so the next
request starts discovery again. When the resource advertises several
authorization servers the one matching the configured issuer is used; the
comparison is issuers_match (exact, root slash aside); a value that is not
an http(s) URL is a ValueError. Omitting it keeps the current behaviour.

This is the same "the authorization server is configuration" model that
IdentityAssertionOAuthProvider already uses, made available to the two
older machine-to-machine providers without changing their defaults.

Author: maxisbey

docs/client/oauth-clients.md

@@ -105,13 +105,14 @@ A nightly job, a CI step, another service. There is no browser and nobody to cli
 
 `ClientCredentialsOAuthProvider` is the same `httpx2.Auth`, minus the human:
 
-```python title="client.py" hl_lines="4 27-33"
+```python title="client.py" hl_lines="4 27-34"
 --8<-- "docs_src/oauth_clients/tutorial002.py"
 ```
 
 What changed:
 
 * No `OAuthClientMetadata`, no handlers. You pass `client_id` and `client_secret`; the provider builds a minimal `client_credentials` registration around them and skips dynamic registration entirely.
+* `issuer` names the authorization server that issued those credentials; use the `issuer` value its `/.well-known/oauth-authorization-server` document returns. Discovery still runs as above, but token requests are only ever built from metadata for *that* issuer; if the MCP server points anywhere else, the flow stops with an `OAuthFlowError` instead. Leave it out and the provider uses whichever authorization server discovery finds.
 * `scope` is a space-separated string, the OAuth wire format.
 * Everything downstream is identical: the same `TokenStorage`, the same `httpx2.AsyncClient(auth=...)`, the same `streamable_http_client`.
 
@@ -124,7 +125,7 @@ By default the secret travels as HTTP Basic auth on the token request (`client_s
     One more provider lives in `mcp.client.auth.extensions.client_credentials`:
     **`PrivateKeyJWTOAuthProvider`**, for clients that authenticate with a JWT instead of a
     shared secret (`private_key_jwt`, the key-pair and workload-identity flavour). It follows
-    the same pattern: construct one, put it on `auth=`. The same module ships
+    the same pattern: construct one (it takes the same optional `issuer`), put it on `auth=`. The same module ships
     `SignedJWTParameters` and `static_assertion_provider`, two helpers that build its assertion.
 
 There is one more no-human situation: the client belongs to an enterprise whose identity provider, not the user, decides which MCP servers it may reach. That is a different grant with its own trust model and its own page, **[Identity assertion](identity-assertion.md)**.

docs_src/oauth_clients/tutorial002.py

@@ -30,6 +30,7 @@ async def set_client_info(self, client_info: OAuthClientInformationFull) -> None
     client_id="reporting-agent",
     client_secret="...",
     scope="user",
+    issuer="http://localhost:9000",
 )
 
 

src/mcp/client/auth/extensions/client_credentials.py

@@ -9,21 +9,58 @@
 import time
 from collections.abc import Awaitable, Callable
 from typing import Any, Literal
+from urllib.parse import urlparse
 from uuid import uuid4
 
 import httpx2
 import jwt
 from pydantic import BaseModel, Field
 
 from mcp.client.auth import OAuthClientProvider, OAuthFlowError, TokenStorage
+from mcp.client.auth.oauth2 import OAuthContext
+from mcp.client.auth.utils import issuers_match
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata
 
 
+def _checked_issuer(issuer: str | None) -> str | None:
+    if issuer is not None and urlparse(issuer).scheme not in ("http", "https"):
+        raise ValueError(f"issuer must be the authorization server's http(s) issuer URL, got {issuer!r}")
+    return issuer
+
+
+def _preferred_authorization_server(advertised: list[str], issuer: str | None) -> str:
+    """The advertised server matching the configured issuer if there is one, else the first."""
+    return next(
+        (server for server in advertised if issuer is not None and issuers_match(server, issuer)), advertised[0]
+    )
+
+
+def _require_metadata_for_configured_issuer(context: OAuthContext, issuer: str | None) -> None:
+    """With an issuer configured, a token request is only built from metadata discovered for that issuer.
+
+    Anything else held is dropped along with the tokens, so the next request starts discovery afresh
+    rather than refreshing against it.
+    """
+    if issuer is None:
+        return
+    metadata = context.oauth_metadata
+    if metadata is not None and issuers_match(str(metadata.issuer), issuer):
+        return
+    context.oauth_metadata = None
+    context.clear_tokens()
+    if metadata is None:
+        raise OAuthFlowError(f"No authorization server metadata discovered for configured issuer {issuer}")
+    raise OAuthFlowError(f"Authorization server metadata issuer mismatch: {metadata.issuer} != {issuer}")
+
+
 class ClientCredentialsOAuthProvider(OAuthClientProvider):
     """OAuth provider for client_credentials grant with client_id + client_secret.
 
     This provider sets client_info directly, bypassing dynamic client registration.
     Use this when you already have client credentials (client_id and client_secret).
+    Pass `issuer` to name the authorization server those credentials belong to: token
+    requests are then only built from authorization server metadata for that issuer, and
+    the flow stops if the MCP server leads anywhere else.
 
     Example:
         ```python
@@ -32,6 +69,7 @@ class ClientCredentialsOAuthProvider(OAuthClientProvider):
             storage=my_token_storage,
             client_id="my-client-id",
             client_secret="my-client-secret",
+            issuer="https://auth.example.com",
         )
         ```
     """
@@ -44,6 +82,7 @@ def __init__(
         client_secret: str,
         token_endpoint_auth_method: Literal["client_secret_basic", "client_secret_post"] = "client_secret_basic",
         scope: str | None = None,
+        issuer: str | None = None,
     ) -> None:
         """Initialize client_credentials OAuth provider.
 
@@ -55,6 +94,11 @@ def __init__(
             token_endpoint_auth_method: Authentication method for token endpoint.
                 Either "client_secret_basic" (default) or "client_secret_post".
             scope: Optional space-separated list of scopes to request.
+            issuer: The issuer identifier of the authorization server that issued
+                `client_id` and `client_secret`. When set, token requests are only built from
+                discovered authorization server metadata whose `issuer` is exactly this string;
+                otherwise the flow stops with `OAuthFlowError`. When omitted, whichever
+                authorization server discovery yields is used.
         """
         # Build minimal client_metadata for the base class
         client_metadata = OAuthClientMetadata(
@@ -64,6 +108,7 @@ def __init__(
             scope=scope,
         )
         super().__init__(server_url, client_metadata, storage, None, None)
+        self._issuer = _checked_issuer(issuer)
         # Store client_info to be set during _initialize - no dynamic registration needed
         self._fixed_client_info = OAuthClientInformationFull(
             redirect_uris=None,
@@ -80,12 +125,17 @@ async def _initialize(self) -> None:
         self.context.client_info = self._fixed_client_info
         self._initialized = True
 
+    def _select_authorization_server(self, advertised: list[str]) -> str:
+        return _preferred_authorization_server(advertised, self._issuer)
+
     async def _perform_authorization(self) -> httpx2.Request:
         """Perform client_credentials authorization."""
         return await self._exchange_token_client_credentials()
 
     async def _exchange_token_client_credentials(self) -> httpx2.Request:
         """Build token exchange request for client_credentials grant."""
+        _require_metadata_for_configured_issuer(self.context, self._issuer)
+
         token_data: dict[str, Any] = {
             "grant_type": "client_credentials",
         }
@@ -196,7 +246,10 @@ class PrivateKeyJWTOAuthProvider(OAuthClientProvider):
 
     The JWT assertion's audience MUST be the authorization server's issuer identifier
     (per RFC 7523bis security updates). The `assertion_provider` callback receives
-    this audience value and must return a JWT with that audience.
+    this audience value and must return a JWT with that audience. Pass `issuer` to name
+    the authorization server this client is registered with: an assertion is then only
+    minted once metadata for that issuer has been discovered, and token requests are only
+    built from that metadata.
 
     **Option 1: Pre-built JWT via Workload Identity Federation**
 
@@ -256,6 +309,7 @@ def __init__(
         client_id: str,
         assertion_provider: Callable[[str], Awaitable[str]],
         scope: str | None = None,
+        issuer: str | None = None,
     ) -> None:
         """Initialize private_key_jwt OAuth provider.
 
@@ -269,6 +323,11 @@ def __init__(
                 `static_assertion_provider()` for pre-built JWTs, or provide your own
                 callback for workload identity federation.
             scope: Optional space-separated list of scopes to request.
+            issuer: The issuer identifier of the authorization server `client_id` is
+                registered with. When set, an assertion is only minted, and token requests
+                are only built, once authorization server metadata whose `issuer` is exactly this
+                string has been discovered; otherwise the flow stops with `OAuthFlowError`.
+                When omitted, whichever authorization server discovery yields is used.
         """
         # Build minimal client_metadata for the base class
         client_metadata = OAuthClientMetadata(
@@ -279,6 +338,7 @@ def __init__(
         )
         super().__init__(server_url, client_metadata, storage, None, None)
         self._assertion_provider = assertion_provider
+        self._issuer = _checked_issuer(issuer)
         # Store client_info to be set during _initialize - no dynamic registration needed
         self._fixed_client_info = OAuthClientInformationFull(
             redirect_uris=None,
@@ -294,6 +354,9 @@ async def _initialize(self) -> None:
         self.context.client_info = self._fixed_client_info
         self._initialized = True
 
+    def _select_authorization_server(self, advertised: list[str]) -> str:
+        return _preferred_authorization_server(advertised, self._issuer)
+
     async def _perform_authorization(self) -> httpx2.Request:
         """Perform client_credentials authorization with private_key_jwt."""
         return await self._exchange_token_client_credentials()
@@ -314,6 +377,8 @@ async def _add_client_authentication_jwt(self, *, token_data: dict[str, Any]) ->
 
     async def _exchange_token_client_credentials(self) -> httpx2.Request:
         """Build token exchange request for client_credentials grant with private_key_jwt."""
+        _require_metadata_for_configured_issuer(self.context, self._issuer)
+
         token_data: dict[str, Any] = {
             "grant_type": "client_credentials",
         }