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. 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, spelled exactly as that server's metadata states it (for an authorization server built with this SDK that is the URL with a trailing slash). 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

@@ -16,14 +16,37 @@
 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 _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
+    try:
+        if context.oauth_metadata is None:
+            raise OAuthFlowError(f"No authorization server metadata discovered for configured issuer {issuer}")
+        validate_metadata_issuer(context.oauth_metadata, issuer)
+    except OAuthFlowError:
+        context.oauth_metadata = None
+        context.clear_tokens()
+        raise
+
+
 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 +55,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 +68,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 +80,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 +94,7 @@ def __init__(
             scope=scope,
         )
         super().__init__(server_url, client_metadata, storage, None, None)
+        self._issuer = issuer
         # Store client_info to be set during _initialize - no dynamic registration needed
         self._fixed_client_info = OAuthClientInformationFull(
             redirect_uris=None,
@@ -86,6 +117,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, self._issuer)
+
         token_data: dict[str, Any] = {
             "grant_type": "client_credentials",
         }
@@ -196,7 +229,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 +292,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 +306,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 +321,7 @@ def __init__(
         )
         super().__init__(server_url, client_metadata, storage, None, None)
         self._assertion_provider = assertion_provider
+        self._issuer = issuer
         # Store client_info to be set during _initialize - no dynamic registration needed
         self._fixed_client_info = OAuthClientInformationFull(
             redirect_uris=None,
@@ -314,6 +357,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",
         }

tests/client/auth/extensions/test_client_credentials.py

@@ -1,9 +1,13 @@
 import urllib.parse
+from collections.abc import AsyncGenerator
 
+import httpx2
 import jwt
 import pytest
+from inline_snapshot import snapshot
 from pydantic import AnyHttpUrl
 
+from mcp.client.auth import OAuthClientProvider, OAuthFlowError
 from mcp.client.auth.extensions.client_credentials import (
     ClientCredentialsOAuthProvider,
     PrivateKeyJWTOAuthProvider,
@@ -27,7 +31,7 @@ def __init__(self):
     async def get_tokens(self) -> OAuthToken | None:
         return self._tokens
 
-    async def set_tokens(self, tokens: OAuthToken) -> None:  # pragma: no cover
+    async def set_tokens(self, tokens: OAuthToken) -> None:
         self._tokens = tokens
 
     async def get_client_info(self) -> OAuthClientInformationFull | None:  # pragma: no cover
@@ -325,3 +329,157 @@ async def test_returns_static_token(self):
 
         assert result1 == token
         assert result2 == token
+
+
+_SERVER_URL = "https://api.example.com/v1/mcp"
+_CONFIGURED_ISSUER = "https://auth.example.com"
+
+
+def _metadata_for(issuer: str) -> dict[str, str]:
+    return {"issuer": issuer, "authorization_endpoint": f"{issuer}/authorize", "token_endpoint": f"{issuer}/token"}
+
+
+def _provider_with_issuer(kind: str, storage: MockTokenStorage, audiences: list[str]) -> OAuthClientProvider:
+    """A ClientCredentials ("secret") or PrivateKeyJWT ("jwt") provider configured for _CONFIGURED_ISSUER;
+    `audiences` records every audience an assertion is minted for."""
+    if kind == "secret":
+        return ClientCredentialsOAuthProvider(
+            server_url=_SERVER_URL, storage=storage, client_id="cid", client_secret="csecret", issuer=_CONFIGURED_ISSUER
+        )
+
+    async def assertion_provider(audience: str) -> str:
+        audiences.append(audience)
+        return "signed-assertion"
+
+    return PrivateKeyJWTOAuthProvider(
+        server_url=_SERVER_URL,
+        storage=storage,
+        client_id="cid",
+        assertion_provider=assertion_provider,
+        issuer=_CONFIGURED_ISSUER,
+    )
+
+
+async def _answer_discovery(
+    flow: AsyncGenerator[httpx2.Request, httpx2.Response],
+    *,
+    authorization_server: str | None,
+    metadata: dict[str, str] | None,
+) -> httpx2.Request:
+    """Answer the provider's first request with a 401 and its discovery requests as described;
+    return the request it builds once discovery is over.
+
+    `authorization_server` is what protected-resource metadata advertises (None: no PRM is
+    served); `metadata` is the authorization server metadata document (None: every well-known
+    404s).
+    """
+    request = await flow.__anext__()
+    request = await flow.asend(httpx2.Response(401, request=request))
+    while "/.well-known/oauth-protected-resource" in str(request.url):
+        if authorization_server is None:
+            response = httpx2.Response(404, request=request)
+        else:
+            prm = {"resource": _SERVER_URL, "authorization_servers": [authorization_server]}
+            response = httpx2.Response(200, json=prm, request=request)
+        request = await flow.asend(response)
+    while "/.well-known/" in str(request.url):
+        if metadata is None:
+            response = httpx2.Response(404, request=request)
+        else:
+            response = httpx2.Response(200, json=metadata, request=request)
+        request = await flow.asend(response)
+    return request
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("kind", ["secret", "jwt"])
+async def test_provider_with_configured_issuer_exchanges_at_that_issuer(mock_storage: MockTokenStorage, kind: str):
+    """SDK-defined: with `issuer=` set and metadata discovered for that issuer, the token request goes
+    to its token endpoint (positive control for the refusals below)."""
+    audiences: list[str] = []
+    provider = _provider_with_issuer(kind, mock_storage, audiences)
+    flow = provider.async_auth_flow(httpx2.Request("POST", _SERVER_URL))
+
+    token_request = await _answer_discovery(
+        flow, authorization_server=_CONFIGURED_ISSUER, metadata=_metadata_for(_CONFIGURED_ISSUER)
+    )
+
+    assert (token_request.method, str(token_request.url)) == ("POST", "https://auth.example.com/token")
+    assert audiences == ([] if kind == "secret" else [_CONFIGURED_ISSUER])
+    await flow.aclose()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("kind", ["secret", "jwt"])
+async def test_provider_refuses_metadata_for_a_different_issuer(mock_storage: MockTokenStorage, kind: str):
+    """SDK-defined: when discovery ends at an authorization server other than the configured `issuer`,
+    no token request is built and no assertion is minted."""
+    audiences: list[str] = []
+    provider = _provider_with_issuer(kind, mock_storage, audiences)
+    flow = provider.async_auth_flow(httpx2.Request("POST", _SERVER_URL))
+
+    with pytest.raises(OAuthFlowError) as exc_info:
+        await _answer_discovery(
+            flow,
+            authorization_server="https://other-as.example.com",
+            metadata=_metadata_for("https://other-as.example.com"),
+        )
+
+    assert str(exc_info.value) == snapshot(
+        "Authorization server metadata issuer mismatch: https://other-as.example.com != https://auth.example.com"
+    )
+    assert audiences == []
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("kind", ["secret", "jwt"])
+async def test_provider_refuses_to_exchange_without_metadata_when_issuer_configured(
+    mock_storage: MockTokenStorage, kind: str
+):
+    """SDK-defined: with `issuer=` set, the 2025-03-26 default `/token` on the resource origin is not
+    used when no authorization server metadata could be discovered."""
+    audiences: list[str] = []
+    provider = _provider_with_issuer(kind, mock_storage, audiences)
+    flow = provider.async_auth_flow(httpx2.Request("POST", _SERVER_URL))
+
+    with pytest.raises(OAuthFlowError) as exc_info:
+        await _answer_discovery(flow, authorization_server=None, metadata=None)
+
+    assert str(exc_info.value) == snapshot(
+        "No authorization server metadata discovered for configured issuer https://auth.example.com"
+    )
+    assert audiences == []
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("kind", ["secret", "jwt"])
+async def test_a_refused_authorization_server_is_forgotten_so_the_next_request_rediscovers(
+    mock_storage: MockTokenStorage, kind: str
+):
+    """SDK-defined: when the exchange is refused because discovery ended somewhere other than the
+    configured issuer, the refused metadata and any token held are dropped; the next request goes out
+    unauthenticated and discovery starts again, rather than a refresh being built from what was refused."""
+    provider = _provider_with_issuer(kind, mock_storage, [])
+    flow = provider.async_auth_flow(httpx2.Request("POST", _SERVER_URL))
+    token_request = await _answer_discovery(
+        flow, authorization_server=_CONFIGURED_ISSUER, metadata=_metadata_for(_CONFIGURED_ISSUER)
+    )
+    token = {"access_token": "first", "token_type": "Bearer", "expires_in": 3600, "refresh_token": "rt"}
+    retried = await flow.asend(httpx2.Response(200, json=token, request=token_request))
+    with pytest.raises(StopAsyncIteration):
+        await flow.asend(httpx2.Response(200, request=retried))
+
+    flow = provider.async_auth_flow(httpx2.Request("POST", _SERVER_URL))
+    with pytest.raises(OAuthFlowError):
+        await _answer_discovery(
+            flow,
+            authorization_server="https://other-as.example.com",
+            metadata=_metadata_for("https://other-as.example.com"),
+        )
+    assert provider.context.oauth_metadata is None
+    assert provider.context.current_tokens is None
+
+    flow = provider.async_auth_flow(httpx2.Request("POST", _SERVER_URL))
+    request = await flow.__anext__()
+    assert (str(request.url), request.headers.get("Authorization")) == (_SERVER_URL, None)
+    await flow.aclose()

tests/docs_src/test_oauth_clients.py

@@ -9,6 +9,7 @@
 from docs_src.oauth_clients import tutorial001, tutorial002
 from mcp.client.auth import OAuthClientProvider, OAuthFlowError, OAuthRegistrationError, OAuthTokenError, TokenStorage
 from mcp.client.auth.extensions.client_credentials import (
+    ClientCredentialsOAuthProvider,
     PrivateKeyJWTOAuthProvider,
     static_assertion_provider,
 )
@@ -80,6 +81,14 @@ async def test_client_credentials_provider_builds_its_own_metadata() -> None:
     assert metadata.scope == "user"
 
 
+@pytest.mark.parametrize("provider_class", [ClientCredentialsOAuthProvider, PrivateKeyJWTOAuthProvider])
+async def test_issuer_is_an_optional_keyword_on_both_machine_to_machine_providers(provider_class: type) -> None:
+    """tutorial002 passes `issuer=`; the page says leaving it out is allowed, on either provider."""
+    issuer = inspect.signature(provider_class.__init__).parameters["issuer"]
+    assert issuer.kind is inspect.Parameter.POSITIONAL_OR_KEYWORD
+    assert issuer.default is None
+
+
 async def test_the_two_remaining_keyword_arguments_have_defaults() -> None:
     """The page names `client_metadata_url` and `validate_resource_url` as the remainder."""
     parameters = inspect.signature(OAuthClientProvider.__init__).parameters