feat(mcp): admit gateway DCR session bearers at the aggregate /mcp scope

Wires the identity-only session token into admission, so a client that completed the gateway
OAuth flow can actually call the aggregate endpoint. The arm sits after the explicit
x-litellm-api-key arm, which keeps winning, and before the generic Authorization arm, which
would otherwise hand the bearer to user_api_key_auth as if it were a litellm key and 401.

The bearer carries no authorization. It reloads the live user record and runs the same
centralized policy gate the key path uses, so deactivating a user, blocking their team, or
exhausting a budget kills outstanding sessions on the next call rather than at expiry. Failures
raise the aggregate challenge, so a spec client re-authorizes instead of giving up; a session
refresh token presented at the tool edge is rejected the same way rather than falling through.

Three ways an admitted subject was quietly weaker than a key, fixed here:

An admin connecting through the front door received every MCP server on the proxy, because
get_allowed_mcp_servers short-circuits to the whole registry for an admin role with no explicit
object permission. That is the dashboard's view-all behavior; a gateway session is excluded from
it, so the connect grid keeps describing what the client can reach.

The org ceiling and the org budget check both return early without an org_id, which the reload
never bound, so neither applied. A ceiling only narrows, so binding it cannot widen access.

Every rate limiter descriptor keys on api_key, team_id or end_user_id, none of which a keyless
subject has, so the limiter enforced nothing at all and logged nothing. The user-level limits
are the only ones this principal can bind, so the reload now carries them.

Extracts _is_aggregate_mcp_scope so the 401 challenge and this arm share one definition of the
aggregate scope instead of two that can drift.

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

@@ -25,6 +25,14 @@ from litellm.proxy._experimental.mcp_server.outbound_credentials.bridge_credenti
 from litellm.proxy._experimental.mcp_server.outbound_credentials.envelope import (
     EnvelopeIdentity,
 )
+from litellm.proxy._experimental.mcp_server.outbound_credentials.session_credentials import (
+    NotSessionBearer,
+    SessionBearerAdmitted,
+    SessionBearerInvalid,
+    is_session_bearer_shaped,
+    resolve_session_bearer,
+    session_keys_from_master_key,
+)
 from litellm.proxy._types import (
     UI_TEAM_ID,
     LiteLLM_TeamTable,
@@ -124,6 +132,25 @@ def _has_client_supplied_mcp_auth(
     return bool(mcp_auth_header) or bool(mcp_server_auth_headers)
 
 
+def _is_aggregate_mcp_scope(
+    route: str,
+    mcp_servers: list[str] | None,
+    mcp_auth_header: str | None,
+    mcp_server_auth_headers: dict[str, dict[str, str]] | None,
+) -> bool:
+    """True when a request targets the aggregate ``/mcp`` endpoint rather than any named server.
+
+    Aggregate scope means no target resolves from the path or ``x-mcp-servers``, and the caller
+    supplied no per-server MCP auth headers (which would make it a scripted client, not a
+    cold-start DCR one). Both the 401 challenge and the gateway session arm consume this, so the
+    scope the challenge advertises and the scope the session is admitted at cannot drift."""
+    if mcp_servers:
+        return False
+    if _has_client_supplied_mcp_auth(mcp_auth_header, mcp_server_auth_headers):
+        return False
+    return len(MCPRequestHandler._resolve_target_server_names(route, mcp_servers)) == 0
+
+
 def _is_aggregate_gateway_dcr_challenge_scope(
     route: str,
     mcp_servers: list[str] | None,
@@ -135,17 +162,11 @@ def _is_aggregate_gateway_dcr_challenge_scope(
     should receive the RFC 9728 401 challenge that advertises the gateway as
     the authorization server.
 
-    Fires only for a genuine 401 on the aggregate scope: any named target
-    (path or ``x-mcp-servers``) belongs to the per-server challenge paths, and
-    client-supplied MCP auth headers mean the caller is not a cold-start DCR
-    client. Fails closed to the original admission error otherwise."""
+    Fires only for a genuine 401 on the aggregate scope. Fails closed to the
+    original admission error otherwise."""
     if not _is_litellm_auth_admission_error(exc):
         return False
-    if mcp_servers:
-        return False
-    if _has_client_supplied_mcp_auth(mcp_auth_header, mcp_server_auth_headers):
-        return False
-    return len(MCPRequestHandler._extract_target_server_names_from_path(route)) == 0
+    return _is_aggregate_mcp_scope(route, mcp_servers, mcp_auth_header, mcp_server_auth_headers)
 
 
 def _aggregate_gateway_dcr_challenge(request: Request, invalid_token: bool) -> HTTPException:
@@ -317,6 +338,20 @@ class MCPRequestHandler:
             # for a delegated server, so validate it: identity / spend / rate
             # limits resolve and any stored upstream token can be forwarded.
             validated_user_api_key_auth = await user_api_key_auth(api_key=litellm_api_key, request=request)
+        elif (
+            oauth2_headers
+            and is_session_bearer_shaped(oauth2_headers["Authorization"])
+            and _is_aggregate_mcp_scope(request_route, mcp_servers, mcp_auth_header, mcp_server_auth_headers)
+        ):
+            # A gateway DCR session bearer at the aggregate scope. Placed after the explicit
+            # x-litellm-api-key arm, which keeps winning, and before the generic Authorization arm,
+            # which would otherwise feed this bearer to user_api_key_auth as if it were a litellm
+            # key and 401 on it.
+            validated_user_api_key_auth = await MCPRequestHandler._admit_gateway_session(
+                authorization_value=oauth2_headers["Authorization"],
+                request=request,
+                route=request_route,
+            )
         elif MCPRequestHandler._target_servers_delegate_auth_to_upstream(
             path=request_route,
             mcp_servers=mcp_servers,
@@ -626,6 +661,52 @@ class MCPRequestHandler:
             case _:
                 assert_never(result)
 
+    @staticmethod
+    async def _admit_gateway_session(
+        authorization_value: str,
+        request: Request,
+        route: str,
+    ) -> UserAPIKeyAuth:
+        """Admit a gateway DCR session bearer at the aggregate ``/mcp`` scope.
+
+        The bearer proves the user completed SSO when it was minted; it carries no authorization,
+        so everything that decides what the request may reach is resolved fresh here. The recovered
+        ``user_id`` reloads the live user record and the same centralized policy gate the key path
+        uses runs over it, which is what makes deactivating a user, blocking their team, or
+        exhausting a budget kill outstanding sessions on the next call instead of at expiry.
+
+        No upstream credential is injected, unlike the bridge envelope arm: the custody model vaults
+        every upstream token server-side, and egress resolves it by user at call time.
+
+        Fails closed with the aggregate challenge rather than a bare 401, so a spec-compliant client
+        re-runs the authorization flow instead of giving up. A session REFRESH token presented here
+        is rejected the same way: it is a valid gateway credential, but only ever at the token
+        endpoint.
+        """
+        from litellm.proxy.proxy_server import master_key
+
+        if not master_key:
+            raise HTTPException(status_code=500, detail="Server misconfigured: master_key is not set")
+
+        await MCPRequestHandler._run_pre_db_read_auth_checks(request=request, route=route)
+
+        result = resolve_session_bearer(
+            authorization_value,
+            session_keys_from_master_key(master_key),
+            datetime.now(timezone.utc),
+        )
+        match result:
+            case SessionBearerAdmitted():
+                admitted = await MCPRequestHandler._reload_admitted_user(
+                    result.principal.user_id, is_mcp_gateway_session=True
+                )
+                await MCPRequestHandler._enforce_admitted_live_policy(admitted=admitted, request=request, route=route)
+                return admitted
+            case SessionBearerInvalid() | NotSessionBearer():
+                raise _aggregate_gateway_dcr_challenge(request, invalid_token=True)
+            case _:
+                assert_never(result)
+
     @staticmethod
     async def _run_pre_db_read_auth_checks(request: Request, route: str) -> None:
         """Run the proxy-wide gates ``user_api_key_auth`` applies before any key lookup: the
@@ -665,7 +746,7 @@ class MCPRequestHandler:
                 assert_never(identity.subject_type)
 
     @staticmethod
-    async def _reload_admitted_user(user_id: str) -> UserAPIKeyAuth:
+    async def _reload_admitted_user(user_id: str, is_mcp_gateway_session: bool = False) -> UserAPIKeyAuth:
         """Reload the live user an interactively-minted envelope references and admit them as
         themselves.
 
@@ -726,6 +807,20 @@ class MCPRequestHandler:
             user_role=user_object.user_role,
             object_permission=object_permission,
             object_permission_id=user_object.object_permission_id,
+            # Bind the org so the org MCP ceiling and the org budget check, both of which return
+            # early without an org_id, actually apply to an admitted subject. A ceiling only ever
+            # narrows, so this cannot widen what the subject reaches.
+            org_id=user_object.organization_id,
+            # The user-level limits are the ONLY rate limits this principal can bind: every other
+            # limiter descriptor keys on api_key, team_id, or end_user_id, none of which a keyless
+            # subject has. Without them _create_rate_limit_descriptors returns an empty list and
+            # the limiter silently enforces nothing.
+            user_rpm_limit=user_object.rpm_limit,
+            user_tpm_limit=user_object.tpm_limit,
+            user_email=user_object.user_email,
+            user_max_budget=user_object.max_budget,
+            user_spend=user_object.spend,
+            is_mcp_gateway_session=is_mcp_gateway_session,
         )
 
     @staticmethod

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -2234,8 +2234,17 @@ class MCPServerManager:
         )
 
         try:
-            # If admin but NO explicit object permission, get all servers
-            if user_api_key_auth and _user_has_admin_view(user_api_key_auth) and not has_explicit_object_permission:
+            # If admin but NO explicit object permission, get all servers.
+            # A gateway DCR session is excluded: it is a keyless subject that a public OAuth client
+            # holds, so letting it inherit the role-based view-all short-circuit would hand an
+            # admin's MCP client every server on the proxy, and the connect grid the user
+            # authorized servers on would no longer describe what their client can reach.
+            if (
+                user_api_key_auth
+                and not user_api_key_auth.is_mcp_gateway_session
+                and _user_has_admin_view(user_api_key_auth)
+                and not has_explicit_object_permission
+            ):
                 verbose_logger.debug("Admin user without explicit object_permission - returning all servers")
                 return list(self.get_registry().keys())
 

litellm/proxy/_types.py

@@ -2605,6 +2605,10 @@ class UserAPIKeyAuth(LiteLLM_VerificationTokenView):  # the expected response ob
     user_max_budget: Optional[float] = None
     request_route: Optional[str] = None
     is_session_token: bool = False
+    # Admitted through the aggregate gateway DCR front door: a keyless subject whose reach comes
+    # entirely from its own grants. Distinct from is_session_token, which means the dashboard UI
+    # session and carries budget semantics of its own.
+    is_mcp_gateway_session: bool = False
     budget_reservation: Optional[Dict[str, Any]] = Field(default=None, exclude=True)
     budget_throttle_pct: Optional[float] = Field(default=None, exclude=True)
     user: Optional[Any] = None  # Expanded user object when expand=user is used

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

@@ -5069,6 +5069,26 @@ class TestMCPDcrBridgeDelegateAdmission:
 
     _MASTER_KEY = "sk-bridge-master-key-for-envelope-derivation"
 
+    @staticmethod
+    def _user_row(**overrides):
+        """A real user row rather than a MagicMock: the reload reads typed fields off it, and a
+        mock silently answers every attribute, so assertions against it can pass vacuously."""
+        from litellm.proxy._types import LiteLLM_UserTable
+
+        return LiteLLM_UserTable(
+            **{
+                "user_id": "sso-user-7",
+                "user_role": None,
+                "max_budget": None,
+                "spend": 0.0,
+                "models": [],
+                "metadata": {"scim_active": True},
+                "object_permission": None,
+                "object_permission_id": None,
+                **overrides,
+            }
+        )
+
     @staticmethod
     def _bridge_delegate_server(server_name="bridge_delegate_server", dcr_bridge=True, alias=None):
         from litellm.types.mcp import MCPAuth
@@ -5299,15 +5319,7 @@ class TestMCPDcrBridgeDelegateAdmission:
             ) as mock_auth,
             patch("litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager") as mock_mgr,
             patch("litellm.proxy.proxy_server.master_key", self._MASTER_KEY),
-            self._patch_user_reload(
-                return_value=MagicMock(
-                    user_id="sso-user-7",
-                    metadata={"scim_active": True},
-                    user_role=None,
-                    object_permission=None,
-                    object_permission_id=None,
-                )
-            ) as get_user_object,
+            self._patch_user_reload(return_value=self._user_row()) as get_user_object,
         ):
             mock_mgr.get_mcp_server_by_name.return_value = self._bridge_delegate_server()
             (auth_result, _h, _s, mcp_server_auth_headers, _o, _r) = await MCPRequestHandler.process_mcp_request(scope)
@@ -5342,10 +5354,7 @@ class TestMCPDcrBridgeDelegateAdmission:
             patch("litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager") as mock_mgr,
             patch("litellm.proxy.proxy_server.master_key", self._MASTER_KEY),
             self._patch_user_reload(
-                return_value=MagicMock(
-                    user_id="sso-user-7",
-                    metadata={"scim_active": True},
-                    user_role=None,
+                return_value=self._user_row(
                     object_permission=object_permission,
                     object_permission_id="op-user-7",
                 )
@@ -6261,3 +6270,193 @@ class TestAggregateGatewayDcrChallenge:
             with pytest.raises(ProxyException) as exc_info:
                 await MCPRequestHandler.process_mcp_request(self._scope())
         assert str(exc_info.value.code) == "500"
+
+
+@pytest.mark.asyncio
+class TestGatewayDCRSessionAdmission:
+    """The aggregate front door's admission arm.
+
+    A gateway DCR session bearer carries identity only, so these tests pin that authorization is
+    resolved fresh from the live user record on every call, that the arm fails closed with the
+    challenge a spec client can act on, and that it never fires outside the aggregate scope.
+    """
+
+    _MASTER_KEY = "sk-gateway-session-tests"
+    _AUTH_PATCH_TARGET = "litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp.user_api_key_auth"
+
+    def _bearer(self, user_id: str = "user-1", client_id: str = "llm_dcrc_client") -> str:
+        from datetime import datetime, timezone
+
+        from litellm.proxy._experimental.mcp_server.outbound_credentials.session_credentials import (
+            session_keys_from_master_key,
+        )
+        from litellm.proxy._experimental.mcp_server.outbound_credentials.session_token import (
+            SessionPrincipal,
+            mint_session_token,
+        )
+
+        minted = mint_session_token(
+            SessionPrincipal(user_id=user_id, client_id=client_id),
+            session_keys_from_master_key(self._MASTER_KEY),
+            datetime.now(timezone.utc),
+        )
+        return minted.token.get_secret_value()
+
+    def _refresh_bearer(self, user_id: str = "user-1") -> str:
+        from datetime import datetime, timezone
+
+        from litellm.proxy._experimental.mcp_server.outbound_credentials.session_credentials import (
+            session_keys_from_master_key,
+        )
+        from litellm.proxy._experimental.mcp_server.outbound_credentials.session_token import (
+            SessionPrincipal,
+            mint_session_refresh_token,
+        )
+
+        minted = mint_session_refresh_token(
+            SessionPrincipal(user_id=user_id, client_id="llm_dcrc_client"),
+            session_keys_from_master_key(self._MASTER_KEY),
+            datetime.now(timezone.utc),
+        )
+        return minted.token.get_secret_value()
+
+    def _scope(self, bearer: str, path: str = "/mcp", extra_headers=None):
+        headers = [(b"authorization", f"Bearer {bearer}".encode())]
+        headers.extend(extra_headers or [])
+        return {"type": "http", "method": "POST", "path": path, "headers": headers}
+
+    def _user_row(self, **overrides):
+        from litellm.proxy._types import LiteLLM_UserTable
+
+        return LiteLLM_UserTable(
+            **{
+                "user_id": "user-1",
+                "user_role": "internal_user",
+                "max_budget": None,
+                "spend": 0.0,
+                "models": [],
+                "organization_id": "org-42",
+                "rpm_limit": 11,
+                "tpm_limit": 2222,
+                "user_email": "user@example.com",
+                **overrides,
+            }
+        )
+
+    @contextlib.contextmanager
+    def _rig(self, user_row=None):
+        """Real session crypto, stubbed persistence and policy gate.
+
+        Only the DB read and the shared policy gate are stubbed; the bearer is genuinely minted and
+        genuinely opened, so a break in the crypto or the arm's wiring surfaces here.
+        """
+        from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import MCPRequestHandler
+
+        async def _no_pre_db_checks(request, route):
+            return None
+
+        async def _no_policy(admitted, request, route):
+            return None
+
+        async def _get_user_object(**kwargs):
+            return user_row if user_row is not None else self._user_row()
+
+        with (
+            patch("litellm.proxy.proxy_server.master_key", self._MASTER_KEY),
+            patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
+            patch.object(MCPRequestHandler, "_run_pre_db_read_auth_checks", _no_pre_db_checks),
+            patch.object(MCPRequestHandler, "_enforce_admitted_live_policy", _no_policy),
+            patch("litellm.proxy.auth.auth_checks.get_user_object", _get_user_object),
+        ):
+            yield
+
+    async def test_a_valid_session_bearer_is_admitted_as_the_sealed_user(self):
+        with self._rig():
+            auth, *_rest = await MCPRequestHandler.process_mcp_request(self._scope(self._bearer()))
+
+        assert auth.user_id == "user-1"
+        assert auth.is_mcp_gateway_session is True
+        assert auth.api_key is None
+
+    async def test_the_admitted_subject_binds_its_org_so_the_org_ceiling_applies(self):
+        """get_allowed_mcp_servers and the org budget check both return early without an org_id,
+        so an unbound org means the ceiling silently never applies."""
+        with self._rig():
+            auth, *_rest = await MCPRequestHandler.process_mcp_request(self._scope(self._bearer()))
+
+        assert auth.org_id == "org-42"
+
+    async def test_the_admitted_subject_binds_the_only_rate_limits_it_can(self):
+        """Every other limiter descriptor keys on api_key, team_id or end_user_id, none of which a
+        keyless subject has; without the user-level limits the limiter enforces nothing at all."""
+        with self._rig():
+            auth, *_rest = await MCPRequestHandler.process_mcp_request(self._scope(self._bearer()))
+
+        assert (auth.user_rpm_limit, auth.user_tpm_limit) == (11, 2222)
+
+    async def test_an_expired_or_tampered_bearer_gets_the_reauthorize_challenge(self):
+        tampered = self._bearer()[:-3] + "AAA"
+
+        with self._rig():
+            with pytest.raises(HTTPException) as exc_info:
+                await MCPRequestHandler.process_mcp_request(self._scope(tampered))
+
+        assert exc_info.value.status_code == 401
+        assert 'error="invalid_token"' in exc_info.value.headers["WWW-Authenticate"]
+        assert "resource_metadata=" in exc_info.value.headers["WWW-Authenticate"]
+
+    async def test_a_refresh_token_is_never_usable_at_the_tool_call_edge(self):
+        """A refresh token is a valid gateway credential, but only at the token endpoint. It must
+        fail closed here rather than fall through to another admission arm."""
+        with self._rig():
+            with pytest.raises(HTTPException) as exc_info:
+                await MCPRequestHandler.process_mcp_request(self._scope(self._refresh_bearer()))
+
+        assert exc_info.value.status_code == 401
+        assert 'error="invalid_token"' in exc_info.value.headers["WWW-Authenticate"]
+
+    async def test_a_session_bearer_does_not_admit_at_a_named_server_scope(self):
+        """The session grants the aggregate scope. A named target belongs to the per-server paths,
+        so the bearer must fall through to normal admission there instead of being honored."""
+        called = {}
+
+        async def _record(api_key, request):
+            called["api_key"] = api_key
+            return UserAPIKeyAuth(user_id="fell-through")
+
+        with self._rig(), patch(self._AUTH_PATCH_TARGET, side_effect=_record):
+            auth, *_rest = await MCPRequestHandler.process_mcp_request(
+                self._scope(self._bearer(), path="/mcp/some_server")
+            )
+
+        assert auth.user_id == "fell-through"
+        assert auth.is_mcp_gateway_session is False
+
+    async def test_an_explicit_litellm_key_still_wins_over_a_session_bearer(self):
+        async def _key_auth(api_key, request):
+            return UserAPIKeyAuth(user_id="key-owner", api_key="sk-real")
+
+        with self._rig(), patch(self._AUTH_PATCH_TARGET, side_effect=_key_auth):
+            auth, *_rest = await MCPRequestHandler.process_mcp_request(
+                self._scope(self._bearer(), extra_headers=[(b"x-litellm-api-key", b"sk-real")])
+            )
+
+        assert auth.user_id == "key-owner"
+        assert auth.is_mcp_gateway_session is False
+
+    async def test_a_bearer_signed_under_a_different_master_key_is_rejected(self):
+        with self._rig():
+            with patch("litellm.proxy.proxy_server.master_key", "sk-a-totally-different-master-key"):
+                with pytest.raises(HTTPException) as exc_info:
+                    await MCPRequestHandler.process_mcp_request(self._scope(self._bearer()))
+
+        assert exc_info.value.status_code == 401
+
+    async def test_a_scim_deactivated_user_cannot_use_an_outstanding_session(self):
+        """The bearer is a reference, not an authorization: offboarding must land on the next call
+        rather than at token expiry."""
+        with self._rig(user_row=self._user_row(metadata={"scim_active": False})):
+            with pytest.raises(HTTPException) as exc_info:
+                await MCPRequestHandler.process_mcp_request(self._scope(self._bearer()))
+
+        assert exc_info.value.status_code == 401

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

@@ -9028,3 +9028,68 @@ class TestUrllessIssuerDiscovery:
         anchored.assert_awaited_once_with("https://idp.example.com", None)
         resource_rooted.assert_not_awaited()
         assert built.token_url == "https://idp.example.com/token"
+
+
+@pytest.mark.asyncio
+class TestGatewayDCRSessionDoesNotInheritAdminViewAll:
+    """A gateway DCR session is a keyless subject held by a public OAuth client.
+
+    ``get_allowed_mcp_servers`` short-circuits to the entire registry for a principal with an admin
+    role and no explicit object permission. That is the dashboard's view-all behavior, and letting a
+    gateway session inherit it would hand an admin's MCP client every server on the proxy, so the
+    connect grid the user authorized servers on would stop describing what their client can reach.
+    """
+
+    @staticmethod
+    def _admin(is_mcp_gateway_session: bool):
+        from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
+
+        return UserAPIKeyAuth(
+            user_id="admin-user",
+            user_role=LitellmUserRoles.PROXY_ADMIN,
+            is_mcp_gateway_session=is_mcp_gateway_session,
+        )
+
+    @staticmethod
+    def _registry():
+        from litellm.types.mcp import MCPAuth
+        from litellm.types.mcp_server.mcp_server_manager import MCPServer
+
+        return {
+            server_id: MCPServer(
+                server_id=server_id,
+                name=server_id,
+                server_name=server_id,
+                transport="http",
+                auth_type=MCPAuth.none,
+                allow_all_keys=False,
+            )
+            for server_id in ("granted-server", "other-server")
+        }
+
+    async def _allowed(self, manager, auth):
+        from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import MCPRequestHandler
+
+        with patch.object(
+            MCPRequestHandler, "get_allowed_mcp_servers", new=AsyncMock(return_value=["granted-server"])
+        ):
+            return await manager.get_allowed_mcp_servers(user_api_key_auth=auth)
+
+    async def test_an_admin_dashboard_principal_still_sees_every_server(self):
+        """The existing view-all behavior must be untouched for everything that is not a session."""
+        manager = MCPServerManager()
+        manager.get_registry = MagicMock(return_value=self._registry())
+
+        assert sorted(await self._allowed(manager, self._admin(is_mcp_gateway_session=False))) == [
+            "granted-server",
+            "other-server",
+        ]
+
+    async def test_an_admin_connecting_through_the_gateway_gets_only_their_grants(self):
+        manager = MCPServerManager()
+        manager.get_registry = MagicMock(return_value=self._registry())
+
+        allowed = await self._allowed(manager, self._admin(is_mcp_gateway_session=True))
+
+        assert allowed == ["granted-server"]
+        assert "other-server" not in allowed