fix(mcp): restore upstream-token admission for legacy delegate servers

Servers with auth_type oauth2 and delegate_auth_to_upstream true accept a bare upstream token again, as they did through v1.102. #40923 started validating that token as a LiteLLM key, so clients that send only the upstream token got a 401

A bearer LiteLLM can prove is its own (the master key, a sealed gateway bearer, or an sk- key the key table holds) still takes normal admission and is never forwarded. Any other bearer is forwarded as the upstream token. When JWT or custom auth also proves the caller and that user is granted the server, the call is attributed to them, and otherwise the caller is admitted anonymously and logged without attribution

The anonymous server list and the upstream token preflight include these servers again, and the startup warning says the flag still works but is deprecated

litellm/proxy/mcp_server/AGENTS.md

@@ -68,8 +68,11 @@ module materially harder to understand.
   behind a single generic branch unless tests prove every mode still behaves
   correctly.
 - Be especially careful with legacy `delegate_auth_to_upstream: true`. `auth_type: oauth2`
-  with `delegate_auth_to_upstream: true` is deprecated: LiteLLM admission is required
-  for matching MCP routes. Use `auth_type: oauth_delegate` for client-forwarded OAuth.
+  with `delegate_auth_to_upstream: true` is deprecated but still supported. A bare
+  `Authorization` that is a LiteLLM credential is validated and never forwarded. Any other
+  bearer is the upstream token: it is forwarded, and the caller is admitted as the JWT or
+  custom-auth user it also proves when that user is granted the server, else anonymously.
+  New servers should use `auth_type: true_passthrough` or `auth_type: oauth_delegate`.
   OAuth discovery endpoints stay public so clients can start the RFC 9728 flow.
 - Keep database-backed fields in sync across migrations, typed models under
   `litellm/types/mcp.py` or `litellm/types/mcp_server/`, config loading, this

litellm/proxy/mcp_server/auth/user_api_key_auth_mcp.py

@@ -100,6 +100,7 @@ if TYPE_CHECKING:
 
 _EMPTY_TOOLSET_GRANTS: Final[Mapping[str, Sequence[str]]] = MappingProxyType({})
 OPTIONAL_STRING_ADAPTER: Final[TypeAdapter[str | None]] = TypeAdapter(str | None)
+_PROXY_SETTINGS_ADAPTER: Final = TypeAdapter(Mapping[str, object])
 _CACHED_VALUE: Final = TypeAdapter(object)
 
 
@@ -233,6 +234,74 @@ def _is_legacy_delegate_cold_start(mcp_servers: list[str] | None, client_ip: str
     return True
 
 
+def _caused_by(exc: BaseException | None, error_type: type[BaseException], depth: int = 16) -> bool:
+    if exc is None or depth == 0:
+        return False
+    return isinstance(exc, error_type) or _caused_by(exc.__cause__ or exc.__context__, error_type, depth - 1)
+
+
+async def _bearer_is_litellm_credential(token: str) -> bool:
+    """True when ``token`` is, or may be, a credential LiteLLM issued: the master key, a sealed gateway
+    bearer, or a virtual key the key table holds (a lookup fault counts as held). Only ``sk-`` tokens are
+    looked up, because no other token can be a virtual key."""
+    from litellm.proxy.mcp_server.bridge_token_flow import (  # noqa: PLC0415  # bridge flow imports this module
+        is_sealed_gateway_bearer,
+        virtual_key_may_exist,
+    )
+    from litellm.proxy.mcp_server.mcp_server_manager import (  # noqa: PLC0415  # manager imports this module
+        LITELLM_VIRTUAL_KEY_PREFIX,
+    )
+    from litellm.proxy.proxy_server import master_key  # noqa: PLC0415  # startup owns the master key
+
+    if master_key and secrets.compare_digest(token.encode(), master_key.encode()):
+        return True
+    if is_sealed_gateway_bearer(token):
+        return True
+    return token.startswith(LITELLM_VIRTUAL_KEY_PREFIX) and await virtual_key_may_exist(token)
+
+
+async def _upstream_bearer_identity(token: str, litellm_api_key: str, request: Request) -> UserAPIKeyAuth | None:
+    """The LiteLLM user an upstream bearer also proves, or ``None`` when it proves none.
+
+    Custom auth decides alone, so any rejection is ``None``. A JWT is tried only when JWT auth is on, its
+    issuer may be LiteLLM's, and its signature verifies. Past that, only a user LiteLLM does not know is
+    ``None``; every other refusal (budget, blocked team, disallowed route) propagates."""
+    from litellm.proxy.auth.handle_jwt import JWTHandler  # noqa: PLC0415  # proxy import cycle
+    from litellm.proxy.mcp_server.bridge_token_flow import (  # noqa: PLC0415  # bridge flow imports this module
+        custom_auth_configured,
+        jwt_issuer_may_be_litellm,
+    )
+    from litellm.proxy.proxy_server import (  # noqa: PLC0415  # startup owns the active auth configuration
+        general_settings,  # pyright: ignore[reportUnknownVariableType]  # legacy global is declared as a bare dict
+        jwt_handler,
+        premium_user,
+    )
+
+    if custom_auth_configured():
+        try:
+            return await user_api_key_auth(api_key=litellm_api_key, request=request)
+        except (HTTPException, ProxyException):
+            return None
+    settings: Final = _PROXY_SETTINGS_ADAPTER.validate_python(general_settings)
+    if (
+        settings.get("enable_jwt_auth") is not True
+        or premium_user is not True
+        or not JWTHandler.is_jwt(token=token)
+        or not jwt_issuer_may_be_litellm(token)
+    ):
+        return None
+    try:
+        await jwt_handler.auth_jwt(token=token)  # pyright: ignore[reportUnknownMemberType]  # returns a bare dict; only whether it raises matters
+    except Exception:  # noqa: BLE001  # any verification failure means LiteLLM does not vouch for this bearer
+        return None
+    try:
+        return await user_api_key_auth(api_key=litellm_api_key, request=request)
+    except (HTTPException, ProxyException) as exc:
+        if _caused_by(exc, UserNotFoundError):
+            return None
+        raise
+
+
 def _is_litellm_auth_admission_error(exc: Exception) -> bool:
     if isinstance(exc, HTTPException):
         return exc.status_code == 401
@@ -704,6 +773,19 @@ class MCPRequestHandler:
                     route=request_route,
                     mcp_servers=mcp_servers,
                 )
+            elif oauth2_headers and (
+                legacy_delegate_targets := MCPRequestHandler._legacy_delegate_targets(
+                    path=request_route,
+                    mcp_servers=mcp_servers,
+                    client_ip=IPAddressUtils.get_mcp_client_ip(request),
+                )
+            ):
+                validated_user_api_key_auth = await MCPRequestHandler._admit_legacy_delegate_bearer(  # rebind-ok: one assignment per admission arm, like its siblings
+                    authorization_value=oauth2_headers["Authorization"],
+                    litellm_api_key=litellm_api_key,
+                    targets=legacy_delegate_targets,
+                    request=request,
+                )
             elif oauth2_headers:
                 # Authorization on a non-delegated server: the bearer must be a real
                 # LiteLLM credential, so a failed validation is a genuine 401/403 and
@@ -741,10 +823,14 @@ class MCPRequestHandler:
             custom_key_header_name: Final = OPTIONAL_STRING_ADAPTER.validate_python(
                 general_settings.get("litellm_key_header_name")
             )
-            admitted_credential: Final = MCPRequestHandler.caller_admission_credential(
-                headers,
-                validated_user_api_key_auth,
-                custom_key_header_name=custom_key_header_name,
+            admitted_credential: Final = (
+                None
+                if validated_user_api_key_auth.mcp_forwards_caller_authorization
+                else MCPRequestHandler.caller_admission_credential(
+                    headers,
+                    validated_user_api_key_auth,
+                    custom_key_header_name=custom_key_header_name,
+                )
             )
             (
                 oauth2_headers,
@@ -934,6 +1020,53 @@ class MCPRequestHandler:
                 return False
         return True
 
+    @staticmethod
+    def _legacy_delegate_targets(
+        path: str, mcp_servers: Sequence[str] | None, client_ip: str | None
+    ) -> tuple[MCPServer, ...]:
+        """Every server the request targets when ALL of them are legacy ``delegate_auth_to_upstream``
+        servers, else ``()``. Names resolve the way downstream routing resolves them, so a permissive
+        ``x-mcp-servers`` header cannot open the gate for a stricter path target."""
+        from litellm.proxy.mcp_server.mcp_server_manager import (  # noqa: PLC0415  # manager imports this module
+            global_mcp_server_manager,
+            is_legacy_upstream_delegate,
+        )
+
+        servers: Final = tuple(
+            global_mcp_server_manager.get_mcp_server_by_name(name, client_ip=client_ip)
+            for name in MCPRequestHandler._resolve_target_server_names(path=path, mcp_servers_header=mcp_servers)
+        )
+        delegates: Final = tuple(
+            server for server in servers if server is not None and is_legacy_upstream_delegate(server)
+        )
+        return delegates if servers and len(delegates) == len(servers) else ()
+
+    @staticmethod
+    async def _admit_legacy_delegate_bearer(
+        authorization_value: str,
+        litellm_api_key: str,
+        targets: tuple[MCPServer, ...],
+        request: Request,
+    ) -> UserAPIKeyAuth:
+        """Admit a bare ``Authorization`` bearer sent to legacy ``delegate_auth_to_upstream`` servers.
+
+        A LiteLLM credential runs normal admission, and its errors propagate. Any other bearer is the
+        upstream token and is marked for forwarding: a user it also proves who is granted every target is
+        admitted as that user, and every other caller is admitted anonymously."""
+        from litellm.proxy.mcp_server.mcp_server_manager import (  # noqa: PLC0415  # manager imports this module
+            global_mcp_server_manager,
+        )
+
+        token: Final = _normalize_caller_admission_credential(authorization_value)
+        if await _bearer_is_litellm_credential(token):
+            return await user_api_key_auth(api_key=litellm_api_key, request=request)
+        identity: Final = await _upstream_bearer_identity(token, litellm_api_key, request)
+        if identity is not None:
+            allowed: Final = frozenset(await global_mcp_server_manager.get_allowed_mcp_servers(identity))
+            if all(server.server_id in allowed for server in targets):
+                return identity.model_copy(update={"mcp_forwards_caller_authorization": True})
+        return UserAPIKeyAuth().model_copy(update={"mcp_forwards_caller_authorization": True})
+
     @staticmethod
     def _single_dcr_bridge_delegate_target(
         path: str, mcp_servers: list[str] | None, client_ip: str | None
@@ -1553,7 +1686,7 @@ class MCPRequestHandler:
         return expiry >= datetime.now(timezone.utc)
 
     @staticmethod
-    def _resolve_target_server_names(path: str, mcp_servers_header: list[str] | None) -> list[str]:
+    def _resolve_target_server_names(path: str, mcp_servers_header: Sequence[str] | None) -> Sequence[str]:
         """
         Resolve the target MCP server names exactly as downstream routing
         does (``server.py::extract_mcp_auth_context``).

litellm/proxy/mcp_server/bridge_token_flow.py

@@ -61,7 +61,6 @@ async def oauth_authorization_uses_gateway_credential(request: Request) -> bool:
     from litellm.proxy.proxy_server import (  # noqa: PLC0415  # startup owns the active auth configuration
         jwt_handler,
         master_key,
-        user_custom_auth,
     )
 
     if "x-litellm-api-key" in request.headers:
@@ -71,10 +70,26 @@ async def oauth_authorization_uses_gateway_credential(request: Request) -> bool:
         return "authorization" in request.headers
     if token.startswith("sk-") or (master_key and secrets.compare_digest(token.encode(), master_key.encode())):
         return True
-    if user_custom_auth is not None or jwt_handler.litellm_jwtauth.oidc_userinfo_enabled:
+    if custom_auth_configured() or jwt_handler.litellm_jwtauth.oidc_userinfo_enabled:
         return True
     if not JWTHandler.is_jwt(token):
-        return await _opaque_bearer_is_gateway_credential(token)
+        return is_sealed_gateway_bearer(token) or await virtual_key_may_exist(token)
+    return jwt_issuer_may_be_litellm(token)
+
+
+def custom_auth_configured() -> bool:
+    from litellm.proxy.proxy_server import (  # noqa: PLC0415  # startup owns the active auth configuration
+        user_custom_auth,
+    )
+
+    return user_custom_auth is not None
+
+
+def jwt_issuer_may_be_litellm(token: str) -> bool:
+    """False only when the JWT names an issuer LiteLLM's JWT auth is not configured to accept."""
+    from litellm.proxy.auth.handle_jwt import JWTHandler  # noqa: PLC0415  # proxy import cycle
+    from litellm.proxy.proxy_server import jwt_handler  # noqa: PLC0415  # startup owns the active auth configuration
+
     claims: Final = JWTHandler.get_unverified_claims(token)
     issuer: Final = claims.get("iss") if claims is not None else None
     global_issuer: Final = os.getenv("JWT_ISSUER")
@@ -86,25 +101,35 @@ async def oauth_authorization_uses_gateway_credential(request: Request) -> bool:
     )
 
 
-async def _opaque_bearer_is_gateway_credential(token: str) -> bool:
-    from litellm.proxy._types import hash_token  # noqa: PLC0415  # proxy import cycle
+def is_sealed_gateway_bearer(token: str) -> bool:
+    """True for a bearer only the gateway issues: a bridge or refresh envelope, a v2 GCM blob, or a
+    dashboard UI token. A UI token that cannot be read still counts, so a lookup fault never reads as foreign."""
     from litellm.proxy.auth.auth_checks import ExperimentalUIJWTToken  # noqa: PLC0415  # proxy import cycle
-    from litellm.proxy.auth.resolvers.exceptions import KeyNotFoundError  # noqa: PLC0415  # proxy import cycle
-    from litellm.proxy.auth.resolvers.store import IdentityStore  # noqa: PLC0415  # proxy import cycle
     from litellm.proxy.mcp_server.outbound_credentials.envelope import (  # noqa: PLC0415  # envelope imports bridge types
         is_envelope,
         is_refresh_envelope,
     )
-    from litellm.proxy.proxy_server import (  # noqa: PLC0415  # startup owns the identity store dependencies
-        prisma_client,
-        user_api_key_cache,
-    )
 
     if is_envelope(token) or is_refresh_envelope(token) or token.startswith(V2_GCM_PREFIX):
         return True
     try:
-        if ExperimentalUIJWTToken.get_key_object_from_ui_hash_key(token) is not None:
-            return True
+        return ExperimentalUIJWTToken.get_key_object_from_ui_hash_key(token) is not None
+    except Exception as exc:  # noqa: BLE001  # an identity lookup fault must not permit cookie fallback
+        verbose_logger.debug("OAuth bearer ownership could not be checked (%s)", type(exc).__name__)
+        return True
+
+
+async def virtual_key_may_exist(token: str) -> bool:
+    """False only when the key table answers that ``token`` is not a virtual key; a lookup fault counts as held."""
+    from litellm.proxy._types import hash_token  # noqa: PLC0415  # proxy import cycle
+    from litellm.proxy.auth.resolvers.exceptions import KeyNotFoundError  # noqa: PLC0415  # proxy import cycle
+    from litellm.proxy.auth.resolvers.store import IdentityStore  # noqa: PLC0415  # proxy import cycle
+    from litellm.proxy.proxy_server import (  # noqa: PLC0415  # startup owns the identity store dependencies
+        prisma_client,
+        user_api_key_cache,
+    )
+
+    try:
         await IdentityStore(prisma_client, user_api_key_cache).resolve(hashed_token=hash_token(token))
     except KeyNotFoundError:
         return False

litellm/proxy/mcp_server/mcp_server_manager.py

@@ -1196,8 +1196,11 @@ def should_strip_caller_authorization(
       pass-through cold-start case (RFC 9728) the bearer in
       ``Authorization`` is the upstream OAuth token and must be
       forwarded, so we keep it.
-    - **Delegated OAuth servers**: admission always runs and there is no
-      anonymous path, so the caller's separate ``Authorization`` is
+    - **Legacy delegate servers**: forward when MCP admission marked the
+      caller's ``Authorization`` as the upstream token
+      (``mcp_forwards_caller_authorization``); otherwise follow the
+      delegated OAuth rule below.
+    - **Delegated OAuth servers**: the caller's separate ``Authorization`` is
       forwarded only when a distinct ``x-litellm-api-key`` carried
       admission. Without that header the ``Authorization`` *was* the
       admission credential — a virtual key, an IdP JWT, or an SSO / OIDC /
@@ -1206,6 +1209,12 @@ def should_strip_caller_authorization(
     - **Every other auth-enabled mode**: strip when ``Authorization`` is the
       credential admission validated.
     """
+    if (
+        user_api_key_auth is not None
+        and user_api_key_auth.mcp_forwards_caller_authorization
+        and is_legacy_upstream_delegate(mcp_server)
+    ):
+        return False
     if _authorization_is_admission_jwt(raw_headers, user_api_key_auth):
         return True
     if mcp_server.auth_type == MCPAuth.oauth2_token_exchange:
@@ -1557,6 +1566,18 @@ async def _materialize_auth_headers(auth: httpx2.Auth | None) -> dict[str, str]
     return {header_name: header_value} if header_value else None
 
 
+def is_legacy_upstream_delegate(server: MCPServer) -> bool:
+    """True for an interactive ``auth_type=oauth2`` server with ``delegate_auth_to_upstream=true``, whose
+    caller ``Authorization`` is the upstream token. The flow is resolved, not read from the column, so an
+    unstamped M2M-shaped row stays excluded: its stored client credentials drive egress, and admitting it
+    anonymously would let any caller act as LiteLLM's service account."""
+    return (
+        server.auth_type == MCPAuth.oauth2
+        and server.delegate_auth_to_upstream is True
+        and MCPServerManager.effective_oauth2_flow(server) != "client_credentials"
+    )
+
+
 def _consumes_caller_authorization(server: MCPServer) -> bool:
     """True when this server's egress forwards the caller's request-wide ``Authorization`` upstream:
     the client-forwarded token modes, legacy OAuth pass-through, and legacy upstream-delegated
@@ -1726,18 +1747,15 @@ def _warn_overlapping_tool_policies(server: MCPServer) -> None:
 
 
 def _warn_legacy_delegate_auth_if_applicable(server: MCPServer, *, source: str) -> None:
-    """Direct legacy delegated OAuth configurations to the admitted replacement."""
-    if server.auth_type != MCPAuth.oauth2:
-        return
-    if getattr(server, "delegate_auth_to_upstream", False) is not True:
-        return
-    if server.has_client_credentials:
+    """Point legacy delegated OAuth configurations at the client-forwarded replacements."""
+    if not is_legacy_upstream_delegate(server):
         return
     label: Final = get_server_prefix(server)
     verbose_logger.warning(
         "MCP server %r (id=%s, source=%s) uses deprecated auth_type=oauth2 with "
-        "delegate_auth_to_upstream=true. LiteLLM admission is now required; migrate to "
-        "auth_type=oauth_delegate for client-forwarded OAuth.",
+        "delegate_auth_to_upstream=true. It still works: a caller without a LiteLLM credential is "
+        "admitted anonymously and its upstream token is forwarded. Migrate to auth_type=true_passthrough "
+        "or auth_type=oauth_delegate.",
         label,
         server.server_id,
         source,
@@ -3766,12 +3784,12 @@ class MCPServerManager:
                 )
             )
             if is_anonymous:
-                passthrough_server_ids: Final = [
+                anonymous_server_ids: Final = [
                     server.server_id
                     for server in self.get_registry().values()
-                    if server.auth_type == MCPAuth.true_passthrough
+                    if server.auth_type == MCPAuth.true_passthrough or is_legacy_upstream_delegate(server)
                 ]
-                combined_servers.update(passthrough_server_ids)
+                combined_servers.update(anonymous_server_ids)
 
             restrict_allow_all: Final = (
                 resolved_general_settings.get("mcp_allow_all_keys_respects_mcp_scope", False)

litellm/proxy/mcp_server/server.py

@@ -588,6 +588,7 @@ if MCP_AVAILABLE:
     from litellm.proxy.mcp_server.mcp_server_manager import (
         MCPServerManager,
         global_mcp_server_manager,
+        is_legacy_upstream_delegate,
     )
 
     ######################################################
@@ -2163,7 +2164,7 @@ if MCP_AVAILABLE:
         mcp_server_auth_headers: dict[str, dict[str, str]] | None,
         raw_headers: Mapping[str, str] | None,
     ) -> None:
-        """Probe pass-through upstream servers in parallel before the MCP session starts.
+        """Probe pass-through and legacy delegate upstream servers in parallel before the MCP session starts.
 
         Only servers the caller's key is already authorized to reach are probed —
         the list is derived from _get_allowed_mcp_servers so that a user cannot
@@ -2186,6 +2187,9 @@ if MCP_AVAILABLE:
             mcp_servers=mcp_servers,
             client_ip=client_ip,
         )
+        # A legacy delegate is probed only on a single-server route, so one rejected token cannot 401 a
+        # multi-server connect; its challenge echoes the requested name, matching the tokenless challenge.
+        single_requested_name: Final = mcp_servers[0] if mcp_servers is not None and len(mcp_servers) == 1 else None
         prepared_headers: Final = tuple(
             (
                 srv,
@@ -2200,7 +2204,7 @@ if MCP_AVAILABLE:
                 ),
             )
             for srv in allowed_servers
-            if srv.is_oauth_passthrough
+            if srv.is_oauth_passthrough or (single_requested_name is not None and is_legacy_upstream_delegate(srv))
         )
         probe_headers: Final = tuple(
             (
@@ -2213,7 +2217,7 @@ if MCP_AVAILABLE:
             for srv, (auth, extra) in prepared_headers
         )
         probe_targets: Final[tuple[tuple[MCPServer, str, str], ...]] = tuple(
-            (srv, auth_header, srv.name)
+            (srv, auth_header, srv.name if srv.is_oauth_passthrough else single_requested_name or srv.name)
             for srv, headers in probe_headers
             if (
                 auth_header := next(

litellm/types/mcp_server/mcp_server_manager.py

@@ -204,9 +204,11 @@ class MCPServer(LiteLLMBaseModel):
     access_groups: list[str] | None = None
     allow_all_keys: bool = False
     available_on_public_internet: bool = True
-    # Legacy opt-in to upstream-delegated authentication for ``oauth2``
-    # servers. LiteLLM admission still applies; use ``oauth_delegate`` for the
-    # supported client-forwarded OAuth flow.
+    # Deprecated opt-in to upstream-delegated authentication for ``oauth2``
+    # servers. A bare ``Authorization`` that is not a LiteLLM credential is the
+    # upstream token and is forwarded; see
+    # ``MCPRequestHandler._admit_legacy_delegate_bearer``. New servers should use
+    # ``true_passthrough`` or ``oauth_delegate``.
     #
     # Honored only for ``auth_type == oauth2``; ignored for any other
     # ``auth_type``. OAuth pass-through for non-oauth2 servers

litellm/types/proxy/auth/user_api_key_auth.py

@@ -169,6 +169,10 @@ class UserAPIKeyAuth(LiteLLM_VerificationTokenView):  # the expected response ob
     # above; a forged value could at most narrow, but the stripping keeps the field's provenance
     # single-owner so its meaning stays trustworthy.
     mcp_session_resource_server_ids: tuple[str, ...] | None = Field(default=None, exclude=True)
+    # Set only by MCP admission when a legacy delegate_auth_to_upstream caller's Authorization is the
+    # upstream token rather than a LiteLLM credential, so egress forwards it. Stripped from validated
+    # input like the markers above: a forged value would forward a LiteLLM credential upstream.
+    mcp_forwards_caller_authorization: bool = Field(default=False, exclude=True)
     mcp_toolset_id: str | None = Field(default=None, exclude=True)
     authenticated_by_custom_auth: bool = Field(default=False, exclude=True)
     via_virtual_key: bool = Field(
@@ -230,6 +234,7 @@ class UserAPIKeyAuth(LiteLLM_VerificationTokenView):  # the expected response ob
         values.pop("mcp_explicit_grants_only", None)
         values.pop("mcp_source_team_rpm_limits", None)
         values.pop("mcp_session_resource_server_ids", None)
+        values.pop("mcp_forwards_caller_authorization", None)
         values.pop("mcp_toolset_id", None)
         values.pop("via_virtual_key", None)
         values.pop("authenticated_by_custom_auth", None)

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

@@ -1663,10 +1663,10 @@ class TestMCPOAuth2AuthFlow:
     as LiteLLM API keys, causing auth failures and empty tool listings.
     """
 
-    async def test_oauth2_token_in_authorization_header_requires_litellm_admission(self):
+    async def test_oauth2_upstream_token_on_legacy_delegate_is_admitted_anonymously(self):
         """
-        A bare Authorization token on the legacy delegated mode must establish
-        a LiteLLM principal rather than entering anonymously.
+        A bare upstream OAuth token on the legacy delegated mode is admitted
+        anonymously without LiteLLM key validation and kept for egress.
         """
         from litellm.types.mcp import MCPAuth
 
@@ -1702,10 +1702,11 @@ class TestMCPOAuth2AuthFlow:
                 raw_headers,
             ) = await MCPRequestHandler.process_mcp_request(scope)
 
-            assert isinstance(auth_result, UserAPIKeyAuth)
-            assert auth_result.user_id == "admitted-user"
-            mock_auth.assert_awaited_once()
-            assert oauth2_headers.get("Authorization") == "Bearer atlassian-oauth2-access-token-xyz"
+            assert auth_result.user_id is None
+            assert auth_result.api_key is None
+            assert auth_result.mcp_forwards_caller_authorization is True
+            mock_auth.assert_not_awaited()
+            assert oauth2_headers == {"Authorization": "Bearer atlassian-oauth2-access-token-xyz"}
 
     async def test_explicit_litellm_key_with_oauth2_authorization(self):
         """
@@ -2545,9 +2546,10 @@ class TestMCPDelegateAuthToUpstream:
     """
     Tests for the ``delegate_auth_to_upstream`` per-server flag.
 
-    The legacy flag no longer bypasses LiteLLM admission. OAuth discovery may
-    still use the anonymous cold-start challenge, but a presented bearer must
-    authenticate to LiteLLM unless a separate admission credential is supplied.
+    A bare bearer that is a LiteLLM credential takes normal admission and is
+    never forwarded. Any other bearer is the upstream token: it is forwarded,
+    and the caller is admitted as the JWT or custom-auth user it also proves
+    when that user is granted the server, else anonymously.
     """
 
     @staticmethod
@@ -2652,37 +2654,237 @@ class TestMCPDelegateAuthToUpstream:
             assert isinstance(auth_result, UserAPIKeyAuth)
             mock_auth.assert_awaited_once()
 
-    async def test_delegate_with_only_upstream_token_requires_litellm_auth(self):
-        """
-        An upstream token cannot establish a LiteLLM principal and must not
-        reopen anonymous admission.
-        """
-        from litellm.types.mcp import MCPAuth
-
-        scope = {
+    @staticmethod
+    def _bearer_scope(token: str) -> dict:
+        return {
             "type": "http",
             "method": "POST",
             "path": "/mcp/delegated_oauth_server",
-            "headers": [(b"authorization", b"Bearer upstream-pkce-token")],
+            "headers": [(b"authorization", f"Bearer {token}".encode())],
         }
 
+    @staticmethod
+    def _jwt(issuer: str = "https://idp.example") -> str:
+        import jwt
+
+        return jwt.encode({"iss": issuer, "sub": "jwt-user"}, "test-signing-secret", algorithm="HS256")
+
+    @staticmethod
+    @contextlib.contextmanager
+    def _delegate_admission(admission: AsyncMock, allowed_server_ids: Sequence[str] = ("test-server-id",)):
+        from litellm.types.mcp import MCPAuth
+
         with (
-            patch(
-                "litellm.proxy.mcp_server.auth.user_api_key_auth_mcp.user_api_key_auth",
-                new_callable=AsyncMock,
-                side_effect=HTTPException(status_code=401, detail="Invalid API key"),
-            ) as mock_auth,
+            patch("litellm.proxy.mcp_server.auth.user_api_key_auth_mcp.user_api_key_auth", admission),
             patch("litellm.proxy.mcp_server.mcp_server_manager.global_mcp_server_manager") as mock_mgr,
         ):
             mock_mgr.get_mcp_server_by_name.return_value = TestMCPDelegateAuthToUpstream._make_server(
                 auth_type=MCPAuth.oauth2,
                 delegate_auth_to_upstream=True,
             )
-            with pytest.raises(HTTPException) as exc_info:
-                await MCPRequestHandler.process_mcp_request(scope)
+            mock_mgr.get_allowed_mcp_servers = AsyncMock(return_value=list(allowed_server_ids))
+            yield
 
-            assert exc_info.value.status_code == 401
-            mock_auth.assert_awaited_once()
+    @staticmethod
+    @contextlib.contextmanager
+    def _jwt_auth_enabled(auth_jwt: AsyncMock):
+        from litellm.proxy import proxy_server
+        from litellm.proxy._types import LiteLLM_JWTAuth
+
+        with (
+            patch.dict(proxy_server.general_settings, {"enable_jwt_auth": True}),
+            patch.object(proxy_server, "premium_user", True),
+            patch.object(proxy_server, "user_custom_auth", None),
+            patch.object(proxy_server.jwt_handler, "auth_jwt", auth_jwt),
+            patch.object(proxy_server.jwt_handler, "litellm_jwtauth", LiteLLM_JWTAuth(), create=True),
+            patch.dict(os.environ),
+        ):
+            os.environ.pop("JWT_ISSUER", None)
+            yield
+
+    async def test_delegate_forwards_unknown_sk_token_after_key_table_miss(self):
+        """An sk- token the key table does not hold, such as a provider API key, belongs to the upstream:
+        it is forwarded anonymously rather than rejected for its prefix."""
+        from litellm.proxy.auth.resolvers.exceptions import KeyNotFoundError
+
+        admission = AsyncMock()
+        with (
+            TestMCPDelegateAuthToUpstream._delegate_admission(admission),
+            patch(
+                "litellm.proxy.auth.resolvers.store.IdentityStore.resolve",
+                new_callable=AsyncMock,
+                side_effect=KeyNotFoundError("hashed-token"),
+            ),
+        ):
+            auth_result, _, _, _, oauth2_headers, _ = await MCPRequestHandler.process_mcp_request(
+                TestMCPDelegateAuthToUpstream._bearer_scope("sk-proj-upstream-provider-key")
+            )
+
+        assert auth_result.api_key is None
+        assert auth_result.mcp_forwards_caller_authorization is True
+        assert oauth2_headers == {"Authorization": "Bearer sk-proj-upstream-provider-key"}
+        admission.assert_not_awaited()
+
+    @pytest.mark.parametrize(
+        "lookup_effect",
+        [None, RuntimeError("key table unreachable")],
+        ids=["key-held", "lookup-fault"],
+    )
+    async def test_delegate_litellm_key_refusal_propagates(self, lookup_effect):
+        """A key the table holds, or one the lookup could not check, takes normal admission, so its
+        refusal reaches the caller instead of becoming an anonymous request that forwards the key."""
+        refusal = ProxyException(message="Budget has been exceeded", type="budget_exceeded", param=None, code=400)
+        with (
+            TestMCPDelegateAuthToUpstream._delegate_admission(AsyncMock(side_effect=refusal)),
+            patch(
+                "litellm.proxy.auth.resolvers.store.IdentityStore.resolve",
+                new_callable=AsyncMock,
+                side_effect=lookup_effect,
+            ),
+        ):
+            with pytest.raises(ProxyException) as exc_info:
+                await MCPRequestHandler.process_mcp_request(TestMCPDelegateAuthToUpstream._bearer_scope("sk-held-key"))
+
+        assert exc_info.value is refusal
+
+    @staticmethod
+    def _unknown_user_refusal() -> ProxyException:
+        from litellm.types.proxy.auth.auth_checks import UserNotFoundError
+
+        refusal = ProxyException(message="User doesn't exist in db", type="auth_error", param=None, code=401)
+        refusal.__cause__ = UserNotFoundError(user_id="jwt-user")
+        return refusal
+
+    @pytest.mark.parametrize(
+        "auth_jwt_effect, admission, allowed_server_ids, expected_user_id",
+        [
+            (
+                None,
+                lambda: AsyncMock(return_value=UserAPIKeyAuth(user_id="jwt-user", jwt_claims={"sub": "jwt-user"})),
+                ["test-server-id"],
+                "jwt-user",
+            ),
+            (
+                None,
+                lambda: AsyncMock(return_value=UserAPIKeyAuth(user_id="jwt-user", jwt_claims={"sub": "jwt-user"})),
+                [],
+                None,
+            ),
+            (
+                None,
+                lambda: AsyncMock(side_effect=TestMCPDelegateAuthToUpstream._unknown_user_refusal()),
+                ["test-server-id"],
+                None,
+            ),
+            (
+                ValueError("Signature verification failed"),
+                lambda: AsyncMock(return_value=UserAPIKeyAuth(user_id="jwt-user")),
+                ["test-server-id"],
+                None,
+            ),
+        ],
+        ids=["granted-user", "user-not-granted", "user-unknown-to-litellm", "signature-rejected"],
+    )
+    async def test_delegate_jwt_is_forwarded_as_granted_user_or_anonymously(
+        self, auth_jwt_effect, admission, allowed_server_ids, expected_user_id
+    ):
+        """A JWT LiteLLM also accepts admits that user when granted the server; a user who is not
+        granted, unknown to LiteLLM, or whose signature fails is admitted anonymously. The JWT is always
+        forwarded, because on this server it is the upstream's token."""
+        token = TestMCPDelegateAuthToUpstream._jwt()
+        with (
+            TestMCPDelegateAuthToUpstream._jwt_auth_enabled(AsyncMock(side_effect=auth_jwt_effect)),
+            TestMCPDelegateAuthToUpstream._delegate_admission(admission(), allowed_server_ids),
+        ):
+            auth_result, _, _, _, oauth2_headers, _ = await MCPRequestHandler.process_mcp_request(
+                TestMCPDelegateAuthToUpstream._bearer_scope(token)
+            )
+
+        assert auth_result.user_id == expected_user_id
+        assert auth_result.mcp_forwards_caller_authorization is True
+        assert oauth2_headers == {"Authorization": f"Bearer {token}"}
+
+    @pytest.mark.parametrize(
+        "refusal",
+        [
+            ProxyException(message="Budget has been exceeded", type="budget_exceeded", param=None, code=400),
+            ProxyException(message="Team=t1 is blocked", type="auth_error", param=None, code=401),
+        ],
+        ids=["over-budget", "blocked-team"],
+    )
+    async def test_delegate_jwt_refusal_after_valid_signature_propagates(self, refusal):
+        """Once the signature verifies, LiteLLM knows who the caller is, so a budget or an explicit
+        disallowment is enforced and the request fails instead of continuing anonymously."""
+        with (
+            TestMCPDelegateAuthToUpstream._jwt_auth_enabled(AsyncMock(return_value={"sub": "jwt-user"})),
+            TestMCPDelegateAuthToUpstream._delegate_admission(AsyncMock(side_effect=refusal)),
+        ):
+            with pytest.raises(ProxyException) as exc_info:
+                await MCPRequestHandler.process_mcp_request(
+                    TestMCPDelegateAuthToUpstream._bearer_scope(TestMCPDelegateAuthToUpstream._jwt())
+                )
+
+        assert exc_info.value is refusal
+
+    async def test_delegate_jwt_from_unconfigured_issuer_is_forwarded_without_jwt_auth(self):
+        """A JWT from an issuer LiteLLM's JWT auth is not configured for never reaches JWT auth."""
+        auth_jwt = AsyncMock(return_value={"sub": "jwt-user"})
+        admission = AsyncMock(return_value=UserAPIKeyAuth(user_id="jwt-user"))
+        with (
+            TestMCPDelegateAuthToUpstream._jwt_auth_enabled(auth_jwt),
+            patch.dict(os.environ, {"JWT_ISSUER": "https://litellm-idp.example"}),
+            TestMCPDelegateAuthToUpstream._delegate_admission(admission),
+        ):
+            auth_result, *_ = await MCPRequestHandler.process_mcp_request(
+                TestMCPDelegateAuthToUpstream._bearer_scope(
+                    TestMCPDelegateAuthToUpstream._jwt(issuer="https://upstream-idp.example")
+                )
+            )
+
+        assert auth_result.user_id is None
+        assert auth_result.mcp_forwards_caller_authorization is True
+        auth_jwt.assert_not_awaited()
+
+    @pytest.mark.parametrize(
+        "admission, expected_user_id",
+        [
+            (
+                lambda: AsyncMock(return_value=UserAPIKeyAuth(user_id="custom-user", api_key="upstream-custom-token")),
+                "custom-user",
+            ),
+            (
+                lambda: AsyncMock(
+                    side_effect=ProxyException(message="custom auth rejected", type="auth_error", param=None, code=401)
+                ),
+                None,
+            ),
+        ],
+        ids=["accepted", "rejected"],
+    )
+    async def test_delegate_custom_auth_decides_alone(self, admission, expected_user_id):
+        """With custom auth configured, its answer alone decides: accepted admits that user, any failure
+        admits anonymously. The bearer is forwarded either way."""
+        from litellm.proxy import proxy_server
+
+        with (
+            patch.object(proxy_server, "user_custom_auth", AsyncMock()),
+            TestMCPDelegateAuthToUpstream._delegate_admission(admission()),
+        ):
+            auth_result, _, _, _, oauth2_headers, _ = await MCPRequestHandler.process_mcp_request(
+                TestMCPDelegateAuthToUpstream._bearer_scope("upstream-custom-token")
+            )
+
+        assert auth_result.user_id == expected_user_id
+        assert auth_result.mcp_forwards_caller_authorization is True
+        assert oauth2_headers == {"Authorization": "Bearer upstream-custom-token"}
+
+    def test_forwarding_marker_cannot_be_forged_from_input(self):
+        """Only admission may mark a caller's Authorization for forwarding; caller-shaped input never can."""
+        assert UserAPIKeyAuth(mcp_forwards_caller_authorization=True).mcp_forwards_caller_authorization is False
+        assert (
+            UserAPIKeyAuth.model_validate({"mcp_forwards_caller_authorization": True}).mcp_forwards_caller_authorization
+            is False
+        )
 
     async def test_delegate_off_still_requires_litellm_auth(self):
         """
@@ -2753,7 +2955,12 @@ class TestMCPDelegateAuthToUpstream:
                 await MCPRequestHandler.process_mcp_request(scope)
             assert exc_info.value.status_code == 401
 
-    async def test_delegate_mixed_targets_fail_closed(self):
+    @pytest.mark.parametrize(
+        "bearer_headers",
+        [[], [(b"authorization", b"Bearer upstream-pkce-token")]],
+        ids=["no-bearer", "upstream-bearer"],
+    )
+    async def test_delegate_mixed_targets_fail_closed(self, bearer_headers):
         """
         x-mcp-servers can list multiple targets. If ANY of them does not opt in
         to delegate_auth_to_upstream, the bypass must NOT fire — otherwise an
@@ -2769,6 +2976,7 @@ class TestMCPDelegateAuthToUpstream:
             "path": "/mcp",
             "headers": [
                 (b"x-mcp-servers", b"delegated_oauth,plain_oauth"),
+                *bearer_headers,
             ],
         }
 
@@ -2863,9 +3071,8 @@ class TestMCPDelegateAuthToUpstream:
 
     async def test_authorization_bearer_on_delegate_server_establishes_litellm_principal(self):
         """
-        A bare Authorization bearer now follows normal LiteLLM admission. A
-        separate x-litellm-api-key is required when Authorization is intended
-        for the upstream server.
+        A bare sk- bearer the key table may hold takes normal LiteLLM admission
+        and is never marked for forwarding upstream.
         """
         from litellm.types.mcp import MCPAuth
 
@@ -2898,6 +3105,7 @@ class TestMCPDelegateAuthToUpstream:
             ) = await MCPRequestHandler.process_mcp_request(scope)
             assert isinstance(auth_result, UserAPIKeyAuth)
             assert auth_result.user_id == "real-user"
+            assert auth_result.mcp_forwards_caller_authorization is False
             assert oauth2_headers.get("Authorization") == "Bearer sk-9876"
             mock_auth.assert_awaited_once()
 
@@ -3079,8 +3287,9 @@ class TestMCPDelegateAuthToUpstream:
             mock_auth.assert_awaited_once()
             assert auth.api_key is None
 
-    async def test_get_allowed_servers_excludes_legacy_delegates(self):
-        """Legacy delegated servers are never added to anonymous access."""
+    async def test_get_allowed_servers_includes_interactive_legacy_delegates_for_anonymous(self):
+        """Anonymous callers reach interactive legacy delegates, never an M2M server whose stored
+        client credentials would let them act as LiteLLM's service account."""
         from litellm.proxy.mcp_server.mcp_server_manager import (
             MCPServerManager,
         )
@@ -3118,7 +3327,7 @@ class TestMCPDelegateAuthToUpstream:
         ):
             result = await manager.get_allowed_mcp_servers(None)
 
-        assert result == []
+        assert result == ["pkce-server"]
 
     async def test_true_passthrough_skips_litellm_auth_anonymously(self):
         """auth_type=true_passthrough performs no admission auth: the caller's Authorization is an

tests/unit/proxy/mcp_server/test_mcp_server_tool_calls_and_headers.py

@@ -669,6 +669,69 @@ def test_prepare_mcp_server_headers_legacy_delegate_strips_repeated_admission_ke
     assert extra_headers is None
 
 
+@pytest.mark.parametrize(
+    "caller_fields",
+    [{}, {"user_id": "jwt-user", "api_key": "upstream-jwt", "jwt_claims": {"sub": "jwt-user"}}],
+    ids=["anonymous", "admitted-jwt-user"],
+)
+def test_prepare_mcp_server_headers_legacy_delegate_forwards_marked_upstream_authorization(caller_fields):
+    """When admission marked the caller's Authorization as the upstream token, a legacy delegate forwards
+    it, including a JWT that also admitted the caller and would otherwise be stripped as an admission JWT."""
+    from litellm.proxy._types import UserAPIKeyAuth
+    from litellm.proxy.mcp_server.server import (
+        _prepare_mcp_server_headers,
+    )
+
+    server = MCPServer(
+        server_id="legacy-delegate-marked",
+        name="legacy-delegate",
+        transport=MCPTransport.http,
+        auth_type=MCPAuth.oauth2,
+        delegate_auth_to_upstream=True,
+    )
+    caller = UserAPIKeyAuth(**caller_fields).model_copy(update={"mcp_forwards_caller_authorization": True})
+
+    server_auth_header, extra_headers = _prepare_mcp_server_headers(
+        server=server,
+        mcp_server_auth_headers=None,
+        mcp_auth_header=None,
+        oauth2_headers={"Authorization": "Bearer upstream-jwt"},
+        raw_headers={"authorization": "Bearer upstream-jwt"},
+        user_api_key_auth=caller,
+    )
+
+    assert server_auth_header is None
+    assert extra_headers == {"Authorization": "Bearer upstream-jwt"}
+
+
+def test_prepare_mcp_server_headers_forwarding_marker_is_ignored_off_legacy_delegates():
+    """The forwarding marker only applies to legacy delegates, so it can never carry a LiteLLM key to an
+    oauth_delegate upstream."""
+    from litellm.proxy._types import UserAPIKeyAuth
+    from litellm.proxy.mcp_server.server import (
+        _prepare_mcp_server_headers,
+    )
+
+    server = MCPServer(
+        server_id="oauth-delegate-marked",
+        name="oauth-delegate",
+        transport=MCPTransport.http,
+        auth_type=MCPAuth.oauth_delegate,
+    )
+    caller = UserAPIKeyAuth(api_key="sk-litellm-key").model_copy(update={"mcp_forwards_caller_authorization": True})
+
+    _, extra_headers = _prepare_mcp_server_headers(
+        server=server,
+        mcp_server_auth_headers=None,
+        mcp_auth_header=None,
+        oauth2_headers={"Authorization": "Bearer sk-litellm-key"},
+        raw_headers={"authorization": "Bearer sk-litellm-key"},
+        user_api_key_auth=caller,
+    )
+
+    assert extra_headers is None
+
+
 def test_prepare_mcp_server_headers_m2m_skips_authorization_from_raw_extra_headers():
     """M2M must not merge caller Authorization from raw_headers when extra_headers lists it."""
     try:
@@ -7265,8 +7328,9 @@ async def test_legacy_delegate_bare_token_is_not_probed_upstream():  # test-qual
 
 
 @pytest.mark.asyncio
-async def test_legacy_delegate_dual_credentials_are_not_probed_upstream():  # test-quality-ok: this removed security-sensitive egress has no return value; non-invocation is the contract
-    """A separate upstream bearer never triggers the removed legacy probe."""
+async def test_legacy_delegate_dual_credentials_probe_only_the_upstream_token():  # test-quality-ok: the credential sent to the upstream probe is the security contract and has no return value
+    """With a separate x-litellm-api-key, the preflight probes the upstream with its own token and never
+    with the LiteLLM key."""
     from litellm.proxy._types import UserAPIKeyAuth
     from litellm.proxy.mcp_server.server import (
         _check_passthrough_upstream_auth,
@@ -7285,9 +7349,9 @@ async def test_legacy_delegate_dual_credentials_are_not_probed_upstream():  # te
             "litellm.proxy.mcp_server.operations._get_allowed_mcp_servers",
             new=AsyncMock(return_value=[server]),
         ),
-        patch(  # test-quality-ok: the removed probe call is the security regression under test
+        patch(  # test-quality-ok: the probe's outbound credential is the security contract under test
             "litellm.proxy.mcp_server.server._probe_upstream_auth",
-            new=AsyncMock(),
+            new=AsyncMock(return_value=(200, None)),
         ) as probe,
     ):
         await _check_passthrough_upstream_auth(
@@ -7300,7 +7364,59 @@ async def test_legacy_delegate_dual_credentials_are_not_probed_upstream():  # te
             raw_headers={"x-litellm-api-key": "sk-litellm-proxy-key", "Authorization": "Bearer upstream-token"},
         )
 
-    probe.assert_not_awaited()
+    assert probe.await_args.args == ("http://upstream:9401/mcp", "Bearer upstream-token")
+
+
+def _expired_delegate_token_preflight(mcp_servers: list[str]):
+    from litellm.proxy._types import UserAPIKeyAuth
+    from litellm.proxy.mcp_server.server import (
+        _check_passthrough_upstream_auth,
+    )
+
+    return _check_passthrough_upstream_auth(
+        scope=_delegate_scope([(b"authorization", b"Bearer expired-upstream-token")]),
+        user_api_key_auth=UserAPIKeyAuth().model_copy(update={"mcp_forwards_caller_authorization": True}),
+        mcp_servers=mcp_servers,
+        client_ip=None,
+        oauth2_headers={"Authorization": "Bearer expired-upstream-token"},
+        mcp_server_auth_headers=None,
+        raw_headers={"authorization": "Bearer expired-upstream-token"},
+    )
+
+
+def _patch_rejecting_delegate_upstream():
+    return (
+        patch(  # test-quality-ok: isolate authorized-server resolution so this test targets the preflight boundary
+            "litellm.proxy.mcp_server.operations._get_allowed_mcp_servers",
+            new=AsyncMock(return_value=[_delegate_auth_mcp_server()]),
+        ),
+        patch(  # test-quality-ok: stands in for the upstream rejecting an expired token
+            "litellm.proxy.mcp_server.server._probe_upstream_auth",
+            new=AsyncMock(return_value=(401, 'Bearer error="invalid_token"')),
+        ),
+    )
+
+
+@pytest.mark.asyncio
+async def test_legacy_delegate_expired_upstream_token_gets_reauthorization_challenge():
+    """An expired upstream token on a single-server route gets the RFC 9728 re-authorization challenge
+    for the requested name, so the client signs in again instead of seeing an empty tool list."""
+    allowed_servers, probe = _patch_rejecting_delegate_upstream()
+    with allowed_servers, probe, pytest.raises(HTTPException) as exc_info:
+        await _expired_delegate_token_preflight(["delegate_test"])
+
+    assert exc_info.value.status_code == 401
+    challenge = exc_info.value.headers["www-authenticate"]
+    assert 'error="invalid_token"' in challenge
+    assert 'resource_metadata="http://localhost:4000/.well-known/oauth-protected-resource/mcp/delegate_test"' in challenge
+
+
+@pytest.mark.asyncio
+async def test_legacy_delegate_multi_server_scope_is_not_failed_by_one_upstream_token():
+    """A multi-server scope is not probed, so one rejected upstream token cannot fail the whole connect."""
+    allowed_servers, probe = _patch_rejecting_delegate_upstream()
+    with allowed_servers, probe:
+        assert await _expired_delegate_token_preflight(["delegate_test", "other_server"]) is None
 
 
 @pytest.mark.parametrize(