Add end-to-end interaction tests for token exchange (SEP-990)

Exercise the full stack - the real TokenExchangeOAuthProvider client against the real
authorization-server token endpoint with InMemoryAuthorizationServerProvider implementing
exchange_token - rather than shimming the grant as the M2M providers do.

Covers: successful exchange + bearer authorizes the request, the subject_token_provider
callback audience, server-disabled rejection (unsupported_grant_type), an unaccepted subject
token, scope adoption from AS metadata, and metadata advertisement of the grant + public-client
auth method. Adds the client-auth:token-exchange* requirement entries and threads
token_exchange_enabled through the auth harness.

Removes a now-covered pragma in OAuthClientProvider._handle_token_response: these tests are the
first to drive a non-2xx token response through the full client flow.

Author: Kludex

src/mcp/client/auth/oauth2.py

@@ -417,9 +417,9 @@ async def _exchange_token_authorization_code(
     async def _handle_token_response(self, response: httpx.Response) -> None:
         """Handle token exchange response."""
         if response.status_code not in {200, 201}:
-            body = await response.aread()  # pragma: no cover
-            body_text = body.decode("utf-8")  # pragma: no cover
-            raise OAuthTokenError(f"Token exchange failed ({response.status_code}): {body_text}")  # pragma: no cover
+            body = await response.aread()
+            body_text = body.decode("utf-8")
+            raise OAuthTokenError(f"Token exchange failed ({response.status_code}): {body_text}")
 
         # Parse and validate response with scope validation
         token_response = await handle_token_response_scopes(response)

tests/interaction/_requirements.py

@@ -3676,6 +3676,53 @@ def __post_init__(self) -> None:
             "elsewhere into the auth provider's state, so the absence cannot be observed end to end."
         ),
     ),
+    "client-auth:token-exchange": Requirement(
+        source="sdk",
+        behavior=(
+            "The token-exchange provider (SEP-990) exchanges an enterprise IdP-issued ID-JAG for an MCP "
+            "access token via the RFC 8693 token-exchange grant, with no authorize or registration step, "
+            "and the issued bearer token authorizes subsequent requests."
+        ),
+        transports=("streamable-http",),
+        note="OAuth is HTTP-only.",
+    ),
+    "client-auth:token-exchange:subject-token-callback": Requirement(
+        source="sdk",
+        behavior=(
+            "The token-exchange provider sources the subject token from its async subject_token_provider "
+            "callback, invoked with the authorization server's issuer as the audience, and sends it as "
+            "subject_token on the RFC 8693 request."
+        ),
+        transports=("streamable-http",),
+        note="OAuth is HTTP-only.",
+    ),
+    "client-auth:token-exchange:disabled-rejected": Requirement(
+        source="sdk",
+        behavior=(
+            "When the authorization server has token exchange disabled, the token endpoint rejects the "
+            "grant with unsupported_grant_type and the connection fails rather than issuing a token."
+        ),
+        transports=("streamable-http",),
+        note="OAuth is HTTP-only.",
+    ),
+    "client-auth:token-exchange:invalid-subject-token": Requirement(
+        source="sdk",
+        behavior=(
+            "A token-exchange request whose subject token the authorization server rejects surfaces as an "
+            "OAuth error and the connection fails rather than proceeding with a bearer token."
+        ),
+        transports=("streamable-http",),
+        note="OAuth is HTTP-only.",
+    ),
+    "client-auth:token-exchange:metadata-advertised": Requirement(
+        source="sdk",
+        behavior=(
+            "When token exchange is enabled, the authorization-server metadata advertises the "
+            "token-exchange grant type and the public-client `none` token-endpoint auth method."
+        ),
+        transports=("streamable-http",),
+        note="OAuth is HTTP-only.",
+    ),
     # ═══════════════════════════════════════════════════════════════════════════
     # stdio transport
     # ═══════════════════════════════════════════════════════════════════════════

tests/interaction/auth/_harness.py

@@ -178,7 +178,10 @@ async def callback_handler(self) -> AuthorizationCodeResult:
 
 
 def auth_settings(
-    *, required_scopes: Sequence[str] = ("mcp",), valid_scopes: Sequence[str] | None = None
+    *,
+    required_scopes: Sequence[str] = ("mcp",),
+    valid_scopes: Sequence[str] | None = None,
+    token_exchange_enabled: bool = False,
 ) -> AuthSettings:
     """Build `AuthSettings` for the co-hosted authorization + resource server.
 
@@ -188,6 +191,9 @@ def auth_settings(
     validation; tests pass a wider set when they need the protected-resource metadata's
     `scopes_supported` (which mirrors `required_scopes`) to differ from what the client may
     register or when AS metadata should advertise additional scopes such as `offline_access`.
+
+    `token_exchange_enabled` advertises and accepts the RFC 8693 token-exchange grant (SEP-990);
+    the provider must implement `exchange_token` for the endpoint to issue tokens.
     """
     required = list(required_scopes)
     valid = list(valid_scopes) if valid_scopes is not None else required
@@ -199,6 +205,7 @@ def auth_settings(
             enabled=True, valid_scopes=valid, default_scopes=required
         ),
         revocation_options=RevocationOptions(enabled=False),
+        token_exchange_enabled=token_exchange_enabled,
     )
 
 

tests/interaction/auth/_provider.py

@@ -18,13 +18,18 @@
     OAuthAuthorizationServerProvider,
     RefreshToken,
     TokenError,
+    TokenExchangeParams,
     construct_redirect_uri,
 )
-from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
+from mcp.shared.auth import OAuthClientInformationFull, OAuthToken, TokenExchangeToken
 from tests.interaction._connect import BASE_URL
 
 _TOKEN_LIFETIME_SECONDS = 3600
 
+# The only subject token the in-memory provider accepts as a valid ID-JAG; any other value is
+# rejected with invalid_grant, standing in for signature/policy validation a real AS performs.
+VALID_SUBJECT_TOKEN = "valid-id-jag"
+
 
 class InMemoryAuthorizationServerProvider(
     OAuthAuthorizationServerProvider[AuthorizationCode, RefreshToken, AccessToken]
@@ -71,6 +76,9 @@ def __init__(
         self.codes: dict[str, AuthorizationCode] = {}
         self.refresh_tokens: dict[str, RefreshToken] = {}
         self.access_tokens: dict[str, AccessToken] = {}
+        # The most recent token-exchange request the SDK handler passed to exchange_token, for
+        # tests to assert what the client sent (None until the first exchange).
+        self.last_exchange_params: TokenExchangeParams | None = None
 
     def _next_expires_in(self) -> int:
         self._tokens_issued += 1
@@ -187,6 +195,27 @@ async def exchange_refresh_token(
             refresh_token=new_refresh,
         )
 
+    async def exchange_token(self, client: OAuthClientInformationFull, params: TokenExchangeParams) -> OAuthToken:
+        """Validate the ID-JAG subject token and mint an MCP access token (RFC 8693 / SEP-990).
+
+        Records `params` for inspection and rejects any subject token other than
+        `VALID_SUBJECT_TOKEN` with invalid_grant (standing in for signature/policy validation). The
+        granted scopes are exactly those the client requested; a real provider would derive them
+        from the validated ID-JAG and policy.
+        """
+        self.last_exchange_params = params
+        assert client.client_id is not None
+        if params.subject_token != VALID_SUBJECT_TOKEN:
+            raise TokenError(error="invalid_grant", error_description="subject token is not valid")
+        scopes = params.scopes if params.scopes is not None else self._default_scopes
+        access = self.mint_access_token(client_id=client.client_id, scopes=scopes, resource=params.resource)
+        return TokenExchangeToken(
+            access_token=access,
+            token_type="Bearer",
+            expires_in=self._next_expires_in(),
+            scope=" ".join(scopes),
+        )
+
     async def revoke_token(self, token: AccessToken | RefreshToken) -> None:
         """Not exercised by this suite; revocation is out of scope for the interaction tests."""
         raise NotImplementedError