Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions docs/migration.md
Original file line number Diff line number Diff line change
Expand Up @@ -1920,6 +1920,32 @@ The WebSocket transport has been removed: `mcp.client.websocket.websocket_client

## OAuth and server auth

### `RFC7523OAuthClientProvider` and `JWTParameters` removed

`RFC7523OAuthClientProvider` (deprecated since 1.23.0) and its `JWTParameters` model have been
removed from `mcp.client.auth.extensions.client_credentials`. The provider implemented the
[RFC 7523](https://datatracker.ietf.org/doc/html/rfc7523) §2.1 `jwt-bearer` *authorization grant*
with an SDK-minted or prebuilt JWT, which no MCP auth extension specifies. Replace it with the
purpose-built provider for the flow you actually run:

- Machine-to-machine with a client secret
([`io.modelcontextprotocol/oauth-client-credentials`](https://modelcontextprotocol.io/extensions/auth/oauth-client-credentials)):
`ClientCredentialsOAuthProvider(server_url, storage, client_id, client_secret)`.
- Machine-to-machine authenticating with a JWT instead of a secret (same extension, RFC 7523 §2.2
`private_key_jwt` client authentication on the `client_credentials` grant, which is the mode the
extension actually specifies for JWTs): `PrivateKeyJWTOAuthProvider(server_url, storage,
client_id, assertion_provider)`. Build the assertion with `SignedJWTParameters(issuer, subject,
signing_key).create_assertion_provider()` (replaces `JWTParameters` signing fields), or wrap a
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
Outdated
prebuilt JWT with `static_assertion_provider(token)` (replaces `JWTParameters(assertion=...)`).
- Presenting an enterprise ID-JAG under the `jwt-bearer` grant
([SEP-990](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/990)):
`IdentityAssertionOAuthProvider` in `mcp.client.auth.extensions.identity_assertion`.

The provider's third mode — the interactive `authorization_code` flow with `private_key_jwt`
client authentication on the token exchange — has no replacement and is intentionally dropped; it
was never exercised by the test suite and no MCP auth extension specifies it. If you depended on
it, open an issue describing the deployment.

### OAuth metadata URLs no longer gain a trailing slash

`OAuthMetadata`, `ProtectedResourceMetadata`, and `OAuthClientMetadata` now set
Expand Down
157 changes: 2 additions & 155 deletions src/mcp/client/auth/extensions/client_credentials.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,9 @@
- ClientCredentialsOAuthProvider: For client_credentials with client_id + client_secret
- PrivateKeyJWTOAuthProvider: For client_credentials with private_key_jwt authentication
(typically using a pre-built JWT from workload identity federation)
- RFC7523OAuthClientProvider: For jwt-bearer grant (RFC 7523 Section 2.1)
"""

import time
import warnings
from collections.abc import Awaitable, Callable
from typing import Any, Literal
from uuid import uuid4
Expand All @@ -17,9 +15,8 @@
import jwt
from pydantic import BaseModel, Field

from mcp.client.auth import OAuthClientProvider, OAuthFlowError, OAuthTokenError, TokenStorage
from mcp.shared.auth import AuthorizationCodeResult, OAuthClientInformationFull, OAuthClientMetadata
from mcp.shared.exceptions import MCPDeprecationWarning
from mcp.client.auth import OAuthClientProvider, OAuthFlowError, TokenStorage
from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata

Check warning on line 19 in src/mcp/client/auth/extensions/client_credentials.py

View check run for this annotation

Claude / Claude Code Review

Dead token_data hook parameter (with mutable default) left behind in OAuthClientProvider._exchange_token_authorization_code

Removing `RFC7523OAuthClientProvider` leaves behind the `token_data` hook parameter it was the sole consumer of: `OAuthClientProvider._exchange_token_authorization_code` in `src/mcp/client/auth/oauth2.py:386-387` still declares `token_data: dict[str, Any] | None = {}` — a mutable default (ruff B006 shape) with a mismatched `| None` annotation — even though no remaining caller passes it. Since the migration notes say the hybrid `authorization_code`+`private_key_jwt` mode is intentionally dropped
Comment thread
claude[bot] marked this conversation as resolved.


class ClientCredentialsOAuthProvider(OAuthClientProvider):
Expand Down Expand Up @@ -334,153 +331,3 @@

token_url = self._get_token_endpoint()
return httpx2.Request("POST", token_url, data=token_data, headers=headers)


class JWTParameters(BaseModel):
"""JWT parameters."""

assertion: str | None = Field(
default=None,
description="JWT assertion for JWT authentication. "
"Will be used instead of generating a new assertion if provided.",
)

issuer: str | None = Field(default=None, description="Issuer for JWT assertions.")
subject: str | None = Field(default=None, description="Subject identifier for JWT assertions.")
audience: str | None = Field(default=None, description="Audience for JWT assertions.")
claims: dict[str, Any] | None = Field(default=None, description="Additional claims for JWT assertions.")
jwt_signing_algorithm: str | None = Field(default="RS256", description="Algorithm for signing JWT assertions.")
jwt_signing_key: str | None = Field(default=None, description="Private key for JWT signing.")
jwt_lifetime_seconds: int = Field(default=300, description="Lifetime of generated JWT in seconds.")

def to_assertion(self, with_audience_fallback: str | None = None) -> str:
if self.assertion is not None:
# Prebuilt JWT (e.g. acquired out-of-band)
assertion = self.assertion
else:
if not self.jwt_signing_key:
raise OAuthFlowError("Missing signing key for JWT bearer grant") # pragma: no cover
if not self.issuer:
raise OAuthFlowError("Missing issuer for JWT bearer grant") # pragma: no cover
if not self.subject:
raise OAuthFlowError("Missing subject for JWT bearer grant") # pragma: no cover

audience = self.audience if self.audience else with_audience_fallback
if not audience:
raise OAuthFlowError("Missing audience for JWT bearer grant") # pragma: no cover

now = int(time.time())
claims: dict[str, Any] = {
"iss": self.issuer,
"sub": self.subject,
"aud": audience,
"exp": now + self.jwt_lifetime_seconds,
"iat": now,
"jti": str(uuid4()),
}
claims.update(self.claims or {})

assertion = jwt.encode(
claims,
self.jwt_signing_key,
algorithm=self.jwt_signing_algorithm or "RS256",
)
return assertion


class RFC7523OAuthClientProvider(OAuthClientProvider):
"""OAuth client provider for RFC 7523 jwt-bearer grant.

.. deprecated::
Use :class:`ClientCredentialsOAuthProvider` for client_credentials with
client_id + client_secret, or :class:`PrivateKeyJWTOAuthProvider` for
client_credentials with private_key_jwt authentication instead.

This provider supports the jwt-bearer authorization grant (RFC 7523 Section 2.1)
where the JWT itself is the authorization grant.
"""

def __init__(
self,
server_url: str,
client_metadata: OAuthClientMetadata,
storage: TokenStorage,
redirect_handler: Callable[[str], Awaitable[None]] | None = None,
callback_handler: Callable[[], Awaitable[AuthorizationCodeResult]] | None = None,
timeout: float = 300.0,
jwt_parameters: JWTParameters | None = None,
) -> None:
warnings.warn(
"RFC7523OAuthClientProvider is deprecated. Use ClientCredentialsOAuthProvider "
"or PrivateKeyJWTOAuthProvider instead.",
MCPDeprecationWarning,
stacklevel=2,
)
super().__init__(server_url, client_metadata, storage, redirect_handler, callback_handler, timeout)
self.jwt_parameters = jwt_parameters

async def _exchange_token_authorization_code(
self, auth_code: str, code_verifier: str, *, token_data: dict[str, Any] | None = None
) -> httpx2.Request: # pragma: no cover
"""Build token exchange request for authorization_code flow."""
token_data = token_data or {}
if self.context.client_metadata.token_endpoint_auth_method == "private_key_jwt":
self._add_client_authentication_jwt(token_data=token_data)
return await super()._exchange_token_authorization_code(auth_code, code_verifier, token_data=token_data)

async def _perform_authorization(self) -> httpx2.Request: # pragma: no cover
"""Perform the authorization flow."""
if "urn:ietf:params:oauth:grant-type:jwt-bearer" in self.context.client_metadata.grant_types:
token_request = await self._exchange_token_jwt_bearer()
return token_request
else:
return await super()._perform_authorization()

def _add_client_authentication_jwt(self, *, token_data: dict[str, Any]): # pragma: no cover
"""Add JWT assertion for client authentication to token endpoint parameters."""
if not self.jwt_parameters:
raise OAuthTokenError("Missing JWT parameters for private_key_jwt flow")
if not self.context.oauth_metadata:
raise OAuthTokenError("Missing OAuth metadata for private_key_jwt flow")

# We need to set the audience to the issuer identifier of the authorization server
# https://datatracker.ietf.org/doc/html/draft-ietf-oauth-rfc7523bis-01#name-updates-to-rfc-7523
issuer = str(self.context.oauth_metadata.issuer)
assertion = self.jwt_parameters.to_assertion(with_audience_fallback=issuer)

# When using private_key_jwt, in a client_credentials flow, we use RFC 7523 Section 2.2
token_data["client_assertion"] = assertion
token_data["client_assertion_type"] = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
# We need to set the audience to the resource server, the audience is different from the one in claims
# it represents the resource server that will validate the token
token_data["audience"] = self.context.get_resource_url()

async def _exchange_token_jwt_bearer(self) -> httpx2.Request:
"""Build token exchange request for JWT bearer grant."""
if not self.context.client_info:
raise OAuthFlowError("Missing client info") # pragma: no cover
if not self.jwt_parameters:
raise OAuthFlowError("Missing JWT parameters") # pragma: no cover
if not self.context.oauth_metadata:
raise OAuthTokenError("Missing OAuth metadata") # pragma: no cover

# We need to set the audience to the issuer identifier of the authorization server
# https://datatracker.ietf.org/doc/html/draft-ietf-oauth-rfc7523bis-01#name-updates-to-rfc-7523
issuer = str(self.context.oauth_metadata.issuer)
assertion = self.jwt_parameters.to_assertion(with_audience_fallback=issuer)

token_data = {
"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
"assertion": assertion,
}

if self.context.should_include_resource_param(self.context.protocol_version): # pragma: no branch
token_data["resource"] = self.context.get_resource_url()

if self.context.client_metadata.scope: # pragma: no branch
token_data["scope"] = self.context.client_metadata.scope

token_url = self._get_token_endpoint()
return httpx2.Request(
"POST", token_url, data=token_data, headers={"Content-Type": "application/x-www-form-urlencoded"}
)
140 changes: 1 addition & 139 deletions tests/client/auth/extensions/test_client_credentials.py
Original file line number Diff line number Diff line change
@@ -1,26 +1,20 @@
import urllib.parse
import warnings

import jwt
import pytest
from pydantic import AnyHttpUrl, AnyUrl
from pydantic import AnyHttpUrl

from mcp.client.auth.extensions.client_credentials import (
ClientCredentialsOAuthProvider,
JWTParameters,
PrivateKeyJWTOAuthProvider,
RFC7523OAuthClientProvider,
SignedJWTParameters,
static_assertion_provider,
)
from mcp.shared.auth import (
AuthorizationCodeResult,
OAuthClientInformationFull,
OAuthClientMetadata,
OAuthMetadata,
OAuthToken,
)
from mcp.shared.exceptions import MCPDeprecationWarning


class MockTokenStorage:
Expand Down Expand Up @@ -48,138 +42,6 @@ def mock_storage():
return MockTokenStorage()


@pytest.fixture
def client_metadata():
return OAuthClientMetadata(
client_name="Test Client",
client_uri=AnyHttpUrl("https://example.com"),
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
scope="read write",
)


@pytest.fixture
def rfc7523_oauth_provider(client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage):
async def redirect_handler(url: str) -> None: # pragma: no cover
"""Mock redirect handler."""
pass

async def callback_handler() -> AuthorizationCodeResult: # pragma: no cover
"""Mock callback handler."""
return AuthorizationCodeResult(code="test_auth_code", state="test_state")

with warnings.catch_warnings():
warnings.simplefilter("ignore", MCPDeprecationWarning)
return RFC7523OAuthClientProvider(
server_url="https://api.example.com/v1/mcp",
client_metadata=client_metadata,
storage=mock_storage,
redirect_handler=redirect_handler,
callback_handler=callback_handler,
)


class TestOAuthFlowClientCredentials:
"""Test OAuth flow behavior for client credentials flows."""

@pytest.mark.anyio
async def test_token_exchange_request_jwt_predefined(self, rfc7523_oauth_provider: RFC7523OAuthClientProvider):
"""Test token exchange request building with a predefined JWT assertion."""
# Set up required context
rfc7523_oauth_provider.context.client_info = OAuthClientInformationFull(
grant_types=["urn:ietf:params:oauth:grant-type:jwt-bearer"],
token_endpoint_auth_method="private_key_jwt",
redirect_uris=None,
scope="read write",
)
rfc7523_oauth_provider.context.oauth_metadata = OAuthMetadata(
issuer=AnyHttpUrl("https://api.example.com"),
authorization_endpoint=AnyHttpUrl("https://api.example.com/authorize"),
token_endpoint=AnyHttpUrl("https://api.example.com/token"),
registration_endpoint=AnyHttpUrl("https://api.example.com/register"),
)
rfc7523_oauth_provider.context.client_metadata = rfc7523_oauth_provider.context.client_info
rfc7523_oauth_provider.context.protocol_version = "2025-06-18"
rfc7523_oauth_provider.jwt_parameters = JWTParameters(
# https://www.jwt.io
assertion="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiYWRtaW4iOnRydWUsImlhdCI6MTUxNjIzOTAyMn0.KMUFsIDTnFmyG3nMiGM6H9FNFUROf3wh7SmqJp-QV30"
)

request = await rfc7523_oauth_provider._exchange_token_jwt_bearer()

assert request.method == "POST"
assert str(request.url) == "https://api.example.com/token"
assert request.headers["Content-Type"] == "application/x-www-form-urlencoded"

# Check form data
content = urllib.parse.unquote_plus(request.content.decode())
assert "grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer" in content
assert "scope=read write" in content
assert "resource=https://api.example.com/v1/mcp" in content
assert (
"assertion=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiYWRtaW4iOnRydWUsImlhdCI6MTUxNjIzOTAyMn0.KMUFsIDTnFmyG3nMiGM6H9FNFUROf3wh7SmqJp-QV30"
in content
)

@pytest.mark.anyio
async def test_token_exchange_request_jwt(self, rfc7523_oauth_provider: RFC7523OAuthClientProvider):
"""Test token exchange request building wiith a generated JWT assertion."""
# Set up required context
rfc7523_oauth_provider.context.client_info = OAuthClientInformationFull(
grant_types=["urn:ietf:params:oauth:grant-type:jwt-bearer"],
token_endpoint_auth_method="private_key_jwt",
redirect_uris=None,
scope="read write",
)
rfc7523_oauth_provider.context.oauth_metadata = OAuthMetadata(
issuer=AnyHttpUrl("https://api.example.com"),
authorization_endpoint=AnyHttpUrl("https://api.example.com/authorize"),
token_endpoint=AnyHttpUrl("https://api.example.com/token"),
registration_endpoint=AnyHttpUrl("https://api.example.com/register"),
)
rfc7523_oauth_provider.context.client_metadata = rfc7523_oauth_provider.context.client_info
rfc7523_oauth_provider.context.protocol_version = "2025-06-18"
rfc7523_oauth_provider.jwt_parameters = JWTParameters(
issuer="foo",
subject="1234567890",
claims={
"name": "John Doe",
"admin": True,
"iat": 1516239022,
},
jwt_signing_algorithm="HS256",
jwt_signing_key="a-string-secret-at-least-256-bits-long",
jwt_lifetime_seconds=300,
)

request = await rfc7523_oauth_provider._exchange_token_jwt_bearer()

assert request.method == "POST"
assert str(request.url) == "https://api.example.com/token"
assert request.headers["Content-Type"] == "application/x-www-form-urlencoded"

# Check form data
content = urllib.parse.unquote_plus(request.content.decode()).split("&")
assert "grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer" in content
assert "scope=read write" in content
assert "resource=https://api.example.com/v1/mcp" in content

# Check assertion
assertion = next(param for param in content if param.startswith("assertion="))[len("assertion=") :]
claims = jwt.decode(
assertion,
key="a-string-secret-at-least-256-bits-long",
algorithms=["HS256"],
audience="https://api.example.com/",
subject="1234567890",
issuer="foo",
verify=True,
)
assert claims["name"] == "John Doe"
assert claims["admin"]
assert claims["iat"] == 1516239022


class TestClientCredentialsOAuthProvider:
"""Test ClientCredentialsOAuthProvider."""

Expand Down
Loading
Loading