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,
discovery only proceeds with that server: the protected resource has to
advertise it (with several advertised, its entry is the one used), or on
the legacy no-PRM path the resource server's origin has to be it;
anything else stops the flow with OAuthFlowError before any of that
server's metadata is fetched. Token requests are then only built from
metadata discovered for that issuer, never from the 2025-03-26 default
endpoints: the client_credentials exchange requires it, and a refresh
waits for it (OAuthContext.configured_issuer, can_refresh_token). 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. Discovery still runs as above, but the token request is 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,48 @@
 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 validate_metadata_issuer
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata
 
 
+def _configure_issuer(context: OAuthContext, issuer: str | None) -> None:
+    """Record `issuer` as the one authorization server this client works with (`OAuthContext.configured_issuer`)."""
+    if issuer is None:
+        return
+    parsed = urlparse(issuer)
+    if parsed.scheme not in ("http", "https") or not parsed.netloc:
+        raise ValueError("issuer, when given, must be the authorization server's issuer URL")
+    context.configured_issuer = issuer
+
+
+def _require_metadata_for_configured_issuer(context: OAuthContext) -> None:
+    """With an issuer configured, a token request is only built from metadata discovered for it."""
+    if context.configured_issuer is None:
+        return
+    if context.oauth_metadata is None:
+        raise OAuthFlowError(
+            f"No authorization server metadata discovered for configured issuer {context.configured_issuer}"
+        )
+    validate_metadata_issuer(context.oauth_metadata, context.configured_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: the flow then
+    only proceeds when the MCP server leads to that server, and the client_credentials token
+    request is only built from its metadata.
 
     Example:
         ```python
@@ -32,6 +59,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 +72,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 +84,12 @@ 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, discovery only proceeds with that
+                server (the resource must advertise it, or be it on the legacy path) and token
+                requests are only built from its metadata; 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 +99,7 @@ def __init__(
             scope=scope,
         )
         super().__init__(server_url, client_metadata, storage, None, None)
+        _configure_issuer(self.context, issuer)
         # Store client_info to be set during _initialize - no dynamic registration needed
         self._fixed_client_info = OAuthClientInformationFull(
             redirect_uris=None,
@@ -86,6 +122,8 @@ async def _perform_authorization(self) -> httpx2.Request:
 
     async def _exchange_token_client_credentials(self) -> httpx2.Request:
         """Build token exchange request for client_credentials grant."""
+        _require_metadata_for_configured_issuer(self.context)
+
         token_data: dict[str, Any] = {
             "grant_type": "client_credentials",
         }
@@ -196,7 +234,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 sent to the token
+    endpoint that metadata publishes.
 
     **Option 1: Pre-built JWT via Workload Identity Federation**
 
@@ -256,6 +297,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 +311,12 @@ 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, discovery only proceeds with that server (the resource
+                must advertise it, or be it on the legacy path) and an assertion is only minted
+                once its metadata 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 +327,7 @@ def __init__(
         )
         super().__init__(server_url, client_metadata, storage, None, None)
         self._assertion_provider = assertion_provider
+        _configure_issuer(self.context, issuer)
         # Store client_info to be set during _initialize - no dynamic registration needed
         self._fixed_client_info = OAuthClientInformationFull(
             redirect_uris=None,
@@ -314,6 +363,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)
+
         token_data: dict[str, Any] = {
             "grant_type": "client_credentials",
         }

src/mcp/client/auth/oauth2.py

@@ -160,6 +160,9 @@ class OAuthContext:
     protected_resource_metadata: ProtectedResourceMetadata | None = None
     oauth_metadata: OAuthMetadata | None = None
     auth_server_url: str | None = None
+    # Set by providers whose credentials were issued by one known authorization server: discovery
+    # only proceeds with that server, and token requests are only built from its metadata.
+    configured_issuer: str | None = None
     protocol_version: str | None = None
 
     # Client registration
@@ -198,7 +201,13 @@ def is_token_valid(self) -> bool:
         )
 
     def can_refresh_token(self) -> bool:
-        """Check if token can be refreshed."""
+        """Check if token can be refreshed.
+
+        With a configured issuer the refresh request is only built from that issuer's metadata, so
+        none is attempted until it has been discovered (the request then re-authenticates instead).
+        """
+        if self.configured_issuer is not None and self.oauth_metadata is None:
+            return False
         return bool(self.current_tokens and self.current_tokens.refresh_token and self.client_info)
 
     def clear_tokens(self) -> None:
@@ -587,12 +596,34 @@ async def _validate_resource_match(self, prm: ProtectedResourceMetadata) -> None
             raise OAuthFlowError(f"Protected resource {prm_resource} does not match expected {default_resource}")
 
     def _select_authorization_server(self, prm: ProtectedResourceMetadata | None) -> str | None:
-        """Pick the authorization server for this pass from the protected resource metadata: the
-        first advertised server, or None on the legacy no-PRM path (the resource server's origin is
-        then used)."""
+        """Pick the authorization server for this pass from the protected resource metadata.
+
+        That is the first advertised server, or None on the legacy no-PRM path (the resource
+        server's origin is then used). When the provider is configured for one issuer, the
+        advertised server (or that origin) has to be it.
+
+        Raises:
+            OAuthFlowError: If an issuer is configured and the resource does not lead to it.
+        """
+        configured = self.context.configured_issuer
         if prm is None:
+            origin = self.context.get_authorization_base_url(self.context.server_url)
+            if configured is not None and not issuers_equal(origin, configured):
+                raise OAuthFlowError(
+                    f"No protected resource metadata, and the resource origin {origin} is not the configured "
+                    f"issuer {configured}"
+                )
             return None
-        return str(prm.authorization_servers[0])
+        advertised = [str(url) for url in prm.authorization_servers]
+        if configured is None:
+            return advertised[0]
+        for candidate in advertised:
+            if issuers_equal(candidate, configured):
+                return candidate
+        raise OAuthFlowError(
+            f"Protected resource advertises authorization servers {advertised}; none is the configured issuer "
+            f"{configured}"
+        )
 
     def _discard_credentials_bound_elsewhere(self, expected_issuer: str) -> None:
         """SEP-2352: stored credentials are bound to the issuer that registered them.