Support RFC 8693 token exchange for enterprise IdP flows (SEP-990)

Add the OAuth 2.0 token-exchange grant (urn:ietf:params:oauth:grant-type:token-exchange),
the wire mechanism behind SEP-990. A client exchanges an enterprise IdP-issued ID-JAG for
an MCP access token at the authorization server's token endpoint.

Client: TokenExchangeOAuthProvider posts the exchange, sourcing the ID-JAG from an async
subject_token_provider(audience) callback.

Server: OAuthAuthorizationServerProvider.exchange_token validates the subject token and
issues the access token; gated by AuthSettings.token_exchange_enabled, which also advertises
the grant (and the public-client 'none' auth method) in metadata. TokenError gains
invalid_target; TokenExchangeToken carries RFC 8693 issued_token_type.

Author: Kludex

docs/migration.md

@@ -1431,6 +1431,35 @@ client_metadata = OAuthClientMetadata(
 
 Under OIDC, omitting `application_type` defaults to `"web"`, which an authorization server may reject for the `localhost` redirect URIs native clients use; sending `"native"` avoids that. Non-OIDC servers ignore the parameter.
 
+### Token exchange for enterprise IdP flows (SEP-990)
+
+The SDK now supports the RFC 8693 token-exchange grant (`urn:ietf:params:oauth:grant-type:token-exchange`), the wire mechanism behind SEP-990's enterprise identity-provider policy controls. A client presents a security token issued by an enterprise IdP - the Identity Assertion Authorization Grant (ID-JAG) - to the MCP authorization server, which validates it and returns an MCP access token. This is additive and opt-in on both sides; existing flows are unchanged.
+
+On the client, `TokenExchangeOAuthProvider` (in `mcp.client.auth.extensions.token_exchange`) is an `httpx.Auth` that posts the exchange request, mirroring the `client_credentials` extension. The ID-JAG is supplied lazily through an async `subject_token_provider(audience)` callback - the SDK does not implement IdP login or the first exchange against the IdP, which are deployment-specific:
+
+```python
+from mcp.client.auth.extensions.token_exchange import TokenExchangeOAuthProvider
+
+
+async def fetch_id_jag(audience: str) -> str:
+    # `audience` is the authorization server's issuer; the ID-JAG must carry it as `aud`.
+    return await my_idp.exchange_id_token_for_id_jag(audience=audience)
+
+
+provider = TokenExchangeOAuthProvider(
+    server_url="https://mcp.example.com/mcp",
+    storage=my_token_storage,
+    client_id="enterprise-mcp-client",
+    subject_token_provider=fetch_id_jag,
+)
+```
+
+The client defaults to a public client (`token_endpoint_auth_method="none"`); pass a `client_secret` (optionally with `token_endpoint_auth_method="client_secret_basic"`) for a confidential client. `requested_token_type` defaults to `urn:ietf:params:oauth:token-type:access_token` since the SEP-990 output is an MCP access token; pass `None` to omit it.
+
+On the authorization server, set `AuthSettings(token_exchange_enabled=True)` (or pass `token_exchange_enabled=True` to `create_auth_routes`) and implement `exchange_token` on your `OAuthAuthorizationServerProvider`. The method receives a `TokenExchangeParams` (subject token, token types, scopes, resource, audience), validates the subject token, decides which scopes to grant, and returns an `OAuthToken` (or a `TokenExchangeToken` to set RFC 8693's `issued_token_type`, which the handler otherwise defaults to the access-token type). The flag gates both metadata advertisement and the token endpoint: when it is off, `/token` rejects the grant with `unsupported_grant_type` even if the provider implements `exchange_token`. When on, the authorization server also advertises the `none` token-endpoint auth method for the common public-client case. The base `exchange_token` rejects every request with `unsupported_grant_type`, and `TokenError` now includes RFC 8693's `invalid_target` for unknown `resource`/`audience` targets.
+
+The provider owns scope decisions: requested scopes must never be granted verbatim, since a valid ID-JAG could otherwise be exchanged for arbitrary access. Derive the granted scopes from the validated ID-JAG and policy and narrow the request against them (see `examples/snippets/servers/token_exchange_server.py`). Note also that the bundled Dynamic Client Registration handler still requires `authorization_code` in `grant_types`, so a token-exchange client must register both grants (or be pre-registered).
+
 ### 2025-11-25 and 2026-07-28 protocol fields modeled
 
 `mcp_types` models the 2025-11-25 and 2026-07-28 protocol fields (e.g. `resultType`, `ttlMs`/`cacheScope` on cacheable results, `inputResponses`/`requestState` on retried requests), so inbound payloads carrying these keys parse into typed fields and round-trip. `ttlMs`/`cacheScope` default to `0`/`"private"` (immediately stale, not shared-cacheable); `resultType` defaults to `"complete"` on concrete results (`None` on `EmptyResult`); the server strips all of them from the wire at pre-2026 versions.

examples/snippets/clients/token_exchange_client.py

@@ -0,0 +1,75 @@
+"""Client side of SEP-990 (enterprise IdP policy controls).
+
+`TokenExchangeOAuthProvider` exchanges a subject token issued by the enterprise IdP - the
+Identity Assertion Authorization Grant (ID-JAG) - for an MCP access token, using the RFC 8693
+token-exchange grant at the MCP authorization server's token endpoint. No browser redirect or
+dynamic client registration is involved.
+
+Obtaining the ID-JAG (logging into the IdP and performing the first exchange against it) is
+deployment-specific and out of scope for the SDK; supply it through the `subject_token_provider`
+callback. The callback receives the authorization server's issuer identifier as its audience.
+"""
+
+import asyncio
+
+import httpx
+
+from mcp import ClientSession
+from mcp.client.auth.extensions.token_exchange import TokenExchangeOAuthProvider
+from mcp.client.streamable_http import streamable_http_client
+from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
+
+
+class InMemoryTokenStorage:
+    """Demo in-memory token storage."""
+
+    def __init__(self) -> None:
+        self.tokens: OAuthToken | None = None
+        self.client_info: OAuthClientInformationFull | None = None
+
+    async def get_tokens(self) -> OAuthToken | None:
+        return self.tokens
+
+    async def set_tokens(self, tokens: OAuthToken) -> None:
+        self.tokens = tokens
+
+    async def get_client_info(self) -> OAuthClientInformationFull | None:
+        return self.client_info
+
+    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
+        self.client_info = client_info
+
+
+async def fetch_id_jag(audience: str) -> str:
+    """Return the ID-JAG to exchange.
+
+    `audience` is the MCP authorization server's issuer identifier; the returned ID-JAG must
+    carry it as the `aud` claim. In production this exchanges the user's IdP ID token for an
+    ID-JAG against the enterprise identity provider.
+    """
+    raise NotImplementedError("Obtain the ID-JAG from your enterprise identity provider")
+
+
+async def main() -> None:
+    oauth_auth = TokenExchangeOAuthProvider(
+        server_url="http://localhost:8001/mcp",
+        storage=InMemoryTokenStorage(),
+        client_id="enterprise-mcp-client",
+        subject_token_provider=fetch_id_jag,
+        scopes="user",
+    )
+
+    async with httpx.AsyncClient(auth=oauth_auth, follow_redirects=True) as http_client:
+        async with streamable_http_client("http://localhost:8001/mcp", http_client=http_client) as (read, write):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
+                tools = await session.list_tools()
+                print(f"Available tools: {[tool.name for tool in tools.tools]}")
+
+
+def run() -> None:
+    asyncio.run(main())
+
+
+if __name__ == "__main__":
+    run()

examples/snippets/pyproject.toml

@@ -21,4 +21,5 @@ completion-client = "clients.completion_client:main"
 direct-execution-server = "servers.direct_execution:main"
 display-utilities-client = "clients.display_utilities:main"
 oauth-client = "clients.oauth_client:run"
+token-exchange-client = "clients.token_exchange_client:run"
 elicitation-client = "clients.url_elicitation_client:run"

examples/snippets/servers/token_exchange_server.py

@@ -0,0 +1,94 @@
+"""Authorization-server side of SEP-990 (enterprise IdP policy controls).
+
+An authorization server enables the RFC 8693 token-exchange grant by setting
+`token_exchange_enabled=True` and implementing `exchange_token` on its provider. The
+provider validates the subject token (the ID-JAG issued by the enterprise IdP), decides which
+scopes the exchange may grant, and mints an MCP access token.
+
+Two responsibilities are the provider's and must not be skipped (both are stubbed here):
+- Validate the ID-JAG: signature against the IdP's keys, issuer/audience/expiry, and the
+  organization's policy. Returning a subject for any non-empty token, as `_validate_id_jag`
+  does below, is NOT safe for production.
+- Narrow the granted scopes. The scopes a client requests must never be granted verbatim; the
+  ID-JAG and policy determine what is permitted. `_grant_scopes` intersects the request with
+  what the subject is allowed.
+
+Wire the returned routes into a Starlette app with `create_auth_routes(...,
+token_exchange_enabled=True)`, or set `AuthSettings(token_exchange_enabled=True)` when using
+`MCPServer`/`Server` with an `auth_server_provider`.
+"""
+
+import secrets
+import time
+
+from mcp.server.auth.provider import (
+    AccessToken,
+    AuthorizationCode,
+    OAuthAuthorizationServerProvider,
+    RefreshToken,
+    TokenError,
+    TokenExchangeParams,
+)
+from mcp.shared.auth import OAuthClientInformationFull, OAuthToken, TokenExchangeToken
+
+# The scopes this server is willing to grant via token exchange. A real implementation derives
+# the permitted set from the validated ID-JAG and the organization's policy, not from a constant.
+ALLOWED_SCOPES = ("mcp",)
+
+
+class TokenExchangeProvider(OAuthAuthorizationServerProvider[AuthorizationCode, RefreshToken, AccessToken]):
+    """Authorization-server provider that accepts an ID-JAG via RFC 8693 token exchange."""
+
+    def __init__(self) -> None:
+        self.access_tokens: dict[str, AccessToken] = {}
+
+    async def exchange_token(self, client: OAuthClientInformationFull, params: TokenExchangeParams) -> OAuthToken:
+        subject = self._validate_id_jag(params.subject_token)
+        if subject is None:
+            raise TokenError(error="invalid_grant", error_description="Invalid or rejected subject token")
+
+        scopes = self._grant_scopes(params.scopes)
+
+        assert client.client_id is not None
+        access_token = f"access_{secrets.token_hex(16)}"
+        self.access_tokens[access_token] = AccessToken(
+            token=access_token,
+            client_id=client.client_id,
+            scopes=scopes,
+            expires_at=int(time.time()) + 3600,
+            resource=params.resource,
+            subject=subject,
+        )
+        # TokenExchangeToken sets RFC 8693's issued_token_type; it defaults to the access-token type.
+        return TokenExchangeToken(
+            access_token=access_token,
+            token_type="Bearer",
+            expires_in=3600,
+            scope=" ".join(scopes),
+        )
+
+    def _validate_id_jag(self, subject_token: str) -> str | None:
+        """Validate the ID-JAG and return the subject, or None to reject.
+
+        A real implementation verifies the JWT signature against the IdP's keys, checks the
+        issuer/audience/expiry, and applies the organization's policy. This stub accepts any
+        non-empty token and uses it as the subject identifier - replace it before production.
+        """
+        return subject_token or None
+
+    def _grant_scopes(self, requested: list[str] | None) -> list[str]:
+        """Return the scopes to grant, never exceeding what this server permits.
+
+        Requested scopes are intersected with the allowed set so a valid ID-JAG can never be
+        exchanged for broader access than policy allows; when none are requested, the full
+        allowed set is granted.
+        """
+        if requested is None:
+            return list(ALLOWED_SCOPES)
+        granted = [scope for scope in requested if scope in ALLOWED_SCOPES]
+        if not granted:
+            raise TokenError(error="invalid_scope", error_description="No requested scope is permitted")
+        return granted
+
+    async def load_access_token(self, token: str) -> AccessToken | None:
+        return self.access_tokens.get(token)

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

@@ -0,0 +1,146 @@
+"""OAuth 2.0 Token Exchange (RFC 8693) client provider for MCP.
+
+Provides `TokenExchangeOAuthProvider`, which exchanges a security token issued by an
+enterprise identity provider for an MCP access token at the MCP authorization server's
+token endpoint. This is the client side of SEP-990 (enterprise IdP policy controls): the
+client first obtains an Identity Assertion Authorization Grant (ID-JAG) from the IdP, then
+exchanges it here for a token usable against the MCP server.
+
+Obtaining the ID-JAG (logging into the IdP and performing the first token exchange against
+it) is deployment-specific and out of scope for the SDK. The caller supplies it through the
+`subject_token_provider` callback, which receives the MCP authorization server's issuer
+identifier as its audience and returns the security token to exchange.
+"""
+
+from collections.abc import Awaitable, Callable
+from typing import Any, Literal
+
+import httpx
+
+from mcp.client.auth import OAuthClientProvider, OAuthFlowError, TokenStorage
+from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata
+
+TOKEN_EXCHANGE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange"
+JWT_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:jwt"
+ACCESS_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token"
+
+
+class TokenExchangeOAuthProvider(OAuthClientProvider):
+    """OAuth provider for the RFC 8693 token-exchange grant.
+
+    Exchanges a subject token (for SEP-990, the IdP-issued ID-JAG) for an MCP access token,
+    bypassing the interactive authorization-code flow and dynamic client registration. The
+    subject token is fetched lazily from `subject_token_provider` so a fresh token is used on
+    each exchange.
+
+    Example:
+        ```python
+        async def fetch_id_jag(audience: str) -> str:
+            # `audience` is the MCP authorization server's issuer identifier; the returned
+            # ID-JAG must carry that as its `aud` claim. How the ID-JAG is obtained from the
+            # enterprise IdP is deployment-specific and not handled by the SDK.
+            return await my_idp.exchange_id_token_for_id_jag(audience=audience)
+
+
+        provider = TokenExchangeOAuthProvider(
+            server_url="https://mcp.example.com/mcp",
+            storage=my_token_storage,
+            client_id="my-client-id",
+            subject_token_provider=fetch_id_jag,
+        )
+        ```
+    """
+
+    def __init__(
+        self,
+        server_url: str,
+        storage: TokenStorage,
+        client_id: str,
+        subject_token_provider: Callable[[str], Awaitable[str]],
+        subject_token_type: str = JWT_TOKEN_TYPE,
+        requested_token_type: str | None = ACCESS_TOKEN_TYPE,
+        scopes: str | None = None,
+        client_secret: str | None = None,
+        token_endpoint_auth_method: Literal["client_secret_basic", "client_secret_post"] = "client_secret_post",
+    ) -> None:
+        """Initialize the token-exchange OAuth provider.
+
+        Args:
+            server_url: The MCP server URL.
+            storage: Token storage implementation.
+            client_id: The OAuth client ID registered with the MCP authorization server.
+            subject_token_provider: Async callback that takes the authorization server's
+                issuer identifier (the audience) and returns the subject token to exchange
+                (the ID-JAG for SEP-990).
+            subject_token_type: RFC 8693 type identifier of the subject token. Defaults to
+                `urn:ietf:params:oauth:token-type:jwt`, the type of an ID-JAG.
+            requested_token_type: RFC 8693 desired type of the issued token. Defaults to
+                `urn:ietf:params:oauth:token-type:access_token`, since SEP-990 yields an MCP
+                access token; pass `None` to omit the parameter.
+            scopes: Optional space-separated list of scopes to request.
+            client_secret: Optional client secret. When set, the request authenticates as a
+                confidential client using `token_endpoint_auth_method`; otherwise the public
+                client form is used and the method is `none`.
+            token_endpoint_auth_method: Authentication method when `client_secret` is set.
+                Either `client_secret_post` (default) or `client_secret_basic`.
+        """
+        auth_method = token_endpoint_auth_method if client_secret is not None else "none"
+        client_metadata = OAuthClientMetadata(
+            redirect_uris=None,
+            grant_types=[TOKEN_EXCHANGE_GRANT_TYPE],
+            token_endpoint_auth_method=auth_method,
+            scope=scopes,
+        )
+        super().__init__(server_url, client_metadata, storage, None, None, 300.0)
+        self._subject_token_provider = subject_token_provider
+        self._subject_token_type = subject_token_type
+        self._requested_token_type = requested_token_type
+        self._fixed_client_info = OAuthClientInformationFull(
+            redirect_uris=None,
+            client_id=client_id,
+            client_secret=client_secret,
+            grant_types=[TOKEN_EXCHANGE_GRANT_TYPE],
+            token_endpoint_auth_method=auth_method,
+            scope=scopes,
+        )
+
+    async def _initialize(self) -> None:
+        """Load stored tokens and set pre-configured client_info."""
+        self.context.current_tokens = await self.context.storage.get_tokens()
+        self.context.client_info = self._fixed_client_info
+        self._initialized = True
+
+    async def _perform_authorization(self) -> httpx.Request:
+        """Perform the token-exchange grant."""
+        return await self._exchange_token()
+
+    async def _exchange_token(self) -> httpx.Request:
+        """Build the RFC 8693 token-exchange request."""
+        if not self.context.oauth_metadata:
+            raise OAuthFlowError("Missing OAuth metadata for token exchange")  # pragma: no cover
+        if not self.context.client_info:
+            raise OAuthFlowError("Missing client info for token exchange")  # pragma: no cover
+
+        audience = str(self.context.oauth_metadata.issuer)
+        subject_token = await self._subject_token_provider(audience)
+
+        token_data: dict[str, Any] = {
+            "grant_type": TOKEN_EXCHANGE_GRANT_TYPE,
+            "subject_token": subject_token,
+            "subject_token_type": self._subject_token_type,
+            "client_id": self.context.client_info.client_id,
+        }
+        if self._requested_token_type is not None:
+            token_data["requested_token_type"] = self._requested_token_type
+
+        headers: dict[str, str] = {"Content-Type": "application/x-www-form-urlencoded"}
+        token_data, headers = self.context.prepare_token_auth(token_data, headers)
+
+        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
+
+        token_url = self._get_token_endpoint()
+        return httpx.Request("POST", token_url, data=token_data, headers=headers)