[v1.x] Let pre-provisioned OAuth clients name their authorization server

Backport of the second half of #3398.

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 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.

Differences from the main change: the keyword follows `scopes` (the 1.x
name); there is no docs page for these providers on 1.x, so the
docstrings carry the description.

Author: maxisbey

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

@@ -11,21 +11,58 @@
 import time
 from collections.abc import Awaitable, Callable
 from typing import Any, Literal
+from urllib.parse import urlparse
 from uuid import uuid4
 
 import httpx
 import jwt
 from pydantic import BaseModel, Field
 
 from mcp.client.auth import OAuthClientProvider, OAuthFlowError, OAuthTokenError, 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
@@ -34,6 +71,7 @@ class ClientCredentialsOAuthProvider(OAuthClientProvider):
             storage=my_token_storage,
             client_id="my-client-id",
             client_secret="my-client-secret",
+            issuer="https://auth.example.com",
         )
         ```
     """
@@ -46,6 +84,7 @@ def __init__(
         client_secret: str,
         token_endpoint_auth_method: Literal["client_secret_basic", "client_secret_post"] = "client_secret_basic",
         scopes: str | None = None,
+        issuer: str | None = None,
     ) -> None:
         """Initialize client_credentials OAuth provider.
 
@@ -57,6 +96,11 @@ def __init__(
             token_endpoint_auth_method: Authentication method for token endpoint.
                 Either "client_secret_basic" (default) or "client_secret_post".
             scopes: 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(
@@ -66,6 +110,7 @@ def __init__(
             scope=scopes,
         )
         super().__init__(server_url, client_metadata, storage, None, None, 300.0)
+        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,
@@ -82,12 +127,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) -> httpx.Request:
         """Perform client_credentials authorization."""
         return await self._exchange_token_client_credentials()
 
     async def _exchange_token_client_credentials(self) -> httpx.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",
         }
@@ -198,7 +248,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**
 
@@ -258,6 +311,7 @@ def __init__(
         client_id: str,
         assertion_provider: Callable[[str], Awaitable[str]],
         scopes: str | None = None,
+        issuer: str | None = None,
     ) -> None:
         """Initialize private_key_jwt OAuth provider.
 
@@ -271,6 +325,11 @@ def __init__(
                 `static_assertion_provider()` for pre-built JWTs, or provide your own
                 callback for workload identity federation.
             scopes: 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(
@@ -281,6 +340,7 @@ def __init__(
         )
         super().__init__(server_url, client_metadata, storage, None, None, 300.0)
         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,
@@ -296,6 +356,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) -> httpx.Request:
         """Perform client_credentials authorization with private_key_jwt."""
         return await self._exchange_token_client_credentials()
@@ -316,6 +379,8 @@ async def _add_client_authentication_jwt(self, *, token_data: dict[str, Any]) ->
 
     async def _exchange_token_client_credentials(self) -> httpx.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 httpx
 import jwt
 import pytest
+from inline_snapshot import snapshot
 from pydantic import AnyHttpUrl, AnyUrl
 
+from mcp.client.auth import OAuthClientProvider, OAuthFlowError
 from mcp.client.auth.extensions.client_credentials import (
     ClientCredentialsOAuthProvider,
     JWTParameters,
@@ -429,3 +433,207 @@ 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[httpx.Request, httpx.Response],
+    *,
+    authorization_server: str | list[str] | None,
+    metadata: dict[str, str] | None,
+) -> httpx.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(httpx.Response(401, request=request))
+    while "/.well-known/oauth-protected-resource" in str(request.url):
+        if authorization_server is None:
+            response = httpx.Response(404, request=request)
+        else:
+            advertised = authorization_server if isinstance(authorization_server, list) else [authorization_server]
+            prm = {"resource": _SERVER_URL, "authorization_servers": advertised}
+            response = httpx.Response(200, json=prm, request=request)
+        request = await flow.asend(response)
+    while "/.well-known/" in str(request.url):
+        if metadata is None:
+            response = httpx.Response(404, request=request)
+        else:
+            response = httpx.Response(200, json=metadata, request=request)
+        request = await flow.asend(response)
+    return request
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "served_issuer", [_CONFIGURED_ISSUER, f"{_CONFIGURED_ISSUER}/"], ids=["as-configured", "root-slash"]
+)
+@pytest.mark.parametrize("kind", ["secret", "jwt"])
+async def test_provider_with_configured_issuer_exchanges_at_that_issuer(
+    mock_storage: MockTokenStorage, kind: str, served_issuer: str
+):
+    """SDK-defined: with `issuer=` set and metadata discovered for that issuer (a root issuer served with
+    its trailing slash is the same server), 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(httpx.Request("POST", _SERVER_URL))
+    metadata = {**_metadata_for(_CONFIGURED_ISSUER), "issuer": served_issuer}
+
+    token_request = await _answer_discovery(flow, authorization_server=served_issuer, metadata=metadata)
+
+    assert (token_request.method, str(token_request.url)) == ("POST", "https://auth.example.com/token")
+    # The SDK's URL type renders a root issuer with its trailing slash, which is the audience used.
+    assert audiences == ([] if kind == "secret" else ["https://auth.example.com/"])
+    await flow.aclose()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("kind", ["secret", "jwt"])
+async def test_provider_picks_its_configured_issuer_among_several_advertised_servers(
+    mock_storage: MockTokenStorage, kind: str
+):
+    """SDK-defined: when the resource lists several authorization servers, the one matching `issuer=` is
+    discovered and used even if it is not listed first."""
+    provider = _provider_with_issuer(kind, mock_storage, [])
+    flow = provider.async_auth_flow(httpx.Request("POST", _SERVER_URL))
+
+    token_request = await _answer_discovery(
+        flow,
+        authorization_server=["https://other-as.example.com", _CONFIGURED_ISSUER],
+        metadata=_metadata_for(_CONFIGURED_ISSUER),
+    )
+
+    assert provider.context.auth_server_url == f"{_CONFIGURED_ISSUER}/"
+    assert str(token_request.url) == "https://auth.example.com/token"
+    await flow.aclose()
+
+
+def test_an_issuer_that_is_not_an_http_url_is_rejected_at_construction(mock_storage: MockTokenStorage) -> None:
+    """SDK-defined: `issuer=` is the authorization server's issuer URL; anything else is a configuration
+    error on both machine-to-machine providers."""
+    with pytest.raises(ValueError) as cc_error:
+        ClientCredentialsOAuthProvider(
+            server_url=_SERVER_URL, storage=mock_storage, client_id="cid", client_secret="s", issuer="auth.example.com"
+        )
+    with pytest.raises(ValueError) as jwt_error:
+        PrivateKeyJWTOAuthProvider(
+            server_url=_SERVER_URL,
+            storage=mock_storage,
+            client_id="cid",
+            assertion_provider=static_assertion_provider("jwt"),
+            issuer="auth.example.com",
+        )
+    assert (
+        str(cc_error.value)
+        == str(jwt_error.value)
+        == snapshot("issuer must be the authorization server's http(s) issuer URL, got 'auth.example.com'")
+    )
+
+
+@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(httpx.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(httpx.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(httpx.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(httpx.Response(200, json=token, request=token_request))
+    with pytest.raises(StopAsyncIteration):
+        await flow.asend(httpx.Response(200, request=retried))
+
+    flow = provider.async_auth_flow(httpx.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(httpx.Request("POST", _SERVER_URL))
+    request = await flow.__anext__()
+    assert (str(request.url), request.headers.get("Authorization")) == (_SERVER_URL, None)
+    await flow.aclose()