fix(mcp): accept validated JWT subjects after virtual-key mapping (#46549)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py

@@ -694,6 +694,7 @@ class MCPRequestHandler:
                 mcp_auth_header=mcp_auth_header,
                 mcp_server_auth_headers=mcp_server_auth_headers,
                 admitted_credential=admitted_credential,
+                preserve_jwt_subject=bool(validated_user_api_key_auth.jwt_claims),
             )
 
             return (
@@ -733,6 +734,7 @@ class MCPRequestHandler:
         mcp_server_auth_headers: dict[str, dict[str, str]] | None,
         *,
         admitted_credential: str | None,
+        preserve_jwt_subject: bool = False,
     ) -> tuple[dict[str, str] | None, dict[str, str], str | None, dict[str, dict[str, str]] | None]:
         """Remove gateway-shaped credentials and the caller's admitted credential from EVERY egress
         header context: top-level ``Authorization`` (oauth2 + raw), the deprecated ``x-mcp-auth``, and
@@ -746,6 +748,14 @@ class MCPRequestHandler:
 
         # 1. Top-level Authorization → oauth2_headers.
         authz: Final = oauth2_headers.get("Authorization") if oauth2_headers else None
+        preserve_subject: Final = bool(
+            preserve_jwt_subject
+            and not admitted
+            and authz
+            and not _normalize_caller_admission_credential(authz).startswith("sk-")
+            and not MCPRequestHandler._is_gateway_admission_credential(authz)
+            and MCPRequestHandler._is_caller_admission_key(authz, admitted_credential)
+        )
         if admitted or cred(authz):
             oauth2_headers = None
 
@@ -754,7 +764,9 @@ class MCPRequestHandler:
         raw_headers = {
             k: v
             for k, v in raw_headers.items()
-            if k.lower() == "x-litellm-api-key" or not ((admitted and k.lower() == "authorization") or cred(v))
+            if k.lower() == "x-litellm-api-key"
+            or (preserve_subject and k.lower() == "authorization")
+            or not ((admitted and k.lower() == "authorization") or cred(v))
         }
 
         # 3. Deprecated x-mcp-auth value.

litellm/proxy/_experimental/mcp_server/exceptions.py

@@ -6,6 +6,11 @@ from fastapi import HTTPException
 
 from litellm.proxy._experimental.mcp_server.oauth_utils import encode_upstream_scope
 
+TOKEN_EXCHANGE_SUBJECT_ERROR: Final = (
+    "Missing or invalid subject token. Authenticate with your identity provider and send its token "
+    "in the Authorization Bearer header. LiteLLM API keys and Admin UI sessions cannot be used for token exchange."
+)
+
 
 class MCPServerURLCredentialsError(HTTPException):
     """A fixed, sanitized URL-credential migration error safe for operator previews."""
@@ -41,12 +46,14 @@ class MCPUpstreamAuthError(Exception):
         server_name: str,
         required_scope: str | None = None,
         server_id: str | None = None,
+        public_detail: object = None,
     ) -> None:
         self.status_code = status_code
         self.www_authenticate = www_authenticate
         self.server_name = server_name
         self.required_scope = required_scope
         self.server_id = server_id
+        self.public_detail = TOKEN_EXCHANGE_SUBJECT_ERROR if public_detail == TOKEN_EXCHANGE_SUBJECT_ERROR else None
         super().__init__(f"Upstream MCP server {server_name!r} returned {status_code}")
 
     def to_http_exception(
@@ -107,7 +114,7 @@ class MCPUpstreamAuthError(Exception):
             if self.required_scope is not None
             else challenge
         )
-        detail: Final = "Forbidden" if self.status_code == 403 else "Unauthorized"
+        detail: Final = self.public_detail or ("Forbidden" if self.status_code == 403 else "Unauthorized")
         return HTTPException(
             status_code=self.status_code,
             detail=detail,

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -1194,6 +1194,8 @@ def should_strip_caller_authorization(
       session token whose ``api_key`` is ``None`` — and must never reach
       the upstream, so it is stripped regardless of the ``api_key`` value.
     """
+    if _authorization_is_admission_jwt(raw_headers, user_api_key_auth):
+        return True
     if mcp_server.auth_type == MCPAuth.oauth2_token_exchange:
         # OBO: the inbound Authorization is the subject token. It is exchanged at the IdP and only the
         # exchanged token is sent upstream, so the raw caller bearer must never be forwarded.
@@ -1253,6 +1255,16 @@ def _authorization_is_litellm_admission_credential(
     return bool(user_api_key_auth and user_api_key_auth.api_key and not admission_header)
 
 
+def _authorization_is_admission_jwt(
+    raw_headers: Mapping[str, str] | None, user_api_key_auth: UserAPIKeyAuth | None
+) -> bool:
+    return bool(
+        user_api_key_auth
+        and user_api_key_auth.jwt_claims
+        and _authorization_is_litellm_admission_credential(raw_headers, user_api_key_auth)
+    )
+
+
 def _server_auth_header_for(
     server: MCPServer,
     mcp_server_auth_headers: Mapping[str, str | dict[str, str]] | None,
@@ -4001,10 +4013,12 @@ class MCPServerManager:
     ) -> str | None:
         """The caller's upstream identity token, or ``None`` when the bearer is a LiteLLM key.
 
-        Rejects the key admission validated and, because virtual keys always carry the ``sk-`` prefix,
-        any other LiteLLM key a client puts in ``Authorization`` next to ``x-litellm-api-key``.
+        Validated JWT claims survive virtual-key mapping, so JWT admission still supplies a subject.
+        Virtual keys are rejected even when the auth context carries JWT claims.
         """
-        if _authorization_is_litellm_admission_credential(raw_headers, user_api_key_auth):
+        if _authorization_is_litellm_admission_credential(raw_headers, user_api_key_auth) and not (
+            user_api_key_auth and user_api_key_auth.jwt_claims
+        ):
             return None
         bearer: Final = MCPServerManager._extract_bearer_token(oauth2_headers, raw_headers)
         if bearer is not None and bearer.startswith(LITELLM_VIRTUAL_KEY_PREFIX):
@@ -4252,7 +4266,7 @@ class MCPServerManager:
         subject_token: Final = self.extract_subject_token(oauth2_headers, raw_headers, user_api_key_auth)
         match server.auth_type:
             case MCPAuth.oauth2_token_exchange:
-                if not self._extract_bearer_token(oauth2_headers, None):
+                if not self._extract_bearer_token(oauth2_headers, raw_headers):
                     return
             case MCPAuth.oauth2_id_jag:
                 if subject_token is not None:
@@ -4552,6 +4566,7 @@ class MCPServerManager:
                     status_code=e.status_code,
                     www_authenticate=None if server.is_dcr_bridge else challenge_header,
                     server_name=server.name,
+                    public_detail=e.detail if server.auth_type == MCPAuth.oauth2_token_exchange else None,
                 ) from e
             verbose_logger.warning("Failed to get tools from server %s: %s", server.name, e)
             raise MCPServerListError(ServerListFault(tag="internal", status_code=e.status_code), server.name) from e

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

@@ -21,7 +21,7 @@ from pydantic import SecretStr
 from typing_extensions import assert_never
 
 from litellm.experimental_mcp_client.client import MCPClient, strip_auth_scheme, to_basic_credentials
-from litellm.proxy._experimental.mcp_server.exceptions import MCPServerURLCredentialsError
+from litellm.proxy._experimental.mcp_server.exceptions import TOKEN_EXCHANGE_SUBJECT_ERROR, MCPServerURLCredentialsError
 from litellm.proxy._experimental.mcp_server.oauth_utils import resolve_upstream_resource
 from litellm.proxy._experimental.mcp_server.outbound_credentials.result import Error, Ok, Result
 from litellm.proxy._experimental.mcp_server.outbound_credentials.types import (
@@ -402,7 +402,7 @@ def raise_token_exchange_challenge(
     )
     raise HTTPException(
         status_code=401,
-        detail="Unauthorized",
+        detail="Unauthorized" if encoded_claims else TOKEN_EXCHANGE_SUBJECT_ERROR,
         headers={"WWW-Authenticate": www_authenticate},
     )
 

litellm/proxy/_experimental/mcp_server/rest_endpoints.py

@@ -623,6 +623,7 @@ if MCP_AVAILABLE:
         ) = MCPRequestHandler.scrub_gateway_admission_credentials(
             admitted=False,
             admitted_credential=admitted_credential,
+            preserve_jwt_subject=bool(user_api_key_dict.jwt_claims),
             oauth2_headers=oauth2_headers_from_request,
             raw_headers=dict(headers),
             mcp_auth_header=MCPRequestHandler.get_mcp_auth_header_from_headers(headers),

litellm/proxy/_experimental/mcp_server/server.py

@@ -1933,7 +1933,15 @@ if MCP_AVAILABLE:
             # so the client discovers the IdP, SSOs, and retries with a subject token, which LiteLLM
             # then exchanges. A tool-call-time 401 would be wrapped into a JSON-RPC error and the
             # header lost, so the discovery flow needs this pre-emptive challenge.
-            if server and server.auth_type == MCPAuth.oauth2_token_exchange and not oauth2_headers:
+            if (
+                server
+                and server.auth_type == MCPAuth.oauth2_token_exchange
+                and not oauth2_headers
+                and operations.global_mcp_server_manager.extract_subject_token(
+                    oauth2_headers, raw_headers, user_api_key_auth
+                )
+                is None
+            ):
                 from litellm.proxy._experimental.mcp_server.outbound_credentials.adapter import (  # noqa: PLC0415  # lazy: adapter pulls MCP subgraph
                     raise_token_exchange_challenge,
                 )

tests/integration/mcp/test_mcp_oauth_flows.py

@@ -250,6 +250,121 @@ def _assert_subject_token_challenge(response: httpx.Response, alias: str) -> Non
     challenge: Final = response.headers["www-authenticate"]
     assert challenge.startswith("Bearer ") and 'error="invalid_token"' in challenge, challenge
     assert f'resource_metadata="/.well-known/oauth-protected-resource/mcp/{alias}"' in challenge, challenge
+    assert "Authenticate with your identity provider" in response.json()["detail"], response.text
+
+
+def test_jwt_mapped_to_a_virtual_key_is_exchanged_for_tools_list_and_call(gateway: Gateway, tmp_path: Path) -> None:
+    private_key: Final = generate_private_key(public_exponent=65537, key_size=2048)
+    public_jwk: Final = TypeAdapter(dict[str, object]).validate_json(
+        jwt.algorithms.RSAAlgorithm.to_jwk(private_key.public_key())
+    )
+    kid: Final = "mapped-obo-key"
+
+    def jwks_response(request: Request) -> Reply:
+        assert request.method == "GET"
+        return Reply(body=json.dumps({"keys": [{**public_jwk, "kid": kid}]}).encode())
+
+    config: Final = _proxy_config_with_general_settings(
+        tmp_path,
+        "mapped_obo.yaml",
+        {
+            "enable_jwt_auth": True,
+            "litellm_jwtauth": {"user_id_jwt_field": "sub", "virtual_key_claim_field": "sub"},
+        },
+    )
+    with (
+        wire_server(jwks_response) as jwks,
+        oauth_server() as auth,
+        mcp_peer() as peer,
+        owned_proxy(gateway, tmp_path, {"JWT_PUBLIC_KEY_URL": jwks.url}, config=config) as candidate,
+        candidate.scenario() as scenario,
+    ):
+        alias: Final = "mappedobo" + uuid.uuid4().hex[:8]
+        identity: Final = register_mcp(
+            scenario,
+            peer,
+            alias,
+            auth_type="oauth2_token_exchange",
+            token_exchange_endpoint=auth.issuer + "/token",
+            credentials={"client_id": "te-client", "client_secret": "te-secret"},
+        )
+        plain_alias: Final = alias + "plain"
+        plain_identity: Final = register_mcp(
+            scenario, peer, plain_alias, auth_type="none", extra_headers=["authorization"]
+        )
+        key: Final = scenario.key(object_permission={"mcp_servers": [identity, plain_identity]})
+        user_id: Final = "mapped-user-" + uuid.uuid4().hex
+        mapping: Final = candidate.post(
+            "/jwt/key/mapping/new", {"jwt_claim_name": "sub", "jwt_claim_value": user_id, "key": key}
+        )
+        scenario.cleanups.callback(candidate.post, "/jwt/key/mapping/delete", {"id": mapping["id"]})
+        subject: Final = _signed_subject_token(private_key, {"sub": user_id}, kid)
+        peer.drain()
+        auth.drain()
+        listing: Final = candidate.request("GET", "/mcp-rest/tools/list", key=subject, params={"server_id": identity})
+        assert listing.status_code == 200, listing.text
+        assert "add" in tuple(tool["name"] for tool in listing.json()["tools"]), listing.text
+        exchanged: Final = auth.token_requests()
+        assert len(exchanged) == 1, exchanged
+        assert exchanged[0]["subject_token"] == subject
+        called: Final = candidate.request(
+            "POST",
+            "/mcp-rest/tools/call",
+            {"server_id": identity, "name": f"{alias}-add", "arguments": ADD},
+            key=subject,
+        )
+        assert called.status_code == 200, called.text
+        assert called.json()["isError"] is False, called.text
+        calls: Final = tool_calls(peer.drain())
+        assert len(calls) == 1, calls
+        authorization: Final = calls[0]["headers"][b"authorization"]
+        assert subject.encode() not in authorization and key.encode() not in authorization
+        assert auth.is_live(authorization.decode().removeprefix("Bearer "))
+        native: Final = McpCaller(
+            candidate, None, "server_mcp", alias=alias, headers={"Authorization": f"Bearer {subject}"}
+        )
+        native_listing: Final = native.list_tools()
+        assert native_listing.status == 200 and native_listing.error is None, native_listing
+        assert f"{alias}-add" in native_listing.tools, native_listing
+        native_called: Final = native.call(f"{alias}-add", ADD)
+        assert native_called.status == 200 and native_called.error is None, native_called
+        native_calls: Final = tool_calls(peer.drain())
+        assert len(native_calls) == 1, native_calls
+        assert native_calls[0]["headers"][b"authorization"] == authorization
+        plain: Final = McpCaller(
+            candidate, None, "server_mcp", alias=plain_alias, headers={"Authorization": f"Bearer {subject}"}
+        )
+        plain_listing: Final = plain.list_tools()
+        assert plain_listing.status == 200 and plain_listing.error is None, plain_listing
+        assert f"{plain_alias}-add" in plain_listing.tools, plain_listing
+        plain_called: Final = plain.call(f"{plain_alias}-add", ADD)
+        assert plain_called.status == 200 and plain_called.error is None, plain_called
+        plain_calls: Final = tool_calls(peer.drain())
+        assert len(plain_calls) == 1, plain_calls
+        assert b"authorization" not in plain_calls[0]["headers"], plain_calls
+        aggregate: Final = McpCaller(
+            candidate,
+            key,
+            "server_mcp",
+            alias=f"{alias},{plain_alias}",
+            headers={"Authorization": "Bearer sk-not-the-admission-key"},
+        ).list_tools()
+        assert aggregate.status == 200 and aggregate.error is None, aggregate
+        assert f"{plain_alias}-add" in aggregate.tools, aggregate
+        assert f"{alias}-add" not in aggregate.tools, aggregate
+        auth.drain()
+        for headers in (
+            {"Authorization": f"Bearer {key}"},
+            {"Authorization": f"Bearer {key}", "x-litellm-api-key": key},
+        ):
+            rejected: Final = candidate.client.post(
+                "/mcp-rest/tools/call",
+                headers=headers,
+                json={"name": f"{alias}-add", "arguments": ADD, "server_id": identity},
+            )
+            _assert_subject_token_challenge(rejected, alias)
+            assert auth.token_requests() == ()
+            assert tool_calls(peer.drain()) == ()
 
 
 @pytest.mark.parametrize("entry", ENTRY_POINTS)

tests/unit/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py

@@ -9722,6 +9722,26 @@ class TestSessionBearerEgressScrub:
         )
         assert result == (None, {empty_header: "", "x-tenant": "tenant-control"}, None, {})
 
+    @pytest.mark.parametrize("claims", [None, {"sub": "caller"}])
+    @pytest.mark.parametrize("bearer", ["validated-jwt", "sk-admission-key"])
+    async def test_scrubbing_preserves_only_validated_jwt_admission(
+        self, claims: dict[str, str] | None, bearer: str
+    ) -> None:
+        headers: Final = Headers({"Authorization": f"Bearer {bearer}"})
+        auth: Final = UserAPIKeyAuth(api_key="stored-key-hash", jwt_claims=claims)
+        credential: Final = MCPRequestHandler.caller_admission_credential(headers, auth, custom_key_header_name=None)
+        result: Final = MCPRequestHandler.scrub_gateway_admission_credentials(
+            admitted=False,
+            oauth2_headers={"Authorization": f"Bearer {bearer}"},
+            raw_headers={**dict(headers), "x-mcp-echo-authorization": f"Bearer {bearer}"},
+            mcp_auth_header=bearer,
+            mcp_server_auth_headers={"echo": {"Authorization": f"Bearer {bearer}"}},
+            admitted_credential=credential,
+            preserve_jwt_subject=bool(claims),
+        )
+        expected: Final = {"Authorization": f"Bearer {bearer}"} if claims and not bearer.startswith("sk-") else None
+        assert result == (None, dict(headers) if expected else {}, None, {})
+
     async def test_caller_admission_credential_prefers_x_litellm_header(self) -> None:
         caller_key: Final = "sk-caller-admission-key-123"
         upstream_authorization: Final = "Bearer unrelated-upstream-token"

tests/unit/proxy/_experimental/mcp_server/outbound_credentials/test_adapter.py

@@ -613,6 +613,10 @@ def test_raise_token_exchange_challenge_is_rfc9728_invalid_token():
     assert 'resource_metadata="/.well-known/oauth-protected-resource/mcp/obo-srv"' in www
     assert 'error="invalid_token"' in www
     assert "error_description=" in www
+    assert exc.detail == (
+        "Missing or invalid subject token. Authenticate with your identity provider and send its token "
+        "in the Authorization Bearer header. LiteLLM API keys and Admin UI sessions cannot be used for token exchange."
+    )
 
 
 def test_raise_token_exchange_challenge_includes_server_root_path(monkeypatch):

tests/unit/proxy/_experimental/mcp_server/test_mcp_oauth_passthrough_tools.py

@@ -15,7 +15,11 @@ if sys.version_info < (3, 11):
     from exceptiongroup import ExceptionGroup
 
 
-from litellm.proxy._experimental.mcp_server.exceptions import MCPServerListError, MCPUpstreamAuthError
+from litellm.proxy._experimental.mcp_server.exceptions import (
+    TOKEN_EXCHANGE_SUBJECT_ERROR,
+    MCPServerListError,
+    MCPUpstreamAuthError,
+)
 from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
     MCPServerManager,
     _extract_upstream_auth_failure,
@@ -216,15 +220,23 @@ async def test_fetch_tools_from_passthrough_returns_tools_on_success():
     assert tools == [tool]
 
 
-def test_to_http_exception_preserves_upstream_www_authenticate():
+@pytest.mark.parametrize("status_code", [401, 403])
+@pytest.mark.parametrize("public_detail", [None, TOKEN_EXCHANGE_SUBJECT_ERROR, "private failure", {"secret": "private"}])
+def test_to_http_exception_preserves_upstream_www_authenticate(public_detail: object, status_code: int):
     err = MCPUpstreamAuthError(
-        status_code=401,
+        status_code=status_code,
         www_authenticate='Bearer resource_metadata="https://upstream/.well-known/oauth-protected-resource"',
         server_name="sample_docs",
+        public_detail=public_detail,
     )
 
     http_exc = err.to_http_exception()
-    assert http_exc.status_code == 401
+    assert http_exc.status_code == status_code
+    assert http_exc.detail == (
+        TOKEN_EXCHANGE_SUBJECT_ERROR
+        if public_detail == TOKEN_EXCHANGE_SUBJECT_ERROR
+        else "Forbidden" if status_code == 403 else "Unauthorized"
+    )
     assert http_exc.headers == {
         "www-authenticate": 'Bearer resource_metadata="https://upstream/.well-known/oauth-protected-resource"'
     }

tests/unit/proxy/_experimental/mcp_server/test_mcp_server_manager.py

@@ -258,6 +258,51 @@ def restore_mcp_manager_singleton():
         module.global_mcp_server_manager = manager
 
 
+@pytest.mark.parametrize(
+    "auth_type",
+    [
+        MCPAuth.none,
+        MCPAuth.api_key,
+        MCPAuth.bearer_token,
+        MCPAuth.basic,
+        MCPAuth.oauth_delegate,
+        MCPAuth.true_passthrough,
+    ],
+)
+@pytest.mark.parametrize("explicit_key", [False, True])
+def test_validated_admission_jwt_is_not_forwarded_as_an_extra_header(auth_type: MCPAuth, explicit_key: bool) -> None:
+    from litellm.proxy._experimental.mcp_server.mcp_server_manager import _openapi_forwarded_extra_headers
+    from litellm.proxy._experimental.mcp_server.operations import _prepare_mcp_server_headers
+
+    server: Final = MCPServer(
+        server_id="header-boundary",
+        name="header-boundary",
+        transport=MCPTransport.http,
+        auth_type=auth_type,
+        extra_headers=["Authorization", "x-tenant"],
+    )
+    raw: Final = {
+        "Authorization": "Bearer validated-jwt",
+        "x-tenant": "tenant-control",
+        **({"x-litellm-api-key": "sk-admission"} if explicit_key else {}),
+    }
+    user: Final = UserAPIKeyAuth(api_key="stored-key-hash", jwt_claims={"sub": "caller"})
+    expected: Final = {
+        "x-tenant": "tenant-control",
+        **({"Authorization": "Bearer validated-jwt"} if explicit_key else {}),
+    }
+    _, upstream_headers = _prepare_mcp_server_headers(
+        server,
+        mcp_auth_header=None,
+        mcp_server_auth_headers=None,
+        oauth2_headers=None,
+        raw_headers=raw,
+        user_api_key_auth=user,
+    )
+    assert upstream_headers == expected
+    assert _openapi_forwarded_extra_headers(server, raw, user) == expected
+
+
 class TestMCPServerManager:
     """Test MCP Server Manager stdio functionality"""
 
@@ -14746,6 +14791,27 @@ class TestLitellmAdmissionKeyIsNeverTheSubjectToken:
     _ADMISSION_KEY: Final = "sk-litellm-virtual-key"
     _USER_TOKEN: Final = "user-idp-jwt"
 
+    @pytest.mark.parametrize("explicit_admission", [None, "", "Bearer mapped-jwt"])
+    @pytest.mark.parametrize("oauth_headers", [None, {"Authorization": "Bearer mapped-jwt"}])
+    def test_validated_jwt_remains_the_subject_after_virtual_key_mapping(
+        self, explicit_admission: str | None, oauth_headers: dict[str, str] | None
+    ) -> None:
+        auth: Final = UserAPIKeyAuth(api_key="hashed-mapped-key", user_id="caller", jwt_claims={"sub": "caller"})
+        headers: Final = {
+            "Authorization": "Bearer mapped-jwt",
+            **({"x-litellm-api-key": explicit_admission} if explicit_admission is not None else {}),
+        }
+        assert MCPServerManager.extract_subject_token(oauth_headers, headers, auth) == "mapped-jwt"
+
+    @pytest.mark.parametrize("explicit_admission", [False, True])
+    def test_litellm_key_is_not_a_subject_even_with_jwt_claims(self, explicit_admission: bool) -> None:
+        auth: Final = UserAPIKeyAuth(api_key="hashed-key", jwt_claims={"sub": "caller"})
+        headers: Final = {
+            "Authorization": f"Bearer {self._ADMISSION_KEY}",
+            **({"x-litellm-api-key": self._ADMISSION_KEY} if explicit_admission else {}),
+        }
+        assert MCPServerManager.extract_subject_token(None, headers, auth) is None
+
     @staticmethod
     def _token_exchange_server(server_id: str) -> MCPServer:
         return MCPServer(