From ef77f1713aef9d6b8721428f4e4a2cda79d48c1e Mon Sep 17 00:00:00 2001
From: jesus <jesus@berri.ai>
Date: Fri, 18 Sep 2026 00:28:13 +0000
Subject: [PATCH 1/8] fix(mcp): gate preemptive OAuth 401 challenge on the
 key's allowed servers

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 .../proxy/_experimental/mcp_server/server.py  |  54 +++--
 .../mcp_server/test_mcp_server.py             | 189 ++++++++++++++++++
 2 files changed, 229 insertions(+), 14 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py
index ad886c66de7..8d6b8907fc9 100644
--- a/litellm/proxy/_experimental/mcp_server/server.py
+++ b/litellm/proxy/_experimental/mcp_server/server.py
@@ -14,6 +14,7 @@ import traceback
 import types
 import uuid
 from collections.abc import AsyncIterator, Callable, Mapping, Sequence
+from collections.abc import Set as AbstractSet
 from datetime import datetime
 from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol
 
@@ -4013,6 +4014,20 @@ if MCP_AVAILABLE:
             )
         return user_api_key_auth.model_copy(update={"object_permission": updated_op})
 
+    async def _challenge_allowed_server_ids(
+        user_api_key_auth: UserAPIKeyAuth | None,
+        mcp_servers: list[str] | None,
+        client_ip: str | None,
+        toolset_allowed_server_ids: AbstractSet[str] | None,
+    ) -> frozenset[str]:
+        allowed_servers: Final = await _get_allowed_mcp_servers(
+            user_api_key_auth=user_api_key_auth, mcp_servers=mcp_servers, client_ip=client_ip
+        )
+        allowed_ids: Final = frozenset(srv.server_id for srv in allowed_servers)
+        if toolset_allowed_server_ids is None:
+            return allowed_ids
+        return allowed_ids & frozenset(toolset_allowed_server_ids)
+
     async def _raise_preemptive_401_for_unauthenticated_servers(
         scope: Scope,
         mcp_servers: list[str] | None,
@@ -4020,7 +4035,7 @@ if MCP_AVAILABLE:
         mcp_server_auth_headers: dict[str, dict[str, str]] | None,
         user_api_key_auth: UserAPIKeyAuth | None,
         client_ip: str | None,
-        allowed_server_ids: set[str] | None = None,
+        allowed_server_ids: AbstractSet[str] | None = None,
         raw_headers: Mapping[str, str] | None = None,
     ) -> None:
         """Fail fast with HTTP 401 for MCP servers that need user auth but
@@ -4028,11 +4043,10 @@ if MCP_AVAILABLE:
         (points clients at the gateway AS metadata) and pass-through OAuth
         (points clients at the upstream resource-metadata via our well-known).
 
-        ``allowed_server_ids`` may be passed by callers that have already
-        narrowed the authorized server set (e.g. toolset scoping); servers
-        not in that set are skipped so a client targeting a toolset that
-        excludes a passthrough server is not pushed into an OAuth flow for
-        a server it will be 403'd on immediately after authentication.
+        ``allowed_server_ids`` is the set of servers the caller's key may
+        reach, narrowed further by any active toolset scope; servers not in
+        that set are skipped so a client is not pushed into an OAuth flow
+        for a server it will be 403'd on immediately after authentication.
         """
         for server_name in mcp_servers or []:
             server = global_mcp_server_manager.get_mcp_server_by_name(server_name, client_ip=client_ip)
@@ -4428,9 +4442,10 @@ if MCP_AVAILABLE:
 
             # https://datatracker.ietf.org/doc/html/rfc9728#name-www-authenticate-response
             # Must run after toolset scoping so the challenge set is derived
-            # from the fully-authorized server set: a passthrough server that
-            # the active toolset excludes should not trigger an OAuth flow
-            # for a server the caller will be 403'd on after authentication.
+            # from the fully-authorized server set (the key's allowed servers,
+            # narrowed by any active toolset scope): a server the key cannot
+            # reach should not trigger an OAuth flow for a server the caller
+            # will be 403'd on after authentication.
             await _raise_preemptive_401_for_unauthenticated_servers(
                 scope=scope,
                 mcp_servers=mcp_servers,
@@ -4438,7 +4453,12 @@ if MCP_AVAILABLE:
                 mcp_server_auth_headers=mcp_server_auth_headers,
                 user_api_key_auth=user_api_key_auth,
                 client_ip=_client_ip,
-                allowed_server_ids=toolset_allowed_server_ids,
+                allowed_server_ids=await _challenge_allowed_server_ids(
+                    user_api_key_auth=user_api_key_auth,
+                    mcp_servers=mcp_servers,
+                    client_ip=_client_ip,
+                    toolset_allowed_server_ids=toolset_allowed_server_ids,
+                ),
                 raw_headers=raw_headers,
             )
 
@@ -4756,9 +4776,10 @@ if MCP_AVAILABLE:
 
             # https://datatracker.ietf.org/doc/html/rfc9728#name-www-authenticate-response
             # Must run after toolset scoping so the challenge set is derived
-            # from the fully-authorized server set: a passthrough server that
-            # the active toolset excludes should not trigger an OAuth flow
-            # for a server the caller will be 403'd on after authentication.
+            # from the fully-authorized server set (the key's allowed servers,
+            # narrowed by any active toolset scope): a server the key cannot
+            # reach should not trigger an OAuth flow for a server the caller
+            # will be 403'd on after authentication.
             await _raise_preemptive_401_for_unauthenticated_servers(
                 scope=scope,
                 mcp_servers=mcp_servers,
@@ -4766,7 +4787,12 @@ if MCP_AVAILABLE:
                 mcp_server_auth_headers=mcp_server_auth_headers,
                 user_api_key_auth=user_api_key_auth,
                 client_ip=_sse_client_ip,
-                allowed_server_ids=toolset_allowed_server_ids,
+                allowed_server_ids=await _challenge_allowed_server_ids(
+                    user_api_key_auth=user_api_key_auth,
+                    mcp_servers=mcp_servers,
+                    client_ip=_sse_client_ip,
+                    toolset_allowed_server_ids=toolset_allowed_server_ids,
+                ),
                 raw_headers=raw_headers,
             )
 
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
index 02182ebbe60..55cab45169e 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
@@ -1,6 +1,8 @@
 import asyncio
+import contextlib
 import contextvars
 import os
+from collections.abc import Iterator
 from datetime import datetime, timedelta
 from types import SimpleNamespace
 from typing import Final
@@ -9,6 +11,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 from fastapi import HTTPException
 from mcp import ReadResourceResult, Resource
+from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.types import (
     BlobResourceContents,
     CallToolResult,
@@ -8651,6 +8654,192 @@ class TestPreemptive401ModeAware:
         await self._run(delegate, self.LITELLM_KEY_HEADERS, has_stored_token=False)
 
 
+class TestPreemptive401GatedByKeyAllowlist:
+    """The preemptive OAuth 401 challenge must be gated on the servers the
+    caller's key is allowed to reach, not only on toolset scoping. A key whose
+    object_permission.mcp_servers excludes an oauth2 server must fall through
+    to the existing 403 denial instead of being pushed into an OAuth flow it
+    can never complete."""
+
+    @staticmethod
+    def _auth() -> UserAPIKeyAuth:
+        return UserAPIKeyAuth(api_key="sk-litellm-virtual-key")
+
+    @staticmethod
+    def _initialize_call(alias: str) -> tuple[dict[str, object], AsyncMock]:
+        body = (
+            b'{"jsonrpc":"2.0","id":1,"method":"initialize","params":'
+            b'{"protocolVersion":"2024-11-05","capabilities":{},'
+            b'"clientInfo":{"name":"test","version":"0"}}}'
+        )
+        receive = AsyncMock(return_value={"type": "http.request", "body": body, "more_body": False})
+        return {"type": "http", "method": "POST", "path": f"/mcp/{alias}", "headers": []}, receive
+
+    @contextlib.contextmanager
+    def _patches(
+        self,
+        server: MCPServer,
+        *,
+        allowed_servers: list[MCPServer],
+        handle_target: StreamableHTTPSessionManager,
+        toolset_auth: UserAPIKeyAuth | None = None,
+    ) -> Iterator[AsyncMock]:
+        from litellm.proxy._experimental.mcp_server import server as server_module
+
+        with contextlib.ExitStack() as stack:
+            stack.enter_context(
+                patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
+                    "litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
+                    new_callable=AsyncMock,
+                    return_value=(self._auth(), None, [server.alias], None, None, None),
+                )
+            )
+            stack.enter_context(
+                patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
+                    "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
+                    new_callable=AsyncMock,
+                    return_value=allowed_servers,
+                )
+            )
+            stack.enter_context(
+                patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
+                    "litellm.proxy._experimental.mcp_server.server._SESSION_MANAGERS_INITIALIZED",
+                    True,
+                )
+            )
+            stack.enter_context(
+                patch.object(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
+                    server_module.global_mcp_server_manager,
+                    "get_mcp_server_by_name",
+                    return_value=server,
+                )
+            )
+            stack.enter_context(
+                patch.object(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
+                    server_module.global_mcp_server_manager,
+                    "has_user_oauth_token",
+                    new_callable=AsyncMock,
+                    return_value=False,
+                )
+            )
+            stack.enter_context(
+                patch.object(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
+                    server_module.global_mcp_server_manager,
+                    "_ensure_upstream_initialize_instructions_cached",
+                    new_callable=AsyncMock,
+                )
+            )
+            if toolset_auth is not None:
+                stack.enter_context(
+                    patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
+                        "litellm.proxy._experimental.mcp_server.server._apply_toolset_scope",
+                        new_callable=AsyncMock,
+                        return_value=toolset_auth,
+                    )
+                )
+            handle_request = stack.enter_context(
+                patch.object(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
+                    handle_target,
+                    "handle_request",
+                    new_callable=AsyncMock,
+                )
+            )
+            yield handle_request
+
+    @pytest.mark.asyncio
+    async def test_key_not_allowed_gets_403_not_401_streamable(self):
+        """Denied key on an oauth2 server must reach the scoped-access denial
+        (403) instead of an OAuth challenge (401) it can never satisfy."""
+        from litellm.proxy._experimental.mcp_server import server as server_module
+
+        server = _make_oauth2_server("restricted")
+        scope, receive = self._initialize_call("restricted")
+
+        with (
+            self._patches(
+                server, allowed_servers=[], handle_target=server_module.session_manager_stateful
+            ) as handle_request,
+            pytest.raises(HTTPException) as exc,
+        ):
+            await server_module.handle_streamable_http_mcp(scope, receive, AsyncMock())
+
+        assert exc.value.status_code == 403
+        handle_request.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_key_allowed_still_challenges_streamable(self):
+        """Key allowlist includes the oauth2 server with no stored token:
+        the 401 challenge still fires, so the gate is not over-broad."""
+        from litellm.proxy._experimental.mcp_server import server as server_module
+
+        server = _make_oauth2_server("restricted")
+        scope, receive = self._initialize_call("restricted")
+
+        with (
+            self._patches(
+                server, allowed_servers=[server], handle_target=server_module.session_manager_stateful
+            ) as handle_request,
+            pytest.raises(HTTPException) as exc,
+        ):
+            await server_module.handle_streamable_http_mcp(scope, receive, AsyncMock())
+
+        assert exc.value.status_code == 401
+        assert "www-authenticate" in {k.lower() for k in exc.value.headers}
+        handle_request.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_toolset_excluding_server_does_not_challenge_streamable(
+        self,
+    ):  # test-quality-ok: reaching the session dispatch is the caller-visible outcome here; the denied-key variants assert the 403 itself
+        """Key allowlist includes the server but the active toolset scope
+        excludes it: no 401, same fall-through as the key allowlist."""
+        from litellm.proxy._experimental.mcp_server import server as server_module
+        from litellm.proxy._types import LiteLLM_ObjectPermissionTable
+
+        server = _make_oauth2_server("restricted")
+        scoped_auth = self._auth().model_copy(
+            update={
+                "object_permission": LiteLLM_ObjectPermissionTable(
+                    object_permission_id="toolset-scope",
+                    mcp_servers=["id-other"],
+                )
+            }
+        )
+        scope, receive = self._initialize_call("restricted")
+        token = server_module._mcp_active_toolset_id.set("toolset-1")
+        try:
+            with self._patches(
+                server,
+                allowed_servers=[server],
+                handle_target=server_module.session_manager_stateful,
+                toolset_auth=scoped_auth,
+            ) as handle_request:
+                await server_module.handle_streamable_http_mcp(scope, receive, AsyncMock())
+        finally:
+            server_module._mcp_active_toolset_id.reset(token)
+
+        handle_request.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_key_not_allowed_gets_403_not_401_sse(self):
+        """Same gate on the SSE transport: a denied key gets 403, not a 401."""
+        from litellm.proxy._experimental.mcp_server import server as server_module
+
+        server = _make_oauth2_server("restricted")
+        scope = {"type": "http", "method": "GET", "path": "/mcp/restricted", "headers": []}
+
+        with (
+            self._patches(
+                server, allowed_servers=[], handle_target=server_module.sse_session_manager
+            ) as handle_request,
+            pytest.raises(HTTPException) as exc,
+        ):
+            await server_module.handle_sse_mcp(scope, AsyncMock(), AsyncMock())
+
+        assert exc.value.status_code == 403
+        handle_request.assert_not_awaited()
+
+
 class TestSingleServerPreflightReachesIdJag:
     """The connect-time preflight is what turns a credential failure into an HTTP status the client
     can read. An oauth2_id_jag server has to reach it: its subject comes from the assertion stored at

From c31b31e509b7ffab983bd6e151d0291d7339efd6 Mon Sep 17 00:00:00 2001
From: jesus <jesus@berri.ai>
Date: Fri, 18 Sep 2026 00:43:06 +0000
Subject: [PATCH 2/8] test(mcp): patch _get_allowed_mcp_servers in
 preemptive-401 handler tests

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 .../mcp_server/test_mcp_stale_session.py      | 40 +++++++++++++++++++
 1 file changed, 40 insertions(+)

diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_stale_session.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_stale_session.py
index 9420eecd222..91dcbb4510c 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_stale_session.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_stale_session.py
@@ -647,6 +647,11 @@ async def test_per_user_oauth_missing_stored_token_returns_preemptive_401():
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=oauth_server,
         ),
+        patch(
+            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
+            new_callable=AsyncMock,
+            return_value=[oauth_server],
+        ),
         patch.object(
             session_manager_stateless,
             "handle_request",
@@ -735,6 +740,11 @@ async def test_admitted_subject_missing_stored_token_challenged_with_resource_me
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=oauth_server,
         ),
+        patch(
+            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
+            new_callable=AsyncMock,
+            return_value=[oauth_server],
+        ),
         patch.object(
             session_manager_stateless,
             "handle_request",
@@ -1030,6 +1040,11 @@ async def test_per_user_oauth_with_stored_token_skips_preemptive_401():
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=oauth_server,
         ),
+        patch(
+            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
+            new_callable=AsyncMock,
+            return_value=[oauth_server],
+        ),
         patch.object(
             session_manager_stateless,
             "handle_request",
@@ -1133,6 +1148,11 @@ async def test_handle_streamable_http_mcp_delegated_server_without_token_returns
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=delegated_server,
         ),
+        patch(
+            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
+            new_callable=AsyncMock,
+            return_value=[delegated_server],
+        ),
         patch.object(
             session_manager_stateful,
             "handle_request",
@@ -1221,6 +1241,11 @@ async def test_handle_streamable_http_mcp_token_exchange_without_subject_returns
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=obo_server,
         ),
+        patch(
+            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
+            new_callable=AsyncMock,
+            return_value=[obo_server],
+        ),
         patch.object(
             session_manager_stateful,
             "handle_request",
@@ -1320,6 +1345,11 @@ async def test_handle_streamable_http_mcp_oauth_delegate_without_token_returns_g
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=od_server,
         ),
+        patch(
+            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
+            new_callable=AsyncMock,
+            return_value=[od_server],
+        ),
         patch.object(
             session_manager_stateful,
             "handle_request",
@@ -1577,6 +1607,11 @@ async def test_handle_streamable_http_mcp_true_passthrough_without_token_surface
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=tp_server,
         ),
+        patch(
+            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
+            new_callable=AsyncMock,
+            return_value=[tp_server],
+        ),
         patch.object(
             session_manager_stateful,
             "handle_request",
@@ -1645,6 +1680,11 @@ async def test_handle_streamable_http_mcp_true_passthrough_dcr_bridge_challenges
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=bridge_server,
         ),
+        patch(
+            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
+            new_callable=AsyncMock,
+            return_value=[bridge_server],
+        ),
         patch.object(
             session_manager_stateful,
             "handle_request",

From 9c5a15ba01a2b63f8409118569e3c0c4a7f1cdac Mon Sep 17 00:00:00 2001
From: jesus <jesus@berri.ai>
Date: Fri, 18 Sep 2026 00:48:16 +0000
Subject: [PATCH 3/8] test(mcp): suppress TQ008 on preemptive-401
 allowed-servers patches

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 .../mcp_server/test_mcp_stale_session.py         | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_stale_session.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_stale_session.py
index 91dcbb4510c..b40f2917aae 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_stale_session.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_stale_session.py
@@ -647,7 +647,7 @@ async def test_per_user_oauth_missing_stored_token_returns_preemptive_401():
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=oauth_server,
         ),
-        patch(
+        patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
             "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
             new_callable=AsyncMock,
             return_value=[oauth_server],
@@ -740,7 +740,7 @@ async def test_admitted_subject_missing_stored_token_challenged_with_resource_me
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=oauth_server,
         ),
-        patch(
+        patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
             "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
             new_callable=AsyncMock,
             return_value=[oauth_server],
@@ -1040,7 +1040,7 @@ async def test_per_user_oauth_with_stored_token_skips_preemptive_401():
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=oauth_server,
         ),
-        patch(
+        patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
             "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
             new_callable=AsyncMock,
             return_value=[oauth_server],
@@ -1148,7 +1148,7 @@ async def test_handle_streamable_http_mcp_delegated_server_without_token_returns
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=delegated_server,
         ),
-        patch(
+        patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
             "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
             new_callable=AsyncMock,
             return_value=[delegated_server],
@@ -1241,7 +1241,7 @@ async def test_handle_streamable_http_mcp_token_exchange_without_subject_returns
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=obo_server,
         ),
-        patch(
+        patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
             "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
             new_callable=AsyncMock,
             return_value=[obo_server],
@@ -1345,7 +1345,7 @@ async def test_handle_streamable_http_mcp_oauth_delegate_without_token_returns_g
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=od_server,
         ),
-        patch(
+        patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
             "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
             new_callable=AsyncMock,
             return_value=[od_server],
@@ -1607,7 +1607,7 @@ async def test_handle_streamable_http_mcp_true_passthrough_without_token_surface
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=tp_server,
         ),
-        patch(
+        patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
             "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
             new_callable=AsyncMock,
             return_value=[tp_server],
@@ -1680,7 +1680,7 @@ async def test_handle_streamable_http_mcp_true_passthrough_dcr_bridge_challenges
             "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
             return_value=bridge_server,
         ),
-        patch(
+        patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
             "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
             new_callable=AsyncMock,
             return_value=[bridge_server],

From 7354ef595cbbd7c73cf8672871e89e92dd977032 Mon Sep 17 00:00:00 2001
From: jesus <jesus@berri.ai>
Date: Fri, 18 Sep 2026 21:36:34 +0000
Subject: [PATCH 4/8] refactor(mcp): resolve the allowed MCP server set once
 per request

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 .../proxy/_experimental/mcp_server/server.py  |  62 ++---
 .../mcp_server/test_mcp_server.py             | 243 +++++++-----------
 2 files changed, 110 insertions(+), 195 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py
index 8d6b8907fc9..57e603d0cb2 100644
--- a/litellm/proxy/_experimental/mcp_server/server.py
+++ b/litellm/proxy/_experimental/mcp_server/server.py
@@ -4014,15 +4014,10 @@ if MCP_AVAILABLE:
             )
         return user_api_key_auth.model_copy(update={"object_permission": updated_op})
 
-    async def _challenge_allowed_server_ids(
-        user_api_key_auth: UserAPIKeyAuth | None,
-        mcp_servers: list[str] | None,
-        client_ip: str | None,
+    def _narrow_allowed_server_ids(
+        allowed_servers: Sequence[MCPServer],
         toolset_allowed_server_ids: AbstractSet[str] | None,
     ) -> frozenset[str]:
-        allowed_servers: Final = await _get_allowed_mcp_servers(
-            user_api_key_auth=user_api_key_auth, mcp_servers=mcp_servers, client_ip=client_ip
-        )
         allowed_ids: Final = frozenset(srv.server_id for srv in allowed_servers)
         if toolset_allowed_server_ids is None:
             return allowed_ids
@@ -4333,14 +4328,12 @@ if MCP_AVAILABLE:
 
     async def _check_passthrough_upstream_auth(
         scope: Scope,
-        user_api_key_auth: UserAPIKeyAuth | None,
-        mcp_servers: list[str] | None,
-        client_ip: str | None,
+        allowed_servers: Sequence[MCPServer],
     ) -> None:
         """Probe pass-through 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
+        Only servers the caller's key is already authorized to reach are probed;
+        the caller passes the resolved allowed-server set so a user cannot
         trigger an upstream probe against a server their key is not permitted for.
 
         The MCP SDK commits HTTP 200 headers before invoking handlers, so a 401
@@ -4354,13 +4347,6 @@ if MCP_AVAILABLE:
         if not forwarded_auth:
             return
 
-        # Use the authorized server set, not the raw user-supplied names, so that
-        # a caller cannot force a probe to a server their key is not allowed to use.
-        allowed_servers: Final = await _get_allowed_mcp_servers(
-            user_api_key_auth=user_api_key_auth,
-            mcp_servers=mcp_servers,
-            client_ip=client_ip,
-        )
         passthrough_targets: Final[tuple[tuple[MCPServer, str, str], ...]] = tuple(
             (srv, forwarded_auth, srv.name)
             for srv in allowed_servers
@@ -4446,6 +4432,11 @@ if MCP_AVAILABLE:
             # narrowed by any active toolset scope): a server the key cannot
             # reach should not trigger an OAuth flow for a server the caller
             # will be 403'd on after authentication.
+            allowed_servers: Final = await _get_allowed_mcp_servers(
+                user_api_key_auth=user_api_key_auth,
+                mcp_servers=mcp_servers,
+                client_ip=_client_ip,
+            )
             await _raise_preemptive_401_for_unauthenticated_servers(
                 scope=scope,
                 mcp_servers=mcp_servers,
@@ -4453,19 +4444,13 @@ if MCP_AVAILABLE:
                 mcp_server_auth_headers=mcp_server_auth_headers,
                 user_api_key_auth=user_api_key_auth,
                 client_ip=_client_ip,
-                allowed_server_ids=await _challenge_allowed_server_ids(
-                    user_api_key_auth=user_api_key_auth,
-                    mcp_servers=mcp_servers,
-                    client_ip=_client_ip,
-                    toolset_allowed_server_ids=toolset_allowed_server_ids,
-                ),
+                allowed_server_ids=_narrow_allowed_server_ids(allowed_servers, toolset_allowed_server_ids),
                 raw_headers=raw_headers,
             )
 
-            # Pre-flight auth check for pass-through servers.  Must run after
-            # toolset scoping so the probe list is derived from the fully-authorized
-            # server set, not the raw user-supplied names.
-            await _check_passthrough_upstream_auth(scope, user_api_key_auth, mcp_servers, _client_ip)
+            # Pre-flight auth check for pass-through servers, probed over the
+            # same fully-authorized server set resolved above.
+            await _check_passthrough_upstream_auth(scope, allowed_servers)
 
             # Inject masked debug headers when client sends x-litellm-mcp-debug: true
             _debug_headers: Final = MCPDebug.maybe_build_debug_headers(
@@ -4780,6 +4765,11 @@ if MCP_AVAILABLE:
             # narrowed by any active toolset scope): a server the key cannot
             # reach should not trigger an OAuth flow for a server the caller
             # will be 403'd on after authentication.
+            allowed_servers: Final = await _get_allowed_mcp_servers(
+                user_api_key_auth=user_api_key_auth,
+                mcp_servers=mcp_servers,
+                client_ip=_sse_client_ip,
+            )
             await _raise_preemptive_401_for_unauthenticated_servers(
                 scope=scope,
                 mcp_servers=mcp_servers,
@@ -4787,22 +4777,16 @@ if MCP_AVAILABLE:
                 mcp_server_auth_headers=mcp_server_auth_headers,
                 user_api_key_auth=user_api_key_auth,
                 client_ip=_sse_client_ip,
-                allowed_server_ids=await _challenge_allowed_server_ids(
-                    user_api_key_auth=user_api_key_auth,
-                    mcp_servers=mcp_servers,
-                    client_ip=_sse_client_ip,
-                    toolset_allowed_server_ids=toolset_allowed_server_ids,
-                ),
+                allowed_server_ids=_narrow_allowed_server_ids(allowed_servers, toolset_allowed_server_ids),
                 raw_headers=raw_headers,
             )
 
             # Pre-flight auth check for pass-through servers: surface upstream
             # 401/403 as a proper challenge before the SSE session commits 200
             # headers, so clients can refresh their OAuth token instead of
-            # being stuck with a silently empty tool list. Must run after
-            # toolset scoping so the probe list is derived from the fully-
-            # authorized server set, not the raw user-supplied names.
-            await _check_passthrough_upstream_auth(scope, user_api_key_auth, mcp_servers, _sse_client_ip)
+            # being stuck with a silently empty tool list. Probed over the same
+            # fully-authorized server set resolved above.
+            await _check_passthrough_upstream_auth(scope, allowed_servers)
             set_auth_context(
                 user_api_key_auth=user_api_key_auth,
                 mcp_auth_header=mcp_auth_header,
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
index 55cab45169e..36b4fbf2469 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
@@ -6252,28 +6252,18 @@ async def test_legacy_delegate_bare_token_is_not_probed_upstream():  # test-qual
     from litellm.proxy._experimental.mcp_server.server import (
         _check_passthrough_upstream_auth,
     )
-    from litellm.proxy._types import UserAPIKeyAuth
 
     server = _delegate_auth_mcp_server()
     scope = _delegate_scope([(b"authorization", b"Bearer bogus-token")])
 
     with (
         _patch_delegate_resolver(server, "delegate_test"),
-        patch(
-            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=[server]),
-        ),
         patch(
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(401, 'Bearer realm="upstream", error="invalid_token"')),
         ) as probe,
     ):
-        await _check_passthrough_upstream_auth(
-            scope=scope,
-            user_api_key_auth=UserAPIKeyAuth(),
-            mcp_servers=["delegate_test"],
-            client_ip=None,
-        )
+        await _check_passthrough_upstream_auth(scope=scope, allowed_servers=[server])
 
     probe.assert_not_awaited()
 
@@ -6284,7 +6274,6 @@ async def test_legacy_delegate_dual_credentials_are_not_probed_upstream():  # te
     from litellm.proxy._experimental.mcp_server.server import (
         _check_passthrough_upstream_auth,
     )
-    from litellm.proxy._types import UserAPIKeyAuth
 
     server = _delegate_auth_mcp_server()
     scope = _delegate_scope(
@@ -6295,21 +6284,12 @@ async def test_legacy_delegate_dual_credentials_are_not_probed_upstream():  # te
     )
 
     with (
-        patch(  # test-quality-ok: isolate authorized-server resolution so this test targets the preflight boundary
-            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=[server]),
-        ),
         patch(  # test-quality-ok: the removed probe call is the security regression under test
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(),
         ) as probe,
     ):
-        await _check_passthrough_upstream_auth(
-            scope=scope,
-            user_api_key_auth=UserAPIKeyAuth(user_id="admitted-user"),
-            mcp_servers=["delegate_test"],
-            client_ip=None,
-        )
+        await _check_passthrough_upstream_auth(scope=scope, allowed_servers=[server])
 
     probe.assert_not_awaited()
 
@@ -6323,7 +6303,6 @@ async def test_oauth_passthrough_preflight_preserves_status_contract(probe_statu
     from litellm.proxy._experimental.mcp_server.server import (
         _check_passthrough_upstream_auth,
     )
-    from litellm.proxy._types import UserAPIKeyAuth
 
     server = MCPServer(
         server_id="passthrough-id",
@@ -6342,31 +6321,17 @@ async def test_oauth_passthrough_preflight_preserves_status_contract(probe_statu
     )
 
     with (
-        patch(  # test-quality-ok: isolate authorized-server resolution so this test exercises the preflight contract
-            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=[server]),
-        ),
         patch(  # test-quality-ok: the upstream transport boundary is the behavior being mapped to an HTTP response
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(probe_status, None)),
         ) as probe,
     ):
         if expected_status is None:
-            result = await _check_passthrough_upstream_auth(
-                scope=scope,
-                user_api_key_auth=UserAPIKeyAuth(user_id="admitted-user"),
-                mcp_servers=["passthrough_server"],
-                client_ip=None,
-            )
+            result = await _check_passthrough_upstream_auth(scope=scope, allowed_servers=[server])
             assert result is None
         else:
             with pytest.raises(HTTPException) as exc_info:
-                await _check_passthrough_upstream_auth(
-                    scope=scope,
-                    user_api_key_auth=UserAPIKeyAuth(user_id="admitted-user"),
-                    mcp_servers=["passthrough_server"],
-                    client_ip=None,
-                )
+                await _check_passthrough_upstream_auth(scope=scope, allowed_servers=[server])
             assert exc_info.value.status_code == expected_status
             if expected_status == 401:
                 assert "passthrough_server" in exc_info.value.headers["www-authenticate"]
@@ -6381,28 +6346,18 @@ async def test_delegate_tokenless_request_not_probed():
     from litellm.proxy._experimental.mcp_server.server import (
         _check_passthrough_upstream_auth,
     )
-    from litellm.proxy._types import UserAPIKeyAuth
 
     server = _delegate_auth_mcp_server()
     scope = _delegate_scope([(b"content-type", b"application/json")])
 
     with (
         _patch_delegate_resolver(server, "delegate_test"),
-        patch(
-            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=[server]),
-        ),
         patch(
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(401, None)),
         ) as probe,
     ):
-        await _check_passthrough_upstream_auth(
-            scope=scope,
-            user_api_key_auth=UserAPIKeyAuth(),
-            mcp_servers=["delegate_test"],
-            client_ip=None,
-        )
+        await _check_passthrough_upstream_auth(scope=scope, allowed_servers=[server])
 
     probe.assert_not_awaited()
 
@@ -6414,28 +6369,18 @@ async def test_delegate_preflight_skipped_on_multi_server_routes():
     from litellm.proxy._experimental.mcp_server.server import (
         _check_passthrough_upstream_auth,
     )
-    from litellm.proxy._types import UserAPIKeyAuth
 
     servers = [_delegate_auth_mcp_server("delegate-1"), _delegate_auth_mcp_server("delegate-2")]
     scope = _delegate_scope([(b"authorization", b"Bearer bogus-token")])
 
     with (
         _patch_delegate_resolver(servers[0], "delegate_test", "other_server"),
-        patch(
-            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=servers),
-        ),
         patch(
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(401, None)),
         ) as probe,
     ):
-        await _check_passthrough_upstream_auth(
-            scope=scope,
-            user_api_key_auth=UserAPIKeyAuth(),
-            mcp_servers=["delegate_test", "other_server"],
-            client_ip=None,
-        )
+        await _check_passthrough_upstream_auth(scope=scope, allowed_servers=servers)
 
     probe.assert_not_awaited()
 
@@ -6449,7 +6394,6 @@ async def test_bare_authorization_never_probes_passthrough_servers():
     from litellm.proxy._experimental.mcp_server.server import (
         _check_passthrough_upstream_auth,
     )
-    from litellm.proxy._types import UserAPIKeyAuth
 
     passthrough_server = MCPServer(
         server_id="pt-1",
@@ -6464,21 +6408,12 @@ async def test_bare_authorization_never_probes_passthrough_servers():
 
     with (
         _patch_delegate_resolver(passthrough_server, "pt_server"),
-        patch(
-            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=[passthrough_server]),
-        ),
         patch(
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(401, None)),
         ) as probe,
     ):
-        await _check_passthrough_upstream_auth(
-            scope=scope,
-            user_api_key_auth=UserAPIKeyAuth(),
-            mcp_servers=["pt_server"],
-            client_ip=None,
-        )
+        await _check_passthrough_upstream_auth(scope=scope, allowed_servers=[passthrough_server])
 
     probe.assert_not_awaited()
 
@@ -6494,7 +6429,6 @@ async def test_delegate_not_probed_when_named_only_via_server_id():
     from litellm.proxy._experimental.mcp_server.server import (
         _check_passthrough_upstream_auth,
     )
-    from litellm.proxy._types import UserAPIKeyAuth
 
     server = _delegate_auth_mcp_server(server_id="delegate-secret-id")
     # Admission's resolver matches alias/server_name/name only, never server_id: the
@@ -6510,21 +6444,12 @@ async def test_delegate_not_probed_when_named_only_via_server_id():
 
     with (
         _patch_delegate_resolver(server, "delegate_test"),
-        patch(
-            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=[server]),
-        ),
         patch(
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(401, None)),
         ) as probe,
     ):
-        await _check_passthrough_upstream_auth(
-            scope=scope,
-            user_api_key_auth=UserAPIKeyAuth(user_id="u1", api_key="hashed-sk"),
-            mcp_servers=["delegate-secret-id"],
-            client_ip=None,
-        )
+        await _check_passthrough_upstream_auth(scope=scope, allowed_servers=[server])
 
     probe.assert_not_awaited()
 
@@ -6537,16 +6462,11 @@ async def test_delegate_probe_not_fanned_out_to_access_group_members():
     from litellm.proxy._experimental.mcp_server.server import (
         _check_passthrough_upstream_auth,
     )
-    from litellm.proxy._types import UserAPIKeyAuth
 
     group_member = _delegate_auth_mcp_server()
 
     with (
         _patch_delegate_resolver(group_member, "delegate_test"),
-        patch(
-            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=[group_member]),
-        ),
         patch(
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(401, None)),
@@ -6554,9 +6474,7 @@ async def test_delegate_probe_not_fanned_out_to_access_group_members():
     ):
         await _check_passthrough_upstream_auth(
             scope=_delegate_scope([(b"authorization", b"Bearer bogus-token")]),
-            user_api_key_auth=UserAPIKeyAuth(),
-            mcp_servers=["prod_tools_group"],
-            client_ip=None,
+            allowed_servers=[group_member],
         )
 
     probe.assert_not_awaited()
@@ -8686,65 +8604,74 @@ class TestPreemptive401GatedByKeyAllowlist:
     ) -> Iterator[AsyncMock]:
         from litellm.proxy._experimental.mcp_server import server as server_module
 
-        with contextlib.ExitStack() as stack:
-            stack.enter_context(
-                patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
-                    "litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
-                    new_callable=AsyncMock,
-                    return_value=(self._auth(), None, [server.alias], None, None, None),
-                )
-            )
-            stack.enter_context(
-                patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
-                    "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
-                    new_callable=AsyncMock,
-                    return_value=allowed_servers,
-                )
-            )
-            stack.enter_context(
-                patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
-                    "litellm.proxy._experimental.mcp_server.server._SESSION_MANAGERS_INITIALIZED",
-                    True,
-                )
-            )
-            stack.enter_context(
-                patch.object(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
-                    server_module.global_mcp_server_manager,
-                    "get_mcp_server_by_name",
-                    return_value=server,
-                )
-            )
-            stack.enter_context(
-                patch.object(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
-                    server_module.global_mcp_server_manager,
-                    "has_user_oauth_token",
-                    new_callable=AsyncMock,
-                    return_value=False,
-                )
-            )
-            stack.enter_context(
-                patch.object(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
-                    server_module.global_mcp_server_manager,
-                    "_ensure_upstream_initialize_instructions_cached",
-                    new_callable=AsyncMock,
-                )
-            )
-            if toolset_auth is not None:
+        def _resolve_allowed(
+            user_api_key_auth: UserAPIKeyAuth | None = None,
+            mcp_servers: list[str] | None = None,
+            client_ip: str | None = None,
+        ) -> list[MCPServer]:
+            scoped = getattr(user_api_key_auth, "object_permission", None)
+            scoped_ids = getattr(scoped, "mcp_servers", None) if scoped else None
+            if scoped_ids:
+                return [srv for srv in allowed_servers if srv.server_id in scoped_ids]
+            return allowed_servers
+
+        registry = server_module.global_mcp_server_manager.registry
+        registry[server.server_id] = server
+        try:
+            with contextlib.ExitStack() as stack:
                 stack.enter_context(
-                    patch(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
-                        "litellm.proxy._experimental.mcp_server.server._apply_toolset_scope",
+                    patch(  # test-quality-ok: admission parse is unrelated to the challenge gate; the handler reads it as a module global
+                        "litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
                         new_callable=AsyncMock,
-                        return_value=toolset_auth,
+                        return_value=(self._auth(), None, [server.alias], None, None, None),
                     )
                 )
-            handle_request = stack.enter_context(
-                patch.object(  # test-quality-ok: transport handler collaborators are module-level globals, no injectable seam
-                    handle_target,
-                    "handle_request",
-                    new_callable=AsyncMock,
+                stack.enter_context(
+                    patch(  # test-quality-ok: allowed-server resolution hits the key/toolset DB; the test must control the allowlist
+                        "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers",
+                        new_callable=AsyncMock,
+                        side_effect=_resolve_allowed,
+                    )
                 )
-            )
-            yield handle_request
+                stack.enter_context(
+                    patch(  # test-quality-ok: module-level startup flag gates the transport; there is no injectable seam
+                        "litellm.proxy._experimental.mcp_server.server._SESSION_MANAGERS_INITIALIZED",
+                        True,
+                    )
+                )
+                stack.enter_context(
+                    patch.object(  # test-quality-ok: no stored per-user token is the scenario; the real lookup hits the token cache/DB
+                        server_module.global_mcp_server_manager,
+                        "has_user_oauth_token",
+                        new_callable=AsyncMock,
+                        return_value=False,
+                    )
+                )
+                stack.enter_context(
+                    patch.object(  # test-quality-ok: the real cache warm-up calls the upstream server over the network
+                        server_module.global_mcp_server_manager,
+                        "_ensure_upstream_initialize_instructions_cached",
+                        new_callable=AsyncMock,
+                    )
+                )
+                if toolset_auth is not None:
+                    stack.enter_context(
+                        patch(  # test-quality-ok: toolset resolution hits the DB; the test controls the scoped auth directly
+                            "litellm.proxy._experimental.mcp_server.server._apply_toolset_scope",
+                            new_callable=AsyncMock,
+                            return_value=toolset_auth,
+                        )
+                    )
+                handle_request = stack.enter_context(
+                    patch.object(  # test-quality-ok: session dispatch is a module-level global; spying on it proves the request was not denied earlier
+                        handle_target,
+                        "handle_request",
+                        new_callable=AsyncMock,
+                    )
+                )
+                yield handle_request
+        finally:
+            registry.pop(server.server_id, None)
 
     @pytest.mark.asyncio
     async def test_key_not_allowed_gets_403_not_401_streamable(self):
@@ -8788,11 +8715,10 @@ class TestPreemptive401GatedByKeyAllowlist:
         handle_request.assert_not_awaited()
 
     @pytest.mark.asyncio
-    async def test_toolset_excluding_server_does_not_challenge_streamable(
-        self,
-    ):  # test-quality-ok: reaching the session dispatch is the caller-visible outcome here; the denied-key variants assert the 403 itself
+    async def test_toolset_excluding_server_gets_403_not_401_streamable(self):
         """Key allowlist includes the server but the active toolset scope
-        excludes it: no 401, same fall-through as the key allowlist."""
+        excludes it: the caller-visible denial is the scoped-access 403, not
+        a 401 challenge."""
         from litellm.proxy._experimental.mcp_server import server as server_module
         from litellm.proxy._types import LiteLLM_ObjectPermissionTable
 
@@ -8808,17 +8734,22 @@ class TestPreemptive401GatedByKeyAllowlist:
         scope, receive = self._initialize_call("restricted")
         token = server_module._mcp_active_toolset_id.set("toolset-1")
         try:
-            with self._patches(
-                server,
-                allowed_servers=[server],
-                handle_target=server_module.session_manager_stateful,
-                toolset_auth=scoped_auth,
-            ) as handle_request:
+            with (
+                self._patches(
+                    server,
+                    allowed_servers=[server],
+                    handle_target=server_module.session_manager_stateful,
+                    toolset_auth=scoped_auth,
+                ) as handle_request,
+                pytest.raises(HTTPException) as exc,
+            ):
                 await server_module.handle_streamable_http_mcp(scope, receive, AsyncMock())
         finally:
             server_module._mcp_active_toolset_id.reset(token)
 
-        handle_request.assert_awaited_once()
+        assert exc.value.status_code == 403
+        assert "www-authenticate" not in {k.lower() for k in exc.value.headers or {}}
+        handle_request.assert_not_awaited()
 
     @pytest.mark.asyncio
     async def test_key_not_allowed_gets_403_not_401_sse(self):

From 5651651cb32605da58ee3d6d811a7b0f231d98dc Mon Sep 17 00:00:00 2001
From: jesus <jesus@berri.ai>
Date: Fri, 18 Sep 2026 21:37:40 +0000
Subject: [PATCH 5/8] test(mcp): drop getattr from the allowlist test double

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 .../proxy/_experimental/mcp_server/test_mcp_server.py | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
index 36b4fbf2469..d8a76cd9e17 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
@@ -8609,11 +8609,12 @@ class TestPreemptive401GatedByKeyAllowlist:
             mcp_servers: list[str] | None = None,
             client_ip: str | None = None,
         ) -> list[MCPServer]:
-            scoped = getattr(user_api_key_auth, "object_permission", None)
-            scoped_ids = getattr(scoped, "mcp_servers", None) if scoped else None
-            if scoped_ids:
-                return [srv for srv in allowed_servers if srv.server_id in scoped_ids]
-            return allowed_servers
+            if user_api_key_auth is None or user_api_key_auth.object_permission is None:
+                return allowed_servers
+            scoped_ids: Final = user_api_key_auth.object_permission.mcp_servers
+            if not scoped_ids:
+                return allowed_servers
+            return [srv for srv in allowed_servers if srv.server_id in scoped_ids]
 
         registry = server_module.global_mcp_server_manager.registry
         registry[server.server_id] = server

From 5323979b44a081f4fa23d7af19d2062bd8fabe6a Mon Sep 17 00:00:00 2001
From: jesus <jesus@berri.ai>
Date: Sat, 17 Oct 2026 01:17:17 +0000
Subject: [PATCH 6/8] refactor(mcp): resolve allowed MCP servers once per
 request for challenge and passthrough probe

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 .../proxy/_experimental/mcp_server/server.py  |  24 +--
 .../test_mcp_server_tool_calls_and_headers.py | 182 +++++++++++-------
 2 files changed, 123 insertions(+), 83 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py
index 69fd12665b6..9592e5a6321 100644
--- a/litellm/proxy/_experimental/mcp_server/server.py
+++ b/litellm/proxy/_experimental/mcp_server/server.py
@@ -2154,8 +2154,7 @@ if MCP_AVAILABLE:
     async def _check_passthrough_upstream_auth(
         scope: Scope,
         user_api_key_auth: UserAPIKeyAuth | None,
-        mcp_servers: list[str] | None,
-        client_ip: str | None,
+        allowed_servers: Sequence[MCPServer],
         *,
         oauth2_headers: Mapping[str, str] | None,
         mcp_server_auth_headers: dict[str, dict[str, str]] | None,
@@ -2163,9 +2162,8 @@ if MCP_AVAILABLE:
     ) -> None:
         """Probe pass-through 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
-        trigger an upstream probe against a server their key is not permitted for.
+        The caller passes the already-resolved allowed-server set so a user cannot
+        trigger a probe against a server their key is not permitted for.
 
         The MCP SDK commits HTTP 200 headers before invoking handlers, so a 401
         can only be returned before that point. This function raises HTTPException(401)
@@ -2177,13 +2175,6 @@ if MCP_AVAILABLE:
         if not oauth2_headers and not mcp_server_auth_headers:
             return
 
-        # Use the authorized server set, not the raw user-supplied names, so that
-        # a caller cannot force a probe to a server their key is not allowed to use.
-        allowed_servers: Final = await operations._get_allowed_mcp_servers(
-            user_api_key_auth=user_api_key_auth,
-            mcp_servers=mcp_servers,
-            client_ip=client_ip,
-        )
         prepared_headers: Final = tuple(
             (
                 srv,
@@ -2302,8 +2293,7 @@ if MCP_AVAILABLE:
         await _check_passthrough_upstream_auth(
             scope,
             context.user_api_key_auth,
-            authorized_names,
-            context.client_ip,
+            allowed,
             oauth2_headers=context.oauth2_headers,
             mcp_server_auth_headers={key: dict(value) for key, value in context.mcp_server_auth_headers.items()}
             if context.mcp_server_auth_headers is not None
@@ -2350,8 +2340,7 @@ if MCP_AVAILABLE:
             await _check_passthrough_upstream_auth(
                 scope,
                 user_api_key_auth,
-                list(mcp_servers) if mcp_servers is not None else None,
-                client_ip,
+                allowed_servers,
                 oauth2_headers=oauth2_headers,
                 mcp_server_auth_headers=(
                     {name: dict(headers) for name, headers in mcp_server_auth_headers.items()}
@@ -2859,8 +2848,7 @@ if MCP_AVAILABLE:
             await _check_passthrough_upstream_auth(
                 scope,
                 user_api_key_auth,
-                mcp_servers,
-                _sse_client_ip,
+                allowed_servers,
                 oauth2_headers=oauth2_headers,
                 mcp_server_auth_headers=mcp_server_auth_headers,
                 raw_headers=raw_headers,
diff --git a/tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py b/tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py
index 4d558df62af..7d98dfe5d33 100644
--- a/tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py
+++ b/tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py
@@ -7151,10 +7151,6 @@ async def test_legacy_delegate_bare_token_is_not_probed_upstream():  # test-qual
 
     with (
         _patch_delegate_resolver(server, "delegate_test"),
-        patch(
-            "litellm.proxy._experimental.mcp_server.operations._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=[server]),
-        ),
         patch(
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(401, 'Bearer realm="upstream", error="invalid_token"')),
@@ -7163,8 +7159,7 @@ async def test_legacy_delegate_bare_token_is_not_probed_upstream():  # test-qual
         await _check_passthrough_upstream_auth(
             scope=scope,
             user_api_key_auth=UserAPIKeyAuth(),
-            mcp_servers=["delegate_test"],
-            client_ip=None,
+            allowed_servers=[server],
             oauth2_headers=None,
             mcp_server_auth_headers=None,
             raw_headers=None,
@@ -7189,10 +7184,6 @@ async def test_legacy_delegate_dual_credentials_are_not_probed_upstream():  # te
     )
 
     with (
-        patch(  # test-quality-ok: isolate authorized-server resolution so this test targets the preflight boundary
-            "litellm.proxy._experimental.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
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(),
@@ -7201,8 +7192,7 @@ async def test_legacy_delegate_dual_credentials_are_not_probed_upstream():  # te
         await _check_passthrough_upstream_auth(
             scope=scope,
             user_api_key_auth=UserAPIKeyAuth(user_id="admitted-user"),
-            mcp_servers=["delegate_test"],
-            client_ip=None,
+            allowed_servers=[server],
             oauth2_headers={"Authorization": "Bearer upstream-token"},
             mcp_server_auth_headers=None,
             raw_headers={"x-litellm-api-key": "sk-litellm-proxy-key", "Authorization": "Bearer upstream-token"},
@@ -7238,10 +7228,6 @@ async def test_oauth_passthrough_preflight_preserves_status_contract(probe_statu
     )
 
     with (
-        patch(  # test-quality-ok: isolate authorized-server resolution so this test exercises the preflight contract
-            "litellm.proxy._experimental.mcp_server.operations._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=[server]),
-        ),
         patch(  # test-quality-ok: the upstream transport boundary is the behavior being mapped to an HTTP response
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(probe_status, None)),
@@ -7251,11 +7237,13 @@ async def test_oauth_passthrough_preflight_preserves_status_contract(probe_statu
             result = await _check_passthrough_upstream_auth(
                 scope=scope,
                 user_api_key_auth=UserAPIKeyAuth(user_id="admitted-user"),
-                mcp_servers=["passthrough_server"],
-                client_ip=None,
-            oauth2_headers={"Authorization": "Bearer upstream-token"},
-            mcp_server_auth_headers=None,
-            raw_headers={"x-litellm-api-key": "sk-litellm-proxy-key", "Authorization": "Bearer upstream-token"},
+                allowed_servers=[server],
+                oauth2_headers={"Authorization": "Bearer upstream-token"},
+                mcp_server_auth_headers=None,
+                raw_headers={
+                    "x-litellm-api-key": "sk-litellm-proxy-key",
+                    "Authorization": "Bearer upstream-token",
+                },
             )
             assert result is None
         else:
@@ -7263,11 +7251,13 @@ async def test_oauth_passthrough_preflight_preserves_status_contract(probe_statu
                 await _check_passthrough_upstream_auth(
                     scope=scope,
                     user_api_key_auth=UserAPIKeyAuth(user_id="admitted-user"),
-                    mcp_servers=["passthrough_server"],
-                    client_ip=None,
-            oauth2_headers={"Authorization": "Bearer upstream-token"},
-            mcp_server_auth_headers=None,
-            raw_headers={"x-litellm-api-key": "sk-litellm-proxy-key", "Authorization": "Bearer upstream-token"},
+                    allowed_servers=[server],
+                    oauth2_headers={"Authorization": "Bearer upstream-token"},
+                    mcp_server_auth_headers=None,
+                    raw_headers={
+                        "x-litellm-api-key": "sk-litellm-proxy-key",
+                        "Authorization": "Bearer upstream-token",
+                    },
                 )
             assert exc_info.value.status_code == expected_status
             if expected_status == 401:
@@ -7289,10 +7279,6 @@ async def test_delegate_tokenless_request_not_probed():
 
     with (
         _patch_delegate_resolver(server, "delegate_test"),
-        patch(
-            "litellm.proxy._experimental.mcp_server.operations._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=[server]),
-        ),
         patch(
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(401, None)),
@@ -7301,8 +7287,7 @@ async def test_delegate_tokenless_request_not_probed():
         await _check_passthrough_upstream_auth(
             scope=scope,
             user_api_key_auth=UserAPIKeyAuth(),
-            mcp_servers=["delegate_test"],
-            client_ip=None,
+            allowed_servers=[server],
             oauth2_headers=None,
             mcp_server_auth_headers=None,
             raw_headers=None,
@@ -7324,10 +7309,6 @@ async def test_delegate_preflight_skipped_on_multi_server_routes():
 
     with (
         _patch_delegate_resolver(servers[0], "delegate_test", "other_server"),
-        patch(
-            "litellm.proxy._experimental.mcp_server.operations._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=servers),
-        ),
         patch(
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(401, None)),
@@ -7336,8 +7317,7 @@ async def test_delegate_preflight_skipped_on_multi_server_routes():
         await _check_passthrough_upstream_auth(
             scope=scope,
             user_api_key_auth=UserAPIKeyAuth(),
-            mcp_servers=["delegate_test", "other_server"],
-            client_ip=None,
+            allowed_servers=servers,
             oauth2_headers=None,
             mcp_server_auth_headers=None,
             raw_headers=None,
@@ -7369,10 +7349,6 @@ async def test_bare_authorization_never_probes_passthrough_servers():
 
     with (
         _patch_delegate_resolver(passthrough_server, "pt_server"),
-        patch(
-            "litellm.proxy._experimental.mcp_server.operations._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=[passthrough_server]),
-        ),
         patch(
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(401, None)),
@@ -7381,8 +7357,7 @@ async def test_bare_authorization_never_probes_passthrough_servers():
         await _check_passthrough_upstream_auth(
             scope=scope,
             user_api_key_auth=UserAPIKeyAuth(),
-            mcp_servers=["pt_server"],
-            client_ip=None,
+            allowed_servers=[passthrough_server],
             oauth2_headers=None,
             mcp_server_auth_headers=None,
             raw_headers=None,
@@ -7417,10 +7392,6 @@ async def test_delegate_not_probed_when_named_only_via_server_id():
 
     with (
         _patch_delegate_resolver(server, "delegate_test"),
-        patch(
-            "litellm.proxy._experimental.mcp_server.operations._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=[server]),
-        ),
         patch(
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(401, None)),
@@ -7429,8 +7400,7 @@ async def test_delegate_not_probed_when_named_only_via_server_id():
         await _check_passthrough_upstream_auth(
             scope=scope,
             user_api_key_auth=UserAPIKeyAuth(user_id="u1", api_key="hashed-sk"),
-            mcp_servers=["delegate-secret-id"],
-            client_ip=None,
+            allowed_servers=[server],
             oauth2_headers=None,
             mcp_server_auth_headers=None,
             raw_headers=None,
@@ -7452,10 +7422,6 @@ async def test_delegate_probe_not_fanned_out_to_access_group_members():
 
     with (
         _patch_delegate_resolver(group_member, "delegate_test"),
-        patch(
-            "litellm.proxy._experimental.mcp_server.operations._get_allowed_mcp_servers",
-            new=AsyncMock(return_value=[group_member]),
-        ),
         patch(
             "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
             new=AsyncMock(return_value=(401, None)),
@@ -7464,8 +7430,7 @@ async def test_delegate_probe_not_fanned_out_to_access_group_members():
         await _check_passthrough_upstream_auth(
             scope=_delegate_scope([(b"authorization", b"Bearer bogus-token")]),
             user_api_key_auth=UserAPIKeyAuth(),
-            mcp_servers=["prod_tools_group"],
-            client_ip=None,
+            allowed_servers=[group_member],
             oauth2_headers=None,
             mcp_server_auth_headers=None,
             raw_headers=None,
@@ -10326,7 +10291,10 @@ class TestPreemptive401GatedByKeyAllowlist:
         allowed_servers: list[MCPServer],
         handle_target: StreamableHTTPSessionManager | None,
         toolset_auth: UserAPIKeyAuth | None = None,
-    ) -> Iterator[AsyncMock]:
+        mcp_servers: list[str] | None = None,
+        oauth2_headers: dict[str, str] | None = None,
+        raw_headers: dict[str, str] | None = None,
+    ) -> Iterator[tuple[AsyncMock, AsyncMock]]:
         def _resolve_allowed(
             user_api_key_auth: UserAPIKeyAuth | None = None,
             mcp_servers: list[str] | None = None,
@@ -10347,10 +10315,17 @@ class TestPreemptive401GatedByKeyAllowlist:
                     patch(  # test-quality-ok: admission parse is unrelated to the challenge gate; the handler reads it as a module global
                         "litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
                         new_callable=AsyncMock,
-                        return_value=(self._auth(), None, [server.alias], None, None, None),
+                        return_value=(
+                            self._auth(),
+                            None,
+                            mcp_servers if mcp_servers is not None else [server.alias],
+                            None,
+                            oauth2_headers,
+                            raw_headers,
+                        ),
                     )
                 )
-                stack.enter_context(
+                allowed_servers_lookup: Final = stack.enter_context(
                     patch(  # test-quality-ok: allowed-server resolution hits the key/toolset DB; the test must control the allowlist
                         "litellm.proxy._experimental.mcp_server.operations._get_allowed_mcp_servers",
                         new_callable=AsyncMock,
@@ -10402,7 +10377,7 @@ class TestPreemptive401GatedByKeyAllowlist:
                             new=handle_request,
                         )
                     )
-                yield handle_request
+                yield allowed_servers_lookup, handle_request
         finally:
             registry.pop(server.server_id, None)
 
@@ -10418,7 +10393,7 @@ class TestPreemptive401GatedByKeyAllowlist:
         with (
             self._patches(
                 server, allowed_servers=[], handle_target=server_module.session_manager_stateful
-            ) as handle_request,
+            ) as (_, handle_request),
             pytest.raises(HTTPException) as exc,
         ):
             await server_module.handle_streamable_http_mcp(scope, receive, AsyncMock())
@@ -10438,7 +10413,7 @@ class TestPreemptive401GatedByKeyAllowlist:
         with (
             self._patches(
                 server, allowed_servers=[server], handle_target=server_module.session_manager_stateful
-            ) as handle_request,
+            ) as (_, handle_request),
             pytest.raises(HTTPException) as exc,
         ):
             await server_module.handle_streamable_http_mcp(scope, receive, AsyncMock())
@@ -10447,6 +10422,83 @@ class TestPreemptive401GatedByKeyAllowlist:
         assert "www-authenticate" in {k.lower() for k in exc.value.headers}
         handle_request.assert_not_awaited()
 
+    @pytest.mark.asyncio
+    async def test_streamable_passthrough_reuses_allowed_servers_for_probe(self, monkeypatch: pytest.MonkeyPatch):
+        from litellm.proxy._experimental.mcp_server import server as server_module
+
+        monkeypatch.setattr("litellm.proxy.proxy_server.general_settings", {"mcp_advertised_versions": ["2026-07-28"]})
+        server: Final = MCPServer(
+            server_id="id-restricted",
+            name="restricted",
+            alias="restricted",
+            server_name="restricted",
+            url="https://restricted.test/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.none,
+            extra_headers=["Authorization"],
+            oauth_passthrough=True,
+        )
+        body: Final = json.dumps(
+            {
+                "jsonrpc": "2.0",
+                "id": 7,
+                "method": "tools/call",
+                "params": {
+                    "name": "confirm",
+                    "arguments": {},
+                    "_meta": {
+                        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
+                        "io.modelcontextprotocol/clientCapabilities": {},
+                    },
+                },
+            }
+        ).encode()
+        scope: Final = {
+            "type": "http",
+            "method": "POST",
+            "path": "/mcp",
+            "scheme": "http",
+            "server": ("localhost", 8000),
+            "query_string": b"",
+            "root_path": "",
+            "headers": [
+                (b"content-type", b"application/json"),
+                (b"x-litellm-api-key", b"sk-litellm-virtual-key"),
+                (b"authorization", b"Bearer upstream-token"),
+                (b"mcp-protocol-version", b"2026-07-28"),
+                (b"mcp-method", b"tools/call"),
+                (b"mcp-name", b"confirm"),
+                (b"accept", b"application/json, text/event-stream"),
+            ],
+        }
+        receive: Final = AsyncMock(
+            return_value={"type": "http.request", "body": body, "more_body": False}
+        )
+
+        with (
+            self._patches(
+                server,
+                allowed_servers=[server],
+                handle_target=server_module.session_manager_stateless,
+                mcp_servers=["restricted", "another"],
+                oauth2_headers={"Authorization": "Bearer upstream-token"},
+                raw_headers={
+                    "x-litellm-api-key": "sk-litellm-virtual-key",
+                    "Authorization": "Bearer upstream-token",
+                },
+            ) as (allowed_servers_lookup, handle_request),
+            patch(
+                "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth",
+                new_callable=AsyncMock,
+                return_value=(200, None),
+            ) as probe,
+        ):
+            await server_module.handle_streamable_http_mcp(scope, receive, AsyncMock())
+
+        allowed_servers_lookup.assert_awaited_once()
+        probe.assert_awaited_once_with(server.url, "Bearer upstream-token")
+        handle_request.assert_awaited_once()
+
     @pytest.mark.asyncio
     async def test_toolset_excluding_server_gets_403_not_401_streamable(self):
         """Key allowlist includes the server but the active toolset scope
@@ -10473,7 +10525,7 @@ class TestPreemptive401GatedByKeyAllowlist:
                     allowed_servers=[server],
                     handle_target=server_module.session_manager_stateful,
                     toolset_auth=scoped_auth,
-                ) as handle_request,
+                ) as (_, handle_request),
                 pytest.raises(HTTPException) as exc,
             ):
                 await server_module.handle_streamable_http_mcp(scope, receive, AsyncMock())
@@ -11811,6 +11863,7 @@ async def test_passthrough_probes_bind_each_token_to_its_authorized_server(allow
         )
         for name in ("first", "second")
     ]
+    allowed_servers: Final = targets if allowed else []
     mcp_operations.global_mcp_server_manager.registry.update({target.server_id: target for target in targets})
     with respx.mock(assert_all_called=False) as upstream:
         first: Final = upstream.post(targets[0].url).respond(200)
@@ -11825,8 +11878,7 @@ async def test_passthrough_probes_bind_each_token_to_its_authorized_server(allow
                     mcp_servers=[target.server_id for target in targets] if allowed else [],
                 ),
             ),
-            mcp_servers=["first", "second"],
-            client_ip=None,
+            allowed_servers=allowed_servers,
             oauth2_headers=None,
             mcp_server_auth_headers={
                 "first": {"x-other" if header_shape == "non_authorization" else "aUtHoRiZaTiOn": "Bearer first-token"},

From 645f13a7f8c0bfc6a2495a24b7eae2e298fc0385 Mon Sep 17 00:00:00 2001
From: jesus <jesus@berri.ai>
Date: Sat, 17 Oct 2026 02:11:39 +0000
Subject: [PATCH 7/8] test(mcp): mark the single allowlist lookup test as a
 count contract

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 .../mcp_server/test_mcp_server_tool_calls_and_headers.py      | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py b/tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py
index 7d98dfe5d33..2b9b3806121 100644
--- a/tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py
+++ b/tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py
@@ -10423,7 +10423,9 @@ class TestPreemptive401GatedByKeyAllowlist:
         handle_request.assert_not_awaited()
 
     @pytest.mark.asyncio
-    async def test_streamable_passthrough_reuses_allowed_servers_for_probe(self, monkeypatch: pytest.MonkeyPatch):
+    async def test_streamable_probe_reuses_allowed_servers(  # test-quality-ok: one lookup per request is the contract
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
         from litellm.proxy._experimental.mcp_server import server as server_module
 
         monkeypatch.setattr("litellm.proxy.proxy_server.general_settings", {"mcp_advertised_versions": ["2026-07-28"]})

From af46f4ca856d416b16a15e5fd24b68701a6a9db7 Mon Sep 17 00:00:00 2001
From: jesus <jesus@berri.ai>
Date: Sat, 17 Oct 2026 22:41:02 +0000
Subject: [PATCH 8/8] fix(mcp): keep the OAuth challenge for keyless cold-start
 MCP requests

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 .../mcp_server/mcp_server_manager.py          | 20 ++--
 .../proxy/_experimental/mcp_server/server.py  | 21 +++-
 .../test_mcp_server_tool_calls_and_headers.py | 98 ++++++++++++++++++-
 3 files changed, 126 insertions(+), 13 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
index b724a61cd94..fa7bfde807a 100644
--- a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
+++ b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
@@ -1853,6 +1853,17 @@ def _listing_task_was_cancelled(error: asyncio.CancelledError) -> bool:
     return error.args != (SESSION_CLEANUP_TIMEOUT_MESSAGE,)
 
 
+def is_anonymous_mcp_caller(user_api_key_auth: UserAPIKeyAuth | None) -> bool:
+    return not (
+        user_api_key_auth
+        and (
+            getattr(user_api_key_auth, "user_id", None)
+            or getattr(user_api_key_auth, "user_role", None)
+            or getattr(user_api_key_auth, "api_key", None)
+        )
+    )
+
+
 class MCPServerManager:
     _STDIO_ENV_TEMPLATE_PATTERN = re.compile(r"^\$\{(X-[^}]+)\}$")
 
@@ -3714,14 +3725,7 @@ class MCPServerManager:
                 )
             )
 
-            is_anonymous: Final = not (
-                user_api_key_auth
-                and (
-                    getattr(user_api_key_auth, "user_id", None)
-                    or getattr(user_api_key_auth, "user_role", None)
-                    or getattr(user_api_key_auth, "api_key", None)
-                )
-            )
+            is_anonymous: Final = is_anonymous_mcp_caller(user_api_key_auth)
             if is_anonymous:
                 passthrough_server_ids: Final = [
                     server.server_id
diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py
index 9592e5a6321..fc3aadab8c4 100644
--- a/litellm/proxy/_experimental/mcp_server/server.py
+++ b/litellm/proxy/_experimental/mcp_server/server.py
@@ -578,6 +578,7 @@ if MCP_AVAILABLE:
     from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
         MCPServerManager,
         global_mcp_server_manager,
+        is_anonymous_mcp_caller,
     )
 
     ######################################################
@@ -1828,6 +1829,15 @@ if MCP_AVAILABLE:
             return allowed_ids
         return allowed_ids & frozenset(toolset_allowed_server_ids)
 
+    def _challenge_server_ids(
+        user_api_key_auth: UserAPIKeyAuth | None,
+        allowed_servers: Sequence[MCPServer],
+        toolset_allowed_server_ids: Set[str] | None,
+    ) -> frozenset[str] | None:
+        if is_anonymous_mcp_caller(user_api_key_auth):
+            return None if toolset_allowed_server_ids is None else frozenset(toolset_allowed_server_ids)
+        return _narrow_allowed_server_ids(allowed_servers, toolset_allowed_server_ids)
+
     _apply_toolset_scope: Final = apply_toolset_scope
 
     async def _toolset_server_ids(toolset_id: str) -> set[str]:
@@ -1854,7 +1864,8 @@ if MCP_AVAILABLE:
         (points clients at the upstream resource-metadata via our well-known).
 
         ``allowed_server_ids`` is the set of servers the caller's key may
-        reach, narrowed further by any active toolset scope; servers not in
+        reach, narrowed further by any active toolset scope; anonymous
+        cold-start callers are not narrowed by a key allowlist. Servers not in
         that set are skipped so a client is not pushed into an OAuth flow
         for a server it will be 403'd on immediately after authentication.
         """
@@ -2331,7 +2342,9 @@ if MCP_AVAILABLE:
                 ),
                 user_api_key_auth=user_api_key_auth,
                 client_ip=client_ip,
-                allowed_server_ids=_narrow_allowed_server_ids(allowed_servers, toolset_allowed_server_ids),
+                allowed_server_ids=_challenge_server_ids(
+                    user_api_key_auth, allowed_servers, toolset_allowed_server_ids
+                ),
                 raw_headers=raw_headers,
             )
             # Pre-flight auth check for pass-through servers.  Must run after
@@ -2835,7 +2848,9 @@ if MCP_AVAILABLE:
                 mcp_server_auth_headers=mcp_server_auth_headers,
                 user_api_key_auth=user_api_key_auth,
                 client_ip=_sse_client_ip,
-                allowed_server_ids=_narrow_allowed_server_ids(allowed_servers, toolset_allowed_server_ids),
+                allowed_server_ids=_challenge_server_ids(
+                    user_api_key_auth, allowed_servers, toolset_allowed_server_ids
+                ),
                 raw_headers=raw_headers,
             )
 
diff --git a/tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py b/tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py
index 2b9b3806121..44260ae1814 100644
--- a/tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py
+++ b/tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py
@@ -10291,6 +10291,7 @@ class TestPreemptive401GatedByKeyAllowlist:
         allowed_servers: list[MCPServer],
         handle_target: StreamableHTTPSessionManager | None,
         toolset_auth: UserAPIKeyAuth | None = None,
+        request_auth: UserAPIKeyAuth | None = None,
         mcp_servers: list[str] | None = None,
         oauth2_headers: dict[str, str] | None = None,
         raw_headers: dict[str, str] | None = None,
@@ -10316,7 +10317,7 @@ class TestPreemptive401GatedByKeyAllowlist:
                         "litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
                         new_callable=AsyncMock,
                         return_value=(
-                            self._auth(),
+                            request_auth if request_auth is not None else self._auth(),
                             None,
                             mcp_servers if mcp_servers is not None else [server.alias],
                             None,
@@ -10422,6 +10423,90 @@ class TestPreemptive401GatedByKeyAllowlist:
         assert "www-authenticate" in {k.lower() for k in exc.value.headers}
         handle_request.assert_not_awaited()
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("server_kind", "transport"),
+        [("pt", "streamable"), ("pt", "sse"), ("dl", "streamable"), ("dl", "sse")],
+    )
+    async def test_keyless_cold_start_still_gets_oauth_challenge(self, server_kind: str, transport: str) -> None:
+        from litellm.proxy._experimental.mcp_server import server as server_module
+
+        server: Final = MCPServer(
+            server_id=f"id-{server_kind}",
+            name=server_kind,
+            alias=server_kind,
+            server_name=server_kind,
+            url=None,
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.none if server_kind == "pt" else MCPAuth.oauth2,
+            extra_headers=["Authorization"] if server_kind == "pt" else [],
+            oauth_passthrough=server_kind == "pt",
+            delegate_auth_to_upstream=server_kind == "dl",
+            oauth2_flow="authorization_code" if server_kind == "dl" else None,
+            authorization_url="https://idp.test/authorize" if server_kind == "dl" else None,
+            token_url="https://idp.test/token" if server_kind == "dl" else None,
+            mcp_info={"server_name": server_kind},
+        )
+        body: Final = (
+            b'{"jsonrpc":"2.0","id":1,"method":"initialize","params":'
+            b'{"protocolVersion":"2024-11-05","capabilities":{},'
+            b'"clientInfo":{"name":"test","version":"0"}}}'
+        )
+        scope: Final = {
+            "type": "http",
+            "method": "POST" if transport == "streamable" else "GET",
+            "path": f"/mcp/{server_kind}",
+            "scheme": "http",
+            "server": ("127.0.0.1", 4000),
+            "root_path": "",
+            "query_string": b"",
+            "headers": [
+                (b"host", b"127.0.0.1:4000"),
+                (b"accept", b"application/json, text/event-stream"),
+                (b"content-type", b"application/json"),
+            ],
+        }
+        receive: Final = AsyncMock(return_value={"type": "http.request", "body": body, "more_body": False})
+        registry: Final = mcp_operations.global_mcp_server_manager.registry
+        expected_challenge: Final = (
+            f'Bearer resource_metadata="http://127.0.0.1:4000/.well-known/oauth-protected-resource/mcp/{server_kind}"'
+        )
+
+        with (
+            patch.dict(registry, {server.server_id: server}),
+            patch(  # test-quality-ok: force the admitted anonymous cold-start auth context without DB or network
+                "litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
+                new_callable=AsyncMock,
+                return_value=(UserAPIKeyAuth(), None, [server.alias], None, None, None),
+            ),
+            patch(  # test-quality-ok: module-level startup flag gates the transport, with no injectable seam
+                "litellm.proxy._experimental.mcp_server.server._SESSION_MANAGERS_INITIALIZED", True
+            ),
+        ):
+            if transport == "streamable":
+                handle_request: Final = AsyncMock()
+                with (
+                    patch.object(
+                        server_module.session_manager_stateful,
+                        "handle_request",
+                        new=handle_request,
+                    ),  # test-quality-ok: rejected requests must not dispatch to the session manager
+                    pytest.raises(HTTPException) as exc,
+                ):
+                    await server_module.handle_streamable_http_mcp(scope, receive, AsyncMock())
+                handle_request.assert_not_awaited()
+            else:
+                with (
+                    patch.object(server_module.sse, "connect_sse") as connect_sse,
+                    pytest.raises(HTTPException) as exc,
+                ):
+                    await server_module.handle_sse_mcp(scope, receive, AsyncMock())
+                connect_sse.assert_not_called()
+
+        assert exc.value.status_code == 401
+        assert exc.value.headers is not None
+        assert exc.value.headers["www-authenticate"] == expected_challenge
+
     @pytest.mark.asyncio
     async def test_streamable_probe_reuses_allowed_servers(  # test-quality-ok: one lookup per request is the contract
         self, monkeypatch: pytest.MonkeyPatch
@@ -10510,7 +10595,15 @@ class TestPreemptive401GatedByKeyAllowlist:
         from litellm.proxy._types import LiteLLM_ObjectPermissionTable
 
         server: Final = _make_oauth2_server("restricted")
-        scoped_auth: Final = self._auth().model_copy(
+        key_auth: Final = self._auth().model_copy(
+            update={
+                "object_permission": LiteLLM_ObjectPermissionTable(
+                    object_permission_id="key-scope",
+                    mcp_servers=[server.server_id],
+                )
+            }
+        )
+        scoped_auth: Final = key_auth.model_copy(
             update={
                 "object_permission": LiteLLM_ObjectPermissionTable(
                     object_permission_id="toolset-scope",
@@ -10527,6 +10620,7 @@ class TestPreemptive401GatedByKeyAllowlist:
                     allowed_servers=[server],
                     handle_target=server_module.session_manager_stateful,
                     toolset_auth=scoped_auth,
+                    request_auth=key_auth,
                 ) as (_, handle_request),
                 pytest.raises(HTTPException) as exc,
             ):