modelcontextprotocol
diff --git a/docs/client/oauth-clients.md b/docs/client/oauth-clients.md
@@ -76,7 +76,7 @@ Look at `main()`. The provider goes on the **httpx2 client**, the httpx2 client
 
 The first time `Client` sends a request, the server answers `401`. The provider takes over:
 
-1. **Discovery.** It reads the `WWW-Authenticate` header, fetches the server's Protected Resource Metadata from `/.well-known/oauth-protected-resource`, learns which authorization server protects this resource, and fetches *that* server's metadata.
+1. **Discovery.** It reads the `WWW-Authenticate` header, fetches the server's Protected Resource Metadata from `/.well-known/oauth-protected-resource`, learns which authorization server protects this resource, and fetches *that* server's metadata. (An older server that publishes no resource metadata is asked for authorization server metadata at its own origin instead.) Either way the metadata must name, as its `issuer`, the server it was fetched for; anything else is refused.
 2. **Registration.** Nothing in storage? It registers you dynamically with your `OAuthClientMetadata` and stores the result.
 3. **Authorization.** It generates the PKCE pair and a `state`, builds the authorization URL, awaits your `redirect_handler`, then awaits your `callback_handler` for the code.
 4. **Exchange.** It trades the code for an `OAuthToken`, stores it, and replays your original request with `Authorization: Bearer ...`.
@@ -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)**.

diff --git a/docs/migration.md b/docs/migration.md
@@ -2499,7 +2499,9 @@ metadata's `issuer` exactly matches the authorization server URL advertised in t
 resource metadata, as required by [RFC 8414](https://datatracker.ietf.org/doc/html/rfc8414)
 section 3.3 ([SEP-2468](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2468)).
 The comparison is a simple string comparison ([RFC 3986](https://datatracker.ietf.org/doc/html/rfc3986)
-section 6.2.1), so even a trailing-slash disagreement counts as a mismatch. v1 accepted the
+section 6.2.1), so even a trailing-slash disagreement counts as a mismatch. (For an older server
+that publishes no protected resource metadata the expected value is the MCP server's own origin,
+and there a root issuer with a trailing slash is accepted too.) v1 accepted the
 metadata without checking, so a server pairing whose two values disagree authenticated fine
 under v1 and now fails the entire flow. For example, when the MCP server's protected resource
 metadata advertises
@@ -2517,7 +2519,7 @@ OAuthFlowError: Authorization server metadata issuer mismatch: https://as.exampl
 
 There is no client-side override. Fix the deployment instead: make the authorization server's
 `issuer` string-equal the URL in the protected resource metadata's `authorization_servers`
-list. See [OAuth metadata URLs no longer gain a trailing slash](#oauth-metadata-urls-no-longer-gain-a-trailing-slash)
+list (or the MCP server's origin, without protected resource metadata). See [OAuth metadata URLs no longer gain a trailing slash](#oauth-metadata-urls-no-longer-gain-a-trailing-slash)
 for how v2 preserves the exact string form of these URLs.
 
 ### OAuth client requests `offline_access` and adds `prompt=consent` when the authorization server supports it ([SEP-2207](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2207))

diff --git a/docs_src/oauth_clients/tutorial002.py b/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",
 )
 
 

diff --git a/src/mcp/client/auth/extensions/client_credentials.py b/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",
         }