Implement RFC 7523 authorization grant flow

Author: LucaButBoring

pyproject.toml

@@ -33,6 +33,7 @@ dependencies = [
     "uvicorn>=0.23.1; sys_platform != 'emscripten'",
     "jsonschema>=4.20.0",
     "pywin32>=310; sys_platform == 'win32'",
+    "pyjwt[crypto]>=2.10.1",
 ]
 
 [project.optional-dependencies]

src/mcp/client/auth.py

@@ -13,11 +13,13 @@
 import time
 from collections.abc import AsyncGenerator, Awaitable, Callable
 from dataclasses import dataclass, field
-from typing import Protocol
+from typing import Any, Protocol
 from urllib.parse import urlencode, urljoin, urlparse
+from uuid import uuid4
 
 import anyio
 import httpx
+import jwt
 from pydantic import BaseModel, Field, ValidationError
 
 from mcp.client.streamable_http import MCP_PROTOCOL_VERSION
@@ -61,6 +63,23 @@ def generate(cls) -> "PKCEParameters":
         return cls(code_verifier=code_verifier, code_challenge=code_challenge)
 
 
+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.")
+    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.")
+
+
 class TokenStorage(Protocol):
     """Protocol for token storage implementations."""
 
@@ -91,6 +110,7 @@ class OAuthContext:
     redirect_handler: Callable[[str], Awaitable[None]] | None
     callback_handler: Callable[[], Awaitable[tuple[str, str | None]]] | None
     timeout: float = 300.0
+    jwt_parameters: JWTParameters | None = None
 
     # Discovered metadata
     protected_resource_metadata: ProtectedResourceMetadata | None = None
@@ -192,6 +212,7 @@ def __init__(
         redirect_handler: Callable[[str], Awaitable[None]] | None = None,
         callback_handler: Callable[[], Awaitable[tuple[str, str | None]]] | None = None,
         timeout: float = 300.0,
+        jwt_parameters: JWTParameters | None = None,
     ):
         """Initialize OAuth2 authentication."""
         self.context = OAuthContext(
@@ -201,6 +222,7 @@ def __init__(
             redirect_handler=redirect_handler,
             callback_handler=callback_handler,
             timeout=timeout,
+            jwt_parameters=jwt_parameters,
         )
         self._initialized = False
 
@@ -314,6 +336,9 @@ async def _perform_authorization(self) -> httpx.Request:
         if "client_credentials" in self.context.client_metadata.grant_types:
             token_request = await self._exchange_token_client_credentials()
             return token_request
+        elif "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:
             auth_code, code_verifier = await self._perform_authorization_code_grant()
             token_request = await self._exchange_token_authorization_code(auth_code, code_verifier)
@@ -372,19 +397,22 @@ async def _perform_authorization_code_grant(self) -> tuple[str, str]:
         # Return auth code and code verifier for token exchange
         return auth_code, pkce_params.code_verifier
 
+    def _get_token_endpoint(self) -> str:
+        if self.context.oauth_metadata and self.context.oauth_metadata.token_endpoint:
+            token_url = str(self.context.oauth_metadata.token_endpoint)
+        else:
+            auth_base_url = self.context.get_authorization_base_url(self.context.server_url)
+            token_url = urljoin(auth_base_url, "/token")
+        return token_url
+
     async def _exchange_token_authorization_code(self, auth_code: str, code_verifier: str) -> httpx.Request:
         """Build token exchange request for authorization_code flow."""
         if self.context.client_metadata.redirect_uris is None:
             raise OAuthFlowError("No redirect URIs provided for authorization code grant")
         if not self.context.client_info:
             raise OAuthFlowError("Missing client info")
 
-        if self.context.oauth_metadata and self.context.oauth_metadata.token_endpoint:
-            token_url = str(self.context.oauth_metadata.token_endpoint)
-        else:
-            auth_base_url = self.context.get_authorization_base_url(self.context.server_url)
-            token_url = urljoin(auth_base_url, "/token")
-
+        token_url = self._get_token_endpoint()
         token_data = {
             "grant_type": "authorization_code",
             "code": auth_code,
@@ -409,19 +437,17 @@ async def _exchange_token_client_credentials(self) -> httpx.Request:
         if not self.context.client_info:
             raise OAuthFlowError("Missing client info")
 
-        if self.context.oauth_metadata and self.context.oauth_metadata.token_endpoint:
-            token_url = str(self.context.oauth_metadata.token_endpoint)
-        else:
-            auth_base_url = self.context.get_authorization_base_url(self.context.server_url)
-            token_url = urljoin(auth_base_url, "/token")
-
+        token_url = self._get_token_endpoint()
         token_data = {
             "grant_type": "client_credentials",
-            "resource": self.context.get_resource_url(),  # RFC 8707
         }
 
         headers = {"Content-Type": "application/x-www-form-urlencoded"}
 
+        # Only include resource param if conditions are met
+        if self.context.should_include_resource_param(self.context.protocol_version):
+            token_data["resource"] = self.context.get_resource_url()  # RFC 8707
+
         if self.context.client_metadata.scope:
             token_data["scope"] = self.context.client_metadata.scope
 
@@ -442,6 +468,57 @@ async def _exchange_token_client_credentials(self) -> httpx.Request:
 
         return httpx.Request("POST", token_url, data=token_data, headers=headers)
 
+    async def _exchange_token_jwt_bearer(self) -> httpx.Request:
+        """Build token exchange request for JWT bearer grant."""
+        if not self.context.client_info:
+            raise OAuthFlowError("Missing client info")
+        if not self.context.jwt_parameters:
+            raise OAuthFlowError("Missing JWT parameters")
+
+        token_url = self._get_token_endpoint()
+
+        if self.context.jwt_parameters.assertion is not None:
+            assertion = self.context.jwt_parameters.assertion
+        else:
+            if not self.context.jwt_parameters.jwt_signing_key:
+                raise OAuthFlowError("Missing signing key for JWT bearer grant")
+            if not self.context.jwt_parameters.issuer:
+                raise OAuthFlowError("Missing issuer for JWT bearer grant")
+            if not self.context.jwt_parameters.subject:
+                raise OAuthFlowError("Missing subject for JWT bearer grant")
+
+            now = int(time.time())
+            claims = {
+                "iss": self.context.jwt_parameters.issuer,
+                "sub": self.context.jwt_parameters.subject,
+                "aud": token_url,
+                "exp": now + self.context.jwt_parameters.jwt_lifetime_seconds,
+                "iat": now,
+                "jti": str(uuid4()),
+            }
+            claims.update(self.context.jwt_parameters.claims or {})
+
+            assertion = jwt.encode(
+                claims,
+                self.context.jwt_parameters.jwt_signing_key,
+                algorithm=self.context.jwt_parameters.jwt_signing_algorithm or "RS256",
+            )
+
+        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):
+            token_data["resource"] = self.context.get_resource_url()
+
+        if self.context.client_metadata.scope:
+            token_data["scope"] = self.context.client_metadata.scope
+
+        return httpx.Request(
+            "POST", token_url, data=token_data, headers={"Content-Type": "application/x-www-form-urlencoded"}
+        )
+
     async def _handle_token_response(self, response: httpx.Response) -> None:
         """Handle token exchange response."""
         if response.status_code != 200:

src/mcp/shared/auth.py

@@ -43,9 +43,15 @@ class OAuthClientMetadata(BaseModel):
 
     redirect_uris: list[AnyUrl] | None = Field(..., min_length=1)
     # supported auth methods for the token endpoint
-    token_endpoint_auth_method: Literal["none", "client_secret_basic", "client_secret_post"] = "client_secret_post"
+    token_endpoint_auth_method: Literal["none", "client_secret_basic", "client_secret_post", "private_key_jwt"] = (
+        "client_secret_post"
+    )
     # supported grant_types of this implementation
-    grant_types: list[Literal["authorization_code", "client_credentials", "refresh_token"]] = [
+    grant_types: list[
+        Literal[
+            "authorization_code", "client_credentials", "refresh_token", "urn:ietf:params:oauth:grant-type:jwt-bearer"
+        ]
+    ] = [
         "authorization_code",
         "refresh_token",
     ]

tests/client/test_auth.py

@@ -9,11 +9,12 @@
 from unittest import mock
 
 import httpx
+import jwt
 import pytest
 from inline_snapshot import Is, snapshot
 from pydantic import AnyHttpUrl, AnyUrl
 
-from mcp.client.auth import OAuthClientProvider, PKCEParameters
+from mcp.client.auth import JWTParameters, OAuthClientProvider, PKCEParameters
 from mcp.shared.auth import (
     OAuthClientInformationFull,
     OAuthClientMetadata,
@@ -434,7 +435,7 @@ async def test_token_exchange_request_authorization_code(self, oauth_provider):
         assert "client_secret=test_secret" in content
 
     @pytest.mark.anyio
-    async def test_token_exchange_request_client_credentials_basic(self, oauth_provider):
+    async def test_token_exchange_request_client_credentials_basic(self, oauth_provider: OAuthClientProvider):
         """Test token exchange request building."""
         # Set up required context
         oauth_provider.context.client_info = oauth_provider.context.client_metadata = OAuthClientInformationFull(
@@ -445,6 +446,7 @@ async def test_token_exchange_request_client_credentials_basic(self, oauth_provi
             redirect_uris=None,
             scope="read write",
         )
+        oauth_provider.context.protocol_version = "2025-06-18"
 
         request = await oauth_provider._exchange_token_client_credentials()
 
@@ -466,7 +468,7 @@ async def test_token_exchange_request_client_credentials_basic(self, oauth_provi
         assert base64.b64decode(request.headers["Authorization"].split(" ")[1]).decode() == "test_client:test_secret"
 
     @pytest.mark.anyio
-    async def test_token_exchange_request_client_credentials_post(self, oauth_provider):
+    async def test_token_exchange_request_client_credentials_post(self, oauth_provider: OAuthClientProvider):
         """Test token exchange request building."""
         # Set up required context
         oauth_provider.context.client_info = oauth_provider.context.client_metadata = OAuthClientInformationFull(
@@ -477,6 +479,7 @@ async def test_token_exchange_request_client_credentials_post(self, oauth_provid
             redirect_uris=None,
             scope="read write",
         )
+        oauth_provider.context.protocol_version = "2025-06-18"
 
         request = await oauth_provider._exchange_token_client_credentials()
 
@@ -492,6 +495,89 @@ async def test_token_exchange_request_client_credentials_post(self, oauth_provid
         assert "client_id=test_client" in content
         assert "client_secret=test_secret" in content
 
+    @pytest.mark.anyio
+    async def test_token_exchange_request_jwt_predefined(self, oauth_provider: OAuthClientProvider):
+        """Test token exchange request building with a predefined JWT assertion."""
+        # Set up required context
+        oauth_provider.context.client_info = oauth_provider.context.client_metadata = OAuthClientInformationFull(
+            grant_types=["urn:ietf:params:oauth:grant-type:jwt-bearer"],
+            token_endpoint_auth_method="private_key_jwt",
+            redirect_uris=None,
+            scope="read write",
+        )
+        oauth_provider.context.protocol_version = "2025-06-18"
+        oauth_provider.context.jwt_parameters = JWTParameters(
+            # https://www.jwt.io
+            assertion="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiYWRtaW4iOnRydWUsImlhdCI6MTUxNjIzOTAyMn0.KMUFsIDTnFmyG3nMiGM6H9FNFUROf3wh7SmqJp-QV30"
+        )
+
+        request = await 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, oauth_provider: OAuthClientProvider):
+        """Test token exchange request building wiith a generated JWT assertion."""
+        # Set up required context
+        oauth_provider.context.client_info = oauth_provider.context.client_metadata = OAuthClientInformationFull(
+            grant_types=["urn:ietf:params:oauth:grant-type:jwt-bearer"],
+            token_endpoint_auth_method="private_key_jwt",
+            redirect_uris=None,
+            scope="read write",
+        )
+        oauth_provider.context.protocol_version = "2025-06-18"
+        oauth_provider.context.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 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/token",
+            subject="1234567890",
+            issuer="foo",
+            verify=True,
+        )
+        assert claims["name"] == "John Doe"
+        assert claims["admin"]
+        assert claims["iat"] == 1516239022
+
     @pytest.mark.anyio
     async def test_refresh_token_request(self, oauth_provider, valid_tokens):
         """Test refresh token request building."""