feat(mcp): migrate the token_exchange (OBO) arm to the v2 resolver

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -71,6 +71,9 @@
 from litellm.proxy._experimental.mcp_server.outbound_credentials.per_user_oauth_store import (
     LazyPerUserOAuthTokenStore,
 )
+from litellm.proxy._experimental.mcp_server.outbound_credentials.token_exchange_provider import (
+    build_token_exchanger,
+)
 from litellm.proxy._experimental.mcp_server.outbound_credentials.types import (
     AuthorizationCodeConfig,
 )
@@ -531,7 +534,8 @@ def _resolve_oauth2_flow(
 
     def __init__(self, cred_provider: Optional[UpstreamCredentialProvider] = None):
         self._cred_provider = cred_provider or UpstreamCredentialProvider(
-            oauth_token_store=LazyPerUserOAuthTokenStore(self.get_mcp_server_by_id)
+            oauth_token_store=LazyPerUserOAuthTokenStore(self.get_mcp_server_by_id),
+            token_exchanger=build_token_exchanger(),
         )
         self.registry: dict[str, MCPServer] = {}
         self.config_mcp_servers: dict[str, MCPServer] = {}
@@ -1977,9 +1981,11 @@ async def _create_mcp_client(
                         ):
                             resolved_auth = None
                     case Error(err):
-                        if err.tag == "unauthorized":
-                            # The arm signals a missing per-user token semantically; raise the
-                            # per-server OAuth challenge here, where the full MCPServer is in hand.
+                        if err.tag == "unauthorized" and isinstance(spec.config, AuthorizationCodeConfig):
+                            # authorization_code's missing per-user token -> the per-server
+                            # browser-OAuth challenge, built here where the full MCPServer is in
+                            # hand. token_exchange and other modes carry their own 401 (e.g. OBO
+                            # needs a caller token, not a browser flow), so they go via raise_public.
                             raise_user_oauth_challenge(server)
                         raise_public(err)
                 return MCPClient(

litellm/proxy/_experimental/mcp_server/outbound_credentials/adapter.py

@@ -26,6 +26,7 @@
     ServerSpec,
     SharedKey,
     Subject,
+    TokenExchangeConfig,
 )
 from litellm.types.mcp import MCPAuth
 
@@ -61,8 +62,9 @@ def to_server_spec(server: MCPServer) -> Optional[ServerSpec]:
     an ``assert_never`` tail, so a newly added auth mode fails the type gate here until it is
     explicitly mapped or explicitly deferred, rather than silently falling through to v1. Live
     modes: ``none``, the static-header family (``api_key`` plus the Authorization schemes,
-    all shared-key), and ``oauth2`` per-user tokens (``authorization_code``); client_credentials
-    (M2M), delegated/passthrough oauth2, token exchange, and SigV4 return None and stay on v1.
+    all shared-key), ``oauth2`` per-user tokens (``authorization_code``), and
+    ``oauth2_token_exchange`` (RFC 8693 OBO); client_credentials (M2M), delegated/passthrough
+    oauth2, and SigV4 return None and stay on v1.
     """
     if server.is_byok:
         return None  # per-user BYOK source not migrated yet -> defer to v1 (any auth_type)
@@ -92,11 +94,39 @@ def to_server_spec(server: MCPServer) -> Optional[ServerSpec]:
                 )
             # client_credentials (M2M) and delegate/passthrough oauth2 stay on v1
             return None
-        case MCPAuth.oauth2_token_exchange | MCPAuth.aws_sigv4:
-            return None  # token exchange and SigV4 are not migrated yet -> defer to v1
+        case MCPAuth.oauth2_token_exchange:
+            return _token_exchange_spec(server, resource)
+        case MCPAuth.aws_sigv4:
+            return None  # SigV4 is not migrated yet -> defer to v1
     assert_never(auth_type)
 
 
+def _token_exchange_spec(server: MCPServer, resource: str) -> Optional[ServerSpec]:
+    """Build a token_exchange (RFC 8693 OBO) spec, or defer (None) if the exchange config is absent.
+
+    Mirrors v1's ``has_token_exchange_config`` precondition: an endpoint (``token_exchange_endpoint``
+    or ``token_url``) plus ``client_id``/``client_secret`` must all be present, else there is nothing
+    to exchange against and the server stays on v1 (parity-safe). ``audience`` is forwarded only when
+    the operator set it; a missing one is omitted, not derived.
+    """
+    endpoint = server.token_exchange_endpoint or server.token_url
+    if not endpoint or not server.client_id or not server.client_secret:
+        return None
+    return ServerSpec(
+        server_id=server.server_id,
+        resource=resource,
+        config=TokenExchangeConfig(
+            subject_token_type=server.subject_token_type,
+            token_exchange_endpoint=endpoint,
+            audience=server.audience,
+            client_id=server.client_id,
+            client_secret=SecretStr(server.client_secret),
+            token_endpoint_auth_method=server.token_endpoint_auth_method,
+            scopes=tuple(server.scopes or ()),
+        ),
+    )
+
+
 def _shared_key_spec(
     server: MCPServer,
     resource: str,

litellm/proxy/_experimental/mcp_server/outbound_credentials/resolver.py

@@ -8,7 +8,8 @@
 at runtime instead of returning `None`.
 
 `none` and `api_key` (shared-key source) are live, as is `authorization_code`, which reads the
-user's token from the injected `OAuthTokenStore`. The remaining arms are `not_implemented` stubs
+user's token from the injected `OAuthTokenStore`, and `token_exchange`, which swaps the caller's
+inbound token through the injected `TokenExchanger`. The remaining arms are `not_implemented` stubs
 that each land in a follow-up PR with their seam. Pure v2: no imports from v1.
 """
 
@@ -31,6 +32,9 @@
     Ok,
     Result,
 )
+from litellm.proxy._experimental.mcp_server.outbound_credentials.token_exchanger import (
+    TokenExchanger,
+)
 from litellm.proxy._experimental.mcp_server.outbound_credentials.types import (
     ApiKeyConfig,
     AuthorizationCodeConfig,
@@ -55,16 +59,31 @@ async def fetch(self, user_id: str, server_id: str) -> OAuthToken | None:
         return None
 
 
+class _NullTokenExchanger:
+    """Fail-closed default: with no exchanger wired, token_exchange cannot produce a credential."""
+
+    async def exchange(
+        self, subject_token: str, server: ServerSpec, config: TokenExchangeConfig
+    ) -> Result[OAuthToken, CredError]:
+        return Error(CredError.of_misconfigured("token exchange collaborator not wired"))
+
+
 class UpstreamCredentialProvider:
     """Produces the one `httpx.Auth` for a `(subject, upstream)` pair, per declared mode.
 
     Collaborators (the per-mode credential stores and token fetchers) are injected as each arm is
     built; the live `none` and `api_key`-shared arms read from the config and need none, while
-    `authorization_code` reads the user's token from the injected `OAuthTokenStore`.
+    `authorization_code` reads the user's token from the injected `OAuthTokenStore` and
+    `token_exchange` swaps the caller's token through the injected `TokenExchanger`.
     """
 
-    def __init__(self, oauth_token_store: OAuthTokenStore | None = None) -> None:
+    def __init__(
+        self,
+        oauth_token_store: OAuthTokenStore | None = None,
+        token_exchanger: TokenExchanger | None = None,
+    ) -> None:
         self._oauth_token_store: OAuthTokenStore = oauth_token_store or _NullOAuthTokenStore()
+        self._token_exchanger: TokenExchanger = token_exchanger or _NullTokenExchanger()
 
     async def resolve_credentials(self, subject: Subject, server: ServerSpec) -> Result[httpx.Auth, CredError]:
         match server.config:
@@ -76,8 +95,8 @@ async def resolve_credentials(self, subject: Subject, server: ServerSpec) -> Res
                 return _not_implemented(AuthSpecKind.passthrough)
             case ClientCredentialsConfig():
                 return _not_implemented(AuthSpecKind.client_credentials)
-            case TokenExchangeConfig():
-                return _not_implemented(AuthSpecKind.token_exchange)
+            case TokenExchangeConfig() as config:
+                return await self._token_exchange(subject, server, config)
             case AuthorizationCodeConfig():
                 return await self._authorization_code(subject, server)
             case AwsSigV4Config():
@@ -110,6 +129,29 @@ async def _authorization_code(self, subject: Subject, server: ServerSpec) -> Res
             return Error(CredError.of_unauthorized("Authorization required: complete the OAuth flow for this server."))
         return Ok(StaticHeaderAuth(f"Bearer {token.access_token}", header_name="Authorization"))
 
+    async def _token_exchange(
+        self, subject: Subject, server: ServerSpec, config: TokenExchangeConfig
+    ) -> Result[StaticHeaderAuth, CredError]:
+        """RFC 8693 OBO: exchange the caller's inbound token for an upstream-bound bearer.
+
+        No inbound token means there is nothing to exchange, so the arm fails closed with a 401 rather
+        than falling through to a weaker source (§1.5); the exchanger handles the IdP round-trip and
+        caching and returns the upstream token or a typed error.
+        """
+        inbound = subject.inbound_token
+        if inbound is None:
+            return Error(
+                CredError.of_unauthorized(
+                    "Token exchange requires a caller token to exchange (OBO).",
+                    www_authenticate='Bearer error="invalid_request"',
+                )
+            )
+        match await self._token_exchanger.exchange(inbound.get_secret_value(), server, config):
+            case Ok(token):
+                return Ok(StaticHeaderAuth(f"Bearer {token.access_token}", header_name="Authorization"))
+            case Error(err):
+                return Error(err)
+
     async def _authz_token(self, subject: Subject, server: ServerSpec) -> OAuthToken | None:
         """The user's authorization_code token, or None when absent or the store is unreachable.
 

litellm/proxy/_experimental/mcp_server/outbound_credentials/token_exchange_provider.py

@@ -0,0 +1,62 @@
+"""Composition root for the v2-native token_exchange (OBO) exchanger.
+
+Wires the pure ``Rfc8693TokenExchanger`` to its runtime edges: the real httpx POST against the IdP and
+the configured cache sizing/TTL constants. ``build_token_exchanger`` is built once at egress
+construction and reused, so the in-process exchanged-token cache survives across requests. Unlike the
+per-user store, nothing here reads a runtime global at build time (the httpx client is acquired per
+call), so it needs no lazy wrapper.
+"""
+
+from __future__ import annotations
+
+from litellm._logging import verbose_logger
+from litellm.constants import (
+    MCP_OAUTH2_TOKEN_CACHE_DEFAULT_TTL,
+    MCP_OAUTH2_TOKEN_CACHE_MIN_TTL,
+    MCP_OAUTH2_TOKEN_EXPIRY_BUFFER_SECONDS,
+    MCP_TOKEN_EXCHANGE_CACHE_MAX_SIZE,
+)
+from litellm.proxy._experimental.mcp_server.outbound_credentials.oauth_token_store import (
+    InMemoryTokenCacheBackend,
+)
+from litellm.proxy._experimental.mcp_server.outbound_credentials.token_exchanger import (
+    Rfc8693TokenExchanger,
+)
+
+
+async def _post_exchange_endpoint(
+    url: str, form: dict[str, str], client_auth_headers: dict[str, str]
+) -> dict[str, object] | None:
+    from litellm.llms.custom_httpx.http_handler import (  # noqa: PLC0415
+        get_async_httpx_client,  # pyright: ignore
+    )
+    from litellm.types.llms.custom_http import httpxSpecialProvider  # noqa: PLC0415
+
+    # litellm's httpx handler and httpx.Response are only partially typed; the IdP returns a JSON
+    # object and the exchanger validates each field, so the untyped boundary is contained here.
+    # A failed exchange is a miss, not a 500 (matches v1), so any error becomes None.
+    headers = {"Accept": "application/json", **client_auth_headers}
+    try:
+        client = get_async_httpx_client(llm_provider=httpxSpecialProvider.MCP)  # pyright: ignore
+        response = await client.post(url, headers=headers, data=form)  # pyright: ignore
+        response.raise_for_status()  # pyright: ignore
+        parsed: object = response.json()  # pyright: ignore
+    except Exception as exc:  # noqa: BLE001
+        verbose_logger.warning("MCP token exchange request failed: %s", exc)
+        return None
+    if not isinstance(parsed, dict):
+        # A valid-but-non-object JSON body (list/string/number) would crash the field parsing; map it
+        # to a miss so it surfaces as a typed upstream_unavailable, not a 500.
+        verbose_logger.warning("MCP token exchange returned non-object JSON (%s)", type(parsed).__name__)
+        return None
+    return parsed  # pyright: ignore
+
+
+def build_token_exchanger() -> Rfc8693TokenExchanger:
+    return Rfc8693TokenExchanger(
+        _post_exchange_endpoint,
+        cache=InMemoryTokenCacheBackend(max_size=MCP_TOKEN_EXCHANGE_CACHE_MAX_SIZE),
+        default_ttl_seconds=MCP_OAUTH2_TOKEN_CACHE_DEFAULT_TTL,
+        min_ttl_seconds=MCP_OAUTH2_TOKEN_CACHE_MIN_TTL,
+        expiry_buffer_seconds=MCP_OAUTH2_TOKEN_EXPIRY_BUFFER_SECONDS,
+    )