[v1.x] Deprecate constructing the pre-provisioned OAuth clients without an issuer

Backport of #3435. Omitting `issuer=` on ClientCredentialsOAuthProvider and
PrivateKeyJWTOAuthProvider keeps working and now warns at construction that
it will be required in 3.0, saying why (without it the MCP server decides
which authorization server receives the credentials) and what to pass.

Difference from main: 1.x has no MCPDeprecationWarning, so the category is
DeprecationWarning like the other deprecations on this branch, and there is
no deprecations docs page to list it on; the docstrings carry the note and
the module's examples pass `issuer=`.

Author: maxisbey

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

@@ -9,6 +9,7 @@
 """
 
 import time
+import warnings
 from collections.abc import Awaitable, Callable
 from typing import Any, Literal
 from urllib.parse import urlparse
@@ -25,7 +26,16 @@
 
 
 def _checked_issuer(issuer: str | None) -> str | None:
-    if issuer is not None and urlparse(issuer).scheme not in ("http", "https"):
+    if issuer is None:
+        warnings.warn(
+            "Omitting `issuer` is deprecated and it will be required in 3.0. Without it, the MCP server "
+            "decides which authorization server receives this client's credentials; pass "
+            "issuer=<your authorization server's issuer URL> so they are only ever sent there.",
+            DeprecationWarning,
+            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
 
@@ -99,7 +109,8 @@ def __init__(
             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
+                otherwise the flow stops with `OAuthFlowError`. Omitting it is deprecated
+                (`DeprecationWarning`) and it will be required in 3.0; until then, whichever
                 authorization server discovery yields is used.
         """
         # Build minimal client_metadata for the base class
@@ -170,6 +181,7 @@ def static_assertion_provider(token: str) -> Callable[[str], Awaitable[str]]:
             storage=my_token_storage,
             client_id="my-client-id",
             assertion_provider=static_assertion_provider(my_prebuilt_jwt),
+            issuer="https://auth.example.com",
         )
         ```
 
@@ -204,6 +216,7 @@ class SignedJWTParameters(BaseModel):
             storage=my_token_storage,
             client_id="my-client-id",
             assertion_provider=jwt_params.create_assertion_provider(),
+            issuer="https://auth.example.com",
         )
         ```
     """
@@ -269,6 +282,7 @@ async def get_workload_identity_token(audience: str) -> str:
             storage=my_token_storage,
             client_id="my-client-id",
             assertion_provider=get_workload_identity_token,
+            issuer="https://auth.example.com",
         )
         ```
 
@@ -282,6 +296,7 @@ async def get_workload_identity_token(audience: str) -> str:
             storage=my_token_storage,
             client_id="my-client-id",
             assertion_provider=static_assertion_provider(my_prebuilt_jwt),
+            issuer="https://auth.example.com",
         )
         ```
 
@@ -300,6 +315,7 @@ async def get_workload_identity_token(audience: str) -> str:
             storage=my_token_storage,
             client_id="my-client-id",
             assertion_provider=jwt_params.create_assertion_provider(),
+            issuer="https://auth.example.com",
         )
         ```
     """
@@ -329,7 +345,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.
+                Omitting it is deprecated (`DeprecationWarning`) and it will be required in
+                3.0; until then, whichever authorization server discovery yields is used.
         """
         # Build minimal client_metadata for the base class
         client_metadata = OAuthClientMetadata(
@@ -474,8 +491,6 @@ def __init__(
         timeout: float = 300.0,
         jwt_parameters: JWTParameters | None = None,
     ) -> None:
-        import warnings
-
         warnings.warn(
             "RFC7523OAuthClientProvider is deprecated. Use ClientCredentialsOAuthProvider "
             "or PrivateKeyJWTOAuthProvider instead.",

tests/client/auth/extensions/test_client_credentials.py

@@ -189,6 +189,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
@@ -209,6 +210,7 @@ async def test_init_with_scopes(self, mock_storage: MockTokenStorage):
             client_id="test-client-id",
             client_secret="test-client-secret",
             scopes="read write",
+            issuer="https://api.example.com",
         )
 
         await provider._initialize()
@@ -224,6 +226,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()
@@ -239,6 +242,7 @@ async def test_exchange_token_client_credentials(self, mock_storage: MockTokenSt
             client_id="test-client-id",
             client_secret="test-client-secret",
             scopes="read write",
+            issuer="https://api.example.com",
         )
         provider.context.oauth_metadata = OAuthMetadata(
             issuer=AnyHttpUrl("https://api.example.com"),
@@ -265,6 +269,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"),
@@ -296,6 +301,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
@@ -319,6 +325,7 @@ async def mock_assertion_provider(audience: str) -> str:
             client_id="test-client-id",
             assertion_provider=mock_assertion_provider,
             scopes="read write",
+            issuer="https://auth.example.com",
         )
         provider.context.oauth_metadata = OAuthMetadata(
             issuer=AnyHttpUrl("https://auth.example.com"),
@@ -350,6 +357,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"),
@@ -541,6 +549,65 @@ 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_is_deprecated(mock_storage: MockTokenStorage, kind: str) -> None:
+    """SDK-defined: leaving `issuer` out is allowed but deprecated, and the provider says so at
+    construction."""
+
+    async def assertion_provider(audience: str) -> str:
+        raise NotImplementedError
+
+    with pytest.warns(DeprecationWarning) 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__
+    assert str(warning.message) == (
+        "Omitting `issuer` is deprecated and it will be required in 3.0. Without it, the MCP server "
+        "decides which authorization server receives this client's credentials; pass "
+        "issuer=<your authorization server's issuer URL> so they are only ever sent there."
+    )
+
+
+@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(DeprecationWarning, match="Omitting `issuer` is deprecated"):
+        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(httpx.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."""