OAuth Update: Adding the Client Credentials & Token Exchange Grant Types

docs/api.md

@@ -1 +1,5 @@
+The Python SDK exposes the entire `mcp` package for use in your own projects.
+It includes an OAuth server implementation with support for the RFC 8693
+`token_exchange` grant type.
+
 ::: mcp

docs/index.md

@@ -3,3 +3,7 @@
 This is the MCP Server implementation in Python.
 
 It only contains the [API Reference](api.md) for the time being.
+
+The built-in OAuth server supports the RFC 8693 `token_exchange` grant type,
+allowing clients to exchange user tokens from external providers for MCP
+access tokens.

examples/servers/simple-auth/mcp_simple_auth/simple_auth_provider.py

@@ -238,6 +238,52 @@ async def exchange_authorization_code(
             scope=" ".join(authorization_code.scopes),
         )
 
+    async def exchange_client_credentials(self, client: OAuthClientInformationFull, scopes: list[str]) -> OAuthToken:
+        """Exchange client credentials for an MCP access token."""
+        mcp_token = f"mcp_{secrets.token_hex(32)}"
+        self.tokens[mcp_token] = AccessToken(
+            token=mcp_token,
+            client_id=client.client_id,
+            scopes=scopes,
+            expires_at=int(time.time()) + 3600,
+        )
+        return OAuthToken(
+            access_token=mcp_token,
+            token_type="Bearer",
+            expires_in=3600,
+            scope=" ".join(scopes),
+        )
+
+    async def exchange_token(
+        self,
+        client: OAuthClientInformationFull,
+        subject_token: str,
+        subject_token_type: str,
+        actor_token: str | None,
+        actor_token_type: str | None,
+        scope: list[str] | None,
+        audience: str | None,
+        resource: str | None,
+    ) -> OAuthToken:
+        """Exchange an external token for an MCP access token."""
+        if not subject_token:
+            raise ValueError("Invalid subject token")
+
+        mcp_token = f"mcp_{secrets.token_hex(32)}"
+        self.tokens[mcp_token] = AccessToken(
+            token=mcp_token,
+            client_id=client.client_id,
+            scopes=scope or [self.settings.mcp_scope],
+            expires_at=int(time.time()) + 3600,
+            resource=resource,
+        )
+        return OAuthToken(
+            access_token=mcp_token,
+            token_type="Bearer",
+            expires_in=3600,
+            scope=" ".join(scope or [self.settings.mcp_scope]),
+        )
+
     async def load_access_token(self, token: str) -> AccessToken | None:
         """Load and validate an access token."""
         access_token = self.tokens.get(token)

src/mcp/client/streamable_http.py

@@ -173,9 +173,11 @@ async def _handle_sse_event(
                 session_message = SessionMessage(message)
                 await read_stream_writer.send(session_message)
 
-                # Call resumption token callback if we have an ID
-                if sse.id and resumption_callback:
-                    await resumption_callback(sse.id)
+                # Call resumption token callback if we have an ID. Only update
+                # the resumption token on notifications to avoid overwriting it
+                # with the token from the final response.
+                if sse.id and resumption_callback and not isinstance(message.root, JSONRPCResponse | JSONRPCError):
+                    await resumption_callback(sse.id.strip())
 
                 # If this is a response or error return True indicating completion
                 # Otherwise, return False to continue listening
@@ -221,7 +223,7 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None:
         """Handle a resumption request using GET with SSE."""
         headers = self._prepare_request_headers(ctx.headers)
         if ctx.metadata and ctx.metadata.resumption_token:
-            headers[LAST_EVENT_ID] = ctx.metadata.resumption_token
+            headers[LAST_EVENT_ID] = ctx.metadata.resumption_token.strip()
         else:
             raise ResumptionError("Resumption request requires a resumption token")
 

src/mcp/server/auth/handlers/register.py

@@ -68,11 +68,22 @@ async def handle(self, request: Request) -> Response:
                     ),
                     status_code=400,
                 )
-        if set(client_metadata.grant_types) != {"authorization_code", "refresh_token"}:
+        grant_types_set: set[str] = set(client_metadata.grant_types)
+        valid_sets = [
+            {"authorization_code", "refresh_token"},
+            {"client_credentials"},
+            {"token_exchange"},
+            {"client_credentials", "token_exchange"},
+        ]
+
+        if grant_types_set not in valid_sets:
             return PydanticJSONResponse(
                 content=RegistrationErrorResponse(
                     error="invalid_client_metadata",
-                    error_description="grant_types must be authorization_code and refresh_token",
+                    error_description=(
+                        "grant_types must be authorization_code and refresh_token "
+                        "or client_credentials or token exchange or client_credentials and token_exchange"
+                    ),
                 ),
                 status_code=400,
             )

src/mcp/server/auth/handlers/token.py

@@ -40,16 +40,39 @@ class RefreshTokenRequest(BaseModel):
     resource: str | None = Field(None, description="Resource indicator for the token")
 
 
+class ClientCredentialsRequest(BaseModel):
+    # See https://datatracker.ietf.org/doc/html/rfc6749#section-4.4
+    grant_type: Literal["client_credentials"]
+    scope: str | None = Field(None, description="Optional scope parameter")
+    client_id: str
+    client_secret: str | None = None
+
+
+class TokenExchangeRequest(BaseModel):
+    """RFC 8693 token exchange request."""
+
+    grant_type: Literal["token_exchange"]
+    subject_token: str = Field(..., description="Token to exchange")
+    subject_token_type: str = Field(..., description="Type of the subject token")
+    actor_token: str | None = Field(None, description="Optional actor token")
+    actor_token_type: str | None = Field(None, description="Type of the actor token if provided")
+    resource: str | None = None
+    audience: str | None = None
+    scope: str | None = None
+    client_id: str
+    client_secret: str | None = None
+
+
 class TokenRequest(
     RootModel[
         Annotated[
-            AuthorizationCodeRequest | RefreshTokenRequest,
+            AuthorizationCodeRequest | RefreshTokenRequest | ClientCredentialsRequest | TokenExchangeRequest,
             Field(discriminator="grant_type"),
         ]
     ]
 ):
     root: Annotated[
-        AuthorizationCodeRequest | RefreshTokenRequest,
+        AuthorizationCodeRequest | RefreshTokenRequest | ClientCredentialsRequest | TokenExchangeRequest,
         Field(discriminator="grant_type"),
     ]
 
@@ -192,10 +215,49 @@ async def handle(self, request: Request):
                         )
                     )
 
+            case ClientCredentialsRequest():
+                scopes = (
+                    token_request.scope.split(" ")
+                    if token_request.scope
+                    else client_info.scope.split(" ")
+                    if client_info.scope
+                    else []
+                )
+                try:
+                    tokens = await self.provider.exchange_client_credentials(client_info, scopes)
+                except TokenError as e:
+                    return self.response(
+                        TokenErrorResponse(
+                            error=e.error,
+                            error_description=e.error_description,
+                        )
+                    )
+
+            case TokenExchangeRequest():
+                scopes = token_request.scope.split(" ") if token_request.scope else []
+                try:
+                    tokens = await self.provider.exchange_token(
+                        client_info,
+                        token_request.subject_token,
+                        token_request.subject_token_type,
+                        token_request.actor_token,
+                        token_request.actor_token_type,
+                        scopes,
+                        token_request.audience,
+                        token_request.resource,
+                    )
+                except TokenError as e:
+                    return self.response(
+                        TokenErrorResponse(
+                            error=e.error,
+                            error_description=e.error_description,
+                        )
+                    )
+
             case RefreshTokenRequest():
                 refresh_token = await self.provider.load_refresh_token(client_info, token_request.refresh_token)
                 if refresh_token is None or refresh_token.client_id != token_request.client_id:
-                    # if token belongs to different client, pretend it doesn't exist
+                    # if token belongs to a different client, pretend it doesn't exist
                     return self.response(
                         TokenErrorResponse(
                             error="invalid_grant",

src/mcp/server/auth/provider.py

@@ -80,6 +80,7 @@ class AuthorizeError(Exception):
     "unauthorized_client",
     "unsupported_grant_type",
     "invalid_scope",
+    "invalid_target",
 ]
 
 
@@ -248,6 +249,24 @@ async def exchange_refresh_token(
         """
         ...
 
+    async def exchange_client_credentials(self, client: OAuthClientInformationFull, scopes: list[str]) -> OAuthToken:
+        """Exchange client credentials for an MCP access token."""
+        ...
+
+    async def exchange_token(
+        self,
+        client: OAuthClientInformationFull,
+        subject_token: str,
+        subject_token_type: str,
+        actor_token: str | None,
+        actor_token_type: str | None,
+        scope: list[str] | None,
+        audience: str | None,
+        resource: str | None,
+    ) -> OAuthToken:
+        """Exchange an external token for an MCP access token."""
+        ...
+
     async def load_access_token(self, token: str) -> AccessTokenT | None:
         """
         Loads an access token by its token.

src/mcp/server/auth/routes.py

@@ -163,7 +163,12 @@ def build_metadata(
         scopes_supported=client_registration_options.valid_scopes,
         response_types_supported=["code"],
         response_modes_supported=None,
-        grant_types_supported=["authorization_code", "refresh_token"],
+        grant_types_supported=[
+            "authorization_code",
+            "refresh_token",
+            "client_credentials",
+            "token_exchange",
+        ],
         token_endpoint_auth_methods_supported=["client_secret_post"],
         token_endpoint_auth_signing_alg_values_supported=None,
         service_documentation=service_documentation_url,

src/mcp/shared/auth.py

@@ -13,6 +13,7 @@ class OAuthToken(BaseModel):
     expires_in: int | None = None
     scope: str | None = None
     refresh_token: str | None = None
+    issued_token_type: str | None = None
 
     @field_validator("token_type", mode="before")
     @classmethod
@@ -46,8 +47,15 @@ class OAuthClientMetadata(BaseModel):
     # client_secret_post;
     # ie: we do not support client_secret_basic
     token_endpoint_auth_method: Literal["none", "client_secret_post"] = "client_secret_post"
-    # grant_types: this implementation only supports authorization_code & refresh_token
-    grant_types: list[Literal["authorization_code", "refresh_token"]] = [
+    # grant_types: this implementation supports authorization_code, refresh_token, client_credentials, & token_exchange
+    grant_types: list[
+        Literal[
+            "authorization_code",
+            "refresh_token",
+            "client_credentials",
+            "token_exchange",
+        ]
+    ] = [
         "authorization_code",
         "refresh_token",
     ]
@@ -114,10 +122,35 @@ class OAuthMetadata(BaseModel):
     registration_endpoint: AnyHttpUrl | None = None
     scopes_supported: list[str] | None = None
     response_types_supported: list[str] = ["code"]
-    response_modes_supported: list[str] | None = None
-    grant_types_supported: list[str] | None = None
-    token_endpoint_auth_methods_supported: list[str] | None = None
-    token_endpoint_auth_signing_alg_values_supported: list[str] | None = None
+    response_modes_supported: (
+        list[
+            Literal[
+                "query",
+                "fragment",
+                "form_post",
+                "query.jwt",
+                "fragment.jwt",
+                "form_post.jwt",
+                "jwt",
+            ]
+        ]
+        | None
+    ) = None
+    grant_types_supported: (
+        list[
+            Literal[
+                "authorization_code",
+                "refresh_token",
+                "client_credentials",
+                "token_exchange",
+            ]
+        ]
+        | None
+    ) = None
+    token_endpoint_auth_methods_supported: list[Literal["none", "client_secret_post", "client_secret_basic"]] | None = (
+        None
+    )
+    token_endpoint_auth_signing_alg_values_supported: None = None
     service_documentation: AnyHttpUrl | None = None
     ui_locales_supported: list[str] | None = None
     op_policy_uri: AnyHttpUrl | None = None

src/mcp/shared/session.py

@@ -312,7 +312,10 @@ async def send_notification(
             message=JSONRPCMessage(jsonrpc_notification),
             metadata=ServerMessageMetadata(related_request_id=related_request_id) if related_request_id else None,
         )
-        await self._write_stream.send(session_message)
+        try:
+            await self._write_stream.send(session_message)
+        except (anyio.BrokenResourceError, anyio.ClosedResourceError):
+            logging.debug("Discarding notification due to closed stream")
 
     async def _send_response(self, request_id: RequestId, response: SendResultT | ErrorData) -> None:
         if isinstance(response, ErrorData):