Warn when a pre-provisioned OAuth client is created without an issuer

ClientCredentialsOAuthProvider and PrivateKeyJWTOAuthProvider send fixed
credentials to whichever authorization server discovery yields unless
`issuer=` names the one they belong to. Leaving it out stays allowed, but
the provider now says so at construction with a UserWarning that names the
server URL and the keyword to pass, so the choice is visible rather than
silent. Nothing else changes: with `issuer=` set there is no warning, and a
value that is not an http(s) URL is still a ValueError.

The example story and the interaction tests pass `issuer=` (their
authorization server is known); the extension tests that exercise the
no-issuer path opt in to the warning explicitly.

Author: maxisbey

docs/client/oauth-clients.md

@@ -112,7 +112,7 @@ A nightly job, a CI step, another service. There is no browser and nobody to cli
 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.
+* `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, and says so with a `UserWarning` when it is constructed.
 * `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`.
 

examples/stories/oauth_client_credentials/client.py

@@ -8,13 +8,13 @@
 
 # MCP_URL pins the resource to :8000, and the server side builds its PRM/AS metadata from
 # the same constant — run the server on 8000 or the discovery chain points at the wrong origin.
-from stories._shared.auth import MCP_URL, InMemoryTokenStorage
+from stories._shared.auth import BASE_URL, MCP_URL, InMemoryTokenStorage
 
 from .server import DEMO_CLIENT_ID, DEMO_CLIENT_SECRET, DEMO_SCOPE
 
 
 def build_auth(_http: httpx2.AsyncClient) -> httpx2.Auth:
-    """The ``httpx2.Auth`` for the ``client_credentials`` grant — five lines of provider config.
+    """The ``httpx2.Auth`` for the ``client_credentials`` grant — six lines of provider config.
 
     The SDK then handles 401 → RFC 9728 PRM → RFC 8414 AS-metadata discovery → token POST →
     Bearer attachment automatically. ``Client(url)`` has no ``auth=`` passthrough yet, so the
@@ -27,6 +27,7 @@ def build_auth(_http: httpx2.AsyncClient) -> httpx2.Auth:
         client_id=DEMO_CLIENT_ID,
         client_secret=DEMO_CLIENT_SECRET,
         scope=DEMO_SCOPE,
+        issuer=BASE_URL,
     )
 
 

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

@@ -7,6 +7,7 @@
 """
 
 import time
+import warnings
 from collections.abc import Awaitable, Callable
 from typing import Any, Literal
 from urllib.parse import urlparse
@@ -22,8 +23,16 @@
 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"):
+def _checked_issuer(issuer: str | None, provider: str, server_url: str) -> str | None:
+    if issuer is None:
+        warnings.warn(
+            f"{provider} created without `issuer`: the client credentials will be sent to whichever "
+            f"authorization server {server_url} advertises. Pass issuer=<the issuer URL your authorization "
+            "server's metadata reports> so that token requests are only ever built for that server.",
+            stacklevel=3,
+        )
+        return None
+    if 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
 
@@ -98,7 +107,7 @@ def __init__(
                 `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.
+                authorization server discovery yields is used, and a `UserWarning` says so.
         """
         # Build minimal client_metadata for the base class
         client_metadata = OAuthClientMetadata(
@@ -108,7 +117,7 @@ def __init__(
             scope=scope,
         )
         super().__init__(server_url, client_metadata, storage, None, None)
-        self._issuer = _checked_issuer(issuer)
+        self._issuer = _checked_issuer(issuer, type(self).__name__, server_url)
         # Store client_info to be set during _initialize - no dynamic registration needed
         self._fixed_client_info = OAuthClientInformationFull(
             redirect_uris=None,
@@ -327,7 +336,8 @@ def __init__(
                 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.
+                When omitted, whichever authorization server discovery yields is used, and a
+                `UserWarning` says so.
         """
         # Build minimal client_metadata for the base class
         client_metadata = OAuthClientMetadata(
@@ -338,7 +348,7 @@ def __init__(
         )
         super().__init__(server_url, client_metadata, storage, None, None)
         self._assertion_provider = assertion_provider
-        self._issuer = _checked_issuer(issuer)
+        self._issuer = _checked_issuer(issuer, type(self).__name__, server_url)
         # Store client_info to be set during _initialize - no dynamic registration needed
         self._fixed_client_info = OAuthClientInformationFull(
             redirect_uris=None,

tests/client/auth/extensions/test_client_credentials.py

@@ -57,6 +57,7 @@ async def test_init_sets_client_info(self, mock_storage: MockTokenStorage):
             storage=mock_storage,
             client_id="test-client-id",
             client_secret="test-client-secret",
+            issuer="https://api.example.com",
         )
 
         # client_info is set during _initialize
@@ -77,6 +78,7 @@ async def test_init_with_scopes(self, mock_storage: MockTokenStorage):
             client_id="test-client-id",
             client_secret="test-client-secret",
             scope="read write",
+            issuer="https://api.example.com",
         )
 
         await provider._initialize()
@@ -92,6 +94,7 @@ async def test_init_with_client_secret_post(self, mock_storage: MockTokenStorage
             client_id="test-client-id",
             client_secret="test-client-secret",
             token_endpoint_auth_method="client_secret_post",
+            issuer="https://api.example.com",
         )
 
         await provider._initialize()
@@ -107,6 +110,7 @@ async def test_exchange_token_client_credentials(self, mock_storage: MockTokenSt
             client_id="test-client-id",
             client_secret="test-client-secret",
             scope="read write",
+            issuer="https://api.example.com",
         )
         provider.context.oauth_metadata = OAuthMetadata(
             issuer=AnyHttpUrl("https://api.example.com"),
@@ -135,6 +139,7 @@ async def test_exchange_token_client_secret_post_includes_client_id(self, mock_s
             client_secret="test-client-secret",
             token_endpoint_auth_method="client_secret_post",
             scope="read write",
+            issuer="https://api.example.com",
         )
         await provider._initialize()
         provider.context.oauth_metadata = OAuthMetadata(
@@ -161,6 +166,7 @@ async def test_exchange_token_without_scopes(self, mock_storage: MockTokenStorag
             storage=mock_storage,
             client_id="test-client-id",
             client_secret="test-client-secret",
+            issuer="https://api.example.com",
         )
         provider.context.oauth_metadata = OAuthMetadata(
             issuer=AnyHttpUrl("https://api.example.com"),
@@ -192,6 +198,7 @@ async def mock_assertion_provider(audience: str) -> str:  # pragma: no cover
             storage=mock_storage,
             client_id="test-client-id",
             assertion_provider=mock_assertion_provider,
+            issuer="https://api.example.com",
         )
 
         # client_info is set during _initialize
@@ -215,6 +222,7 @@ async def mock_assertion_provider(audience: str) -> str:
             client_id="test-client-id",
             assertion_provider=mock_assertion_provider,
             scope="read write",
+            issuer="https://auth.example.com",
         )
         provider.context.oauth_metadata = OAuthMetadata(
             issuer=AnyHttpUrl("https://auth.example.com"),
@@ -246,6 +254,7 @@ async def mock_assertion_provider(audience: str) -> str:
             storage=mock_storage,
             client_id="test-client-id",
             assertion_provider=mock_assertion_provider,
+            issuer="https://auth.example.com",
         )
         provider.context.oauth_metadata = OAuthMetadata(
             issuer=AnyHttpUrl("https://auth.example.com"),
@@ -436,6 +445,68 @@ async def test_provider_picks_its_configured_issuer_among_several_advertised_ser
     await flow.aclose()
 
 
+@pytest.mark.parametrize("kind", ["secret", "jwt"])
+def test_constructing_without_issuer_warns_where_the_credentials_will_go(
+    mock_storage: MockTokenStorage, kind: str
+) -> None:
+    """SDK-defined: leaving `issuer` out is allowed, and the provider says at construction that
+    token requests will follow whichever authorization server the MCP server advertises."""
+
+    async def assertion_provider(audience: str) -> str:
+        raise NotImplementedError
+
+    with pytest.warns(UserWarning) as recorded:
+        if kind == "secret":
+            ClientCredentialsOAuthProvider(
+                server_url=_SERVER_URL, storage=mock_storage, client_id="c", client_secret="s"
+            )
+        else:
+            PrivateKeyJWTOAuthProvider(
+                server_url=_SERVER_URL, storage=mock_storage, client_id="c", assertion_provider=assertion_provider
+            )
+
+    [warning] = recorded
+    assert warning.filename == __file__
+    provider = "ClientCredentialsOAuthProvider" if kind == "secret" else "PrivateKeyJWTOAuthProvider"
+    assert str(warning.message) == (
+        f"{provider} created without `issuer`: the client credentials will be sent to whichever "
+        "authorization server https://api.example.com/v1/mcp advertises. Pass issuer=<the issuer URL your "
+        "authorization server's metadata reports> so that token requests are only ever built for that server."
+    )
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("kind", ["secret", "jwt"])
+async def test_without_issuer_the_exchange_follows_whichever_server_was_discovered(
+    mock_storage: MockTokenStorage, kind: str
+) -> None:
+    """SDK-defined: with no `issuer` configured the token request is built from whatever metadata
+    discovery produced, as before."""
+
+    async def assertion_provider(audience: str) -> str:
+        return "jwt"
+
+    with pytest.warns(UserWarning, match="created without `issuer`"):
+        if kind == "secret":
+            provider: OAuthClientProvider = ClientCredentialsOAuthProvider(
+                server_url=_SERVER_URL, storage=mock_storage, client_id="c", client_secret="s"
+            )
+        else:
+            provider = PrivateKeyJWTOAuthProvider(
+                server_url=_SERVER_URL, storage=mock_storage, client_id="c", assertion_provider=assertion_provider
+            )
+    flow = provider.async_auth_flow(httpx2.Request("POST", _SERVER_URL))
+
+    token_request = await _answer_discovery(
+        flow,
+        authorization_server="https://elsewhere.example.com",
+        metadata=_metadata_for("https://elsewhere.example.com"),
+    )
+
+    assert (token_request.method, str(token_request.url)) == ("POST", "https://elsewhere.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."""

tests/docs_src/test_oauth_clients.py

@@ -105,6 +105,7 @@ async def test_the_one_more_provider_is_private_key_jwt() -> None:
         storage=tutorial002.InMemoryTokenStorage(),
         client_id="reporting-agent",
         assertion_provider=static_assertion_provider("a.prebuilt.jwt"),
+        issuer="http://localhost:9000",
     )
     assert isinstance(provider, OAuthClientProvider)
     assert isinstance(provider, httpx2.Auth)

tests/interaction/auth/test_lifecycle.py

@@ -373,6 +373,7 @@ async def test_client_credentials_provider_obtains_a_token_without_an_authorize_
         client_id="m2m-client",
         client_secret="m2m-secret",
         scope="mcp",
+        issuer=BASE_URL,
     )
 
     with anyio.fail_after(5):
@@ -424,6 +425,7 @@ async def assertion_provider(audience: str) -> str:
         client_id="m2m-jwt-client",
         assertion_provider=assertion_provider,
         scope="mcp",
+        issuer=BASE_URL,
     )
 
     with anyio.fail_after(5):