fix(mcp): annotate connected-app reachability on the gateway connect page

litellm/models/mcp_server.py

@@ -102,6 +102,7 @@ class LiteLLM_MCPServerTable(LiteLLMPydanticObjectBase):
     byok_description: List[str] = Field(default_factory=list)
     byok_api_key_help_url: Optional[str] = None
     has_user_credential: Optional[bool] = None
+    connected_app_reachable: bool | None = None
     source_url: Optional[str] = None
     timeout: Optional[float] = None
     max_concurrent_requests: Optional[int] = None

litellm/proxy/_experimental/mcp_server/rest_endpoints.py

@@ -21,6 +21,7 @@
     list_fault_http_status,
 )
 from litellm.proxy._experimental.mcp_server.ui_session_utils import (
+    acting_user_auth,
     build_effective_auth_contexts,
 )
 from litellm.proxy._experimental.mcp_server.utils import (
@@ -644,13 +645,19 @@ def _as_query_str(value: Any) -> str | None:
         """Coerce an Optional[str] Query param to str|None, dropping unresolved FastAPI defaults."""
         return value if isinstance(value, str) else None
 
-    async def _resolve_toolset_scope(
+    async def _resolve_acting_auth(
         toolset_name: str | None,
         user_api_key_dict: UserAPIKeyAuth,
     ) -> UserAPIKeyAuth:
-        """Resolve ``toolset_name`` to its scoped ``UserAPIKeyAuth``, or return unchanged."""
+        """The one credential this tools request acts as.
+
+        A toolset name narrows the caller's own credential to that toolset; otherwise a dashboard
+        session is swapped for its admitted subject. The two are mutually exclusive by construction,
+        which is why they share an owner: the admitted subject resolves per grant source and a team
+        source deliberately carries none of the caller's ``object_permission``, so a toolset
+        narrowing layered on top would evaporate on every team-granted server."""
         if not toolset_name:
-            return user_api_key_dict
+            return await acting_user_auth(user_api_key_dict)
 
         from litellm.proxy.utils import get_prisma_client_or_throw
 
@@ -708,15 +715,14 @@ async def list_tool_rest_api(
         try:
             mcp_server_name = _as_query_str(mcp_server_name)
             toolset_name = _as_query_str(toolset_name)
+            user_api_key_dict = await _resolve_acting_auth(toolset_name, user_api_key_dict)
 
             # The full catalog (allowlist filter skipped) is admin-only so the
             # REST endpoint can't be used to enumerate deliberately-disabled tools.
             apply_tool_filters = not (
                 include_disabled_tools and user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN
             )
 
-            user_api_key_dict = await _resolve_toolset_scope(toolset_name, user_api_key_dict)
-
             if server_id is None:
                 server_id = mcp_server_name
 
@@ -905,6 +911,7 @@ async def call_tool_rest_api(
         )
 
         try:
+            user_api_key_dict = await acting_user_auth(user_api_key_dict)
             data = await request.json()
 
             tool_name = data.get("name")

litellm/proxy/_experimental/mcp_server/ui_session_utils.py

@@ -1,9 +1,11 @@
-"""Helpers to resolve real team contexts for UI session tokens."""
+"""Helpers to resolve the identity a dashboard UI session token acts as."""
 
 from __future__ import annotations
 
 from typing import List
 
+from fastapi import HTTPException
+
 from litellm._logging import verbose_logger
 from litellm.constants import UI_SESSION_TOKEN_TEAM_ID
 from litellm.proxy._types import UserAPIKeyAuth
@@ -23,12 +25,19 @@ def clone_user_api_key_auth_with_team(
     return cloned_auth
 
 
+def is_ui_session_credential(user_api_key_auth: UserAPIKeyAuth) -> bool:
+    """Whether the caller is the dashboard's SSO-minted session token acting as its user,
+    the only credential shape allowed to widen a request to the owning user's identity."""
+
+    return user_api_key_auth.team_id == UI_SESSION_TOKEN_TEAM_ID and bool(user_api_key_auth.user_id)
+
+
 async def resolve_ui_session_team_ids(
     user_api_key_auth: UserAPIKeyAuth,
 ) -> List[str]:
     """Resolve the real team ids backing a UI session token."""
 
-    if user_api_key_auth.team_id != UI_SESSION_TOKEN_TEAM_ID or not user_api_key_auth.user_id:
+    if not is_ui_session_credential(user_api_key_auth):
         return []
 
     from litellm.proxy.auth.auth_checks import get_user_object
@@ -68,12 +77,63 @@ async def resolve_ui_session_team_ids(
     return resolved_team_ids
 
 
+async def admitted_user_context(user_api_key_auth: UserAPIKeyAuth) -> UserAPIKeyAuth | None:
+    """THE owner of "resolve this dashboard session's user identity": the same admitted-subject auth a
+    gateway OAuth session for this user resolves with, carrying the user row's own object permission,
+    on this request's tracing span. None for any other credential (a caller-passed key is never
+    widened) and on reload failure, which every caller reads as "no user-level identity available"."""
+
+    user_id = user_api_key_auth.user_id
+    if not is_ui_session_credential(user_api_key_auth) or user_id is None:
+        return None
+    from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import (
+        MCPRequestHandler,
+    )
+
+    try:
+        admitted = await MCPRequestHandler._reload_admitted_user(user_id)
+    except HTTPException as e:
+        verbose_logger.warning(f"MCP dashboard session: admitted-subject reload failed for {user_id}: {e.detail}")
+        return None
+    return admitted.model_copy(update={"parent_otel_span": user_api_key_auth.parent_otel_span})
+
+
+async def acting_user_auth(user_api_key_auth: UserAPIKeyAuth) -> UserAPIKeyAuth:
+    """The principal acting-as-user MCP routes resolve permissions with. A non-admin dashboard
+    session acts as the admitted subject, the same identity a gateway session resolves with, so
+    server reachability, per-source tool ceilings, rate limits, and billing bind identically on
+    both surfaces. An admin session keeps its operator view and any caller-passed credential is
+    returned unchanged, never widened.
+
+    Do not combine this with a narrowing that rewrites a single credential's ``object_permission``
+    (toolset scope): the admitted subject resolves per grant source and a team source deliberately
+    carries none of the caller's own grants, so the narrowing would silently evaporate on every
+    team-granted server. A request carrying such a scope keeps the caller's own credential."""
+
+    if not is_ui_session_credential(user_api_key_auth):
+        return user_api_key_auth
+    from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view
+
+    if _user_has_admin_view(user_api_key_auth):
+        return user_api_key_auth
+    admitted = await admitted_user_context(user_api_key_auth)
+    return admitted if admitted is not None else user_api_key_auth
+
+
 async def build_effective_auth_contexts(
     user_api_key_auth: UserAPIKeyAuth,
 ) -> List[UserAPIKeyAuth]:
-    """Return auth contexts that reflect the actual teams for UI session tokens."""
+    """Every auth context a management or listing surface must resolve a UI session token through:
+    one per real team backing the session, plus the session user's own admitted identity, so a grant
+    made directly to the user row is as visible to the dashboard as it is to a gateway session."""
 
     resolved_team_ids = await resolve_ui_session_team_ids(user_api_key_auth)
-    if resolved_team_ids:
-        return [clone_user_api_key_auth_with_team(user_api_key_auth, team_id) for team_id in resolved_team_ids]
-    return [user_api_key_auth]
+    team_contexts = (
+        [clone_user_api_key_auth_with_team(user_api_key_auth, team_id) for team_id in resolved_team_ids]
+        if resolved_team_ids
+        else [user_api_key_auth]
+    )
+    admitted_context = await admitted_user_context(user_api_key_auth)
+    if admitted_context is None:
+        return team_contexts
+    return [*team_contexts, admitted_context]

litellm/proxy/management_endpoints/mcp_management_endpoints.py

@@ -148,7 +148,9 @@ def validate_tool_name(name: str) -> _ToolNameValidationResult:  # type: ignore[
         global_mcp_server_manager,
     )
     from litellm.proxy._experimental.mcp_server.ui_session_utils import (
+        admitted_user_context,
         build_effective_auth_contexts,
+        is_ui_session_credential,
     )
     from litellm.proxy._types import (
         LiteLLM_MCPServerTable,
@@ -939,6 +941,16 @@ async def _resolve_accessible_mcp_servers(
                 aggregated.setdefault(server.server_id, server)
         return list(aggregated.values())
 
+    async def _connected_app_reachable_server_ids(user_api_key_dict: UserAPIKeyAuth) -> frozenset[str]:
+        """Server ids a connected app authorized by this dashboard user is served on the aggregate
+        MCP endpoint, resolved through the one owner of the admitted subject so the page and the
+        session cannot drift. Empty when that identity cannot be built, which is the true answer:
+        the same user cannot open a gateway session either."""
+        admitted = await admitted_user_context(user_api_key_dict)
+        if admitted is None:
+            return frozenset()
+        return frozenset(await global_mcp_server_manager.get_allowed_mcp_servers(admitted))
+
     @router.get(
         "/server",
         description="Returns the mcp server list with associated teams",
@@ -953,6 +965,12 @@ async def fetch_all_mcp_servers(
             "servers the team has access to plus globally available (allow_all_keys) servers. "
             "Used by the Create Key UI to show team-scoped MCP servers.",
         ),
+        connected_app_view: bool = Query(
+            False,
+            description="Annotate each returned server with connected_app_reachable: whether a "
+            "connected app authorized by the calling user (a gateway OAuth session) is served "
+            "this server on the aggregate MCP endpoint.",
+        ),
     ):
         """
         Get all of the configured mcp servers for the user in the db with their associated teams
@@ -1009,6 +1027,11 @@ async def fetch_all_mcp_servers(
             servers = await _resolve_accessible_mcp_servers(user_api_key_dict)
             redacted_mcp_servers = _redact_mcp_credentials_list(servers)
 
+        if connected_app_view is True and is_ui_session_credential(user_api_key_dict):
+            reachable_ids = await _connected_app_reachable_server_ids(user_api_key_dict)
+            for server in redacted_mcp_servers:
+                server.connected_app_reachable = server.server_id in reachable_ids
+
         # augment the mcp servers with public status
         if litellm.public_mcp_servers is not None:
             for server in redacted_mcp_servers:

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

@@ -769,8 +769,179 @@ async def fake_get_tools(
         assert captured["server"] is stub_server
         assert result["tools"] == ["tool-1"]
         assert result["error"] is None
+
+    async def test_non_admin_ui_session_resolves_as_admitted_subject(self, monkeypatch):
+        """LIT-4861: a non-admin dashboard session must act as the admitted subject on this
+        route, so server reachability AND tool ceilings bind to the user's grants exactly as
+        they do for a gateway session, never to the bare session key."""
+        from litellm.constants import UI_SESSION_TOKEN_TEAM_ID
+
+        session_auth = UserAPIKeyAuth(
+            team_id=UI_SESSION_TOKEN_TEAM_ID, user_id="grant-user", user_role="internal_user"
+        )
+        admitted_auth = UserAPIKeyAuth(user_id="grant-user", org_id="admitted-org")
+
+        async def fake_reload(user_id):
+            assert user_id == "grant-user"
+            return admitted_auth
+
+        monkeypatch.setattr(
+            "litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp.MCPRequestHandler._reload_admitted_user",
+            fake_reload,
+        )
+
+        seen_server_resolution_auths = []
+
+        async def fake_get_allowed_mcp_servers(user_api_key_auth=None, **kwargs):
+            seen_server_resolution_auths.append(user_api_key_auth)
+            return ["server-1"]
+
+        class StubServer:
+            alias = "server-1"
+            server_name = "server-1"
+            name = "stub"
+            allowed_tools = None
+            mcp_info = {"server_name": "stub"}
+            available_on_public_internet = True
+
+        stub_server = StubServer()
+        captured = {}
+
+        async def fake_get_tools(
+            server,
+            server_auth_header,
+            raw_headers=None,
+            user_api_key_auth=None,
+            extra_headers=None,
+            apply_tool_filters=True,
+        ):
+            captured["user_api_key_auth"] = user_api_key_auth
+            return ["tool-1"]
+
+        monkeypatch.setattr(
+            rest_endpoints.global_mcp_server_manager,
+            "get_allowed_mcp_servers",
+            fake_get_allowed_mcp_servers,
+            raising=False,
+        )
+        monkeypatch.setattr(
+            rest_endpoints.global_mcp_server_manager,
+            "get_mcp_server_by_id",
+            lambda server_id: stub_server if server_id == "server-1" else None,
+            raising=False,
+        )
+        monkeypatch.setattr(
+            rest_endpoints,
+            "_get_tools_for_single_server",
+            fake_get_tools,
+            raising=False,
+        )
+
+        request = _build_request(path="/mcp-rest/tools/list", method="GET")
+        result = await rest_endpoints.list_tool_rest_api(
+            request,
+            server_id="server-1",
+            user_api_key_dict=session_auth,
+        )
+
+        resolved = [*seen_server_resolution_auths, captured["user_api_key_auth"]]
+        assert seen_server_resolution_auths
+        assert all(a.org_id == "admitted-org" and a.team_id is None for a in resolved)
+        assert result["tools"] == ["tool-1"]
         assert result["message"] == "Successfully retrieved tools"
 
+    async def test_toolset_scoped_request_keeps_the_caller_credential(self, monkeypatch):
+        """LIT-4861: the admitted subject resolves per grant source and a team source deliberately
+        carries none of the caller's own object_permission, so a toolset narrowing layered on top
+        would evaporate on every team-granted server. A toolset-scoped request therefore stays on
+        the caller's own credential, exactly as it did before the acting-as-user swap."""
+        from litellm.constants import UI_SESSION_TOKEN_TEAM_ID
+        from litellm.proxy._types import LiteLLM_ObjectPermissionTable
+
+        session_auth = UserAPIKeyAuth(
+            team_id=UI_SESSION_TOKEN_TEAM_ID, user_id="grant-user", user_role="internal_user"
+        )
+        scoped_auth = UserAPIKeyAuth(
+            object_permission=LiteLLM_ObjectPermissionTable(
+                object_permission_id="toolset-scope",
+                mcp_servers=["toolset-server-1"],
+            )
+        )
+        reload_calls: list[str] = []
+        scope_inputs: list[UserAPIKeyAuth] = []
+
+        async def record_reload(user_id):
+            reload_calls.append(user_id)
+            return UserAPIKeyAuth(user_id=user_id)
+
+        class StubToolset:
+            toolset_id = "toolset-1"
+
+        class StubServer:
+            alias = "toolset-server-1"
+            server_name = "toolset-server-1"
+            name = "toolset-server-1"
+            allowed_tools = None
+            mcp_info = {"server_name": "toolset-server-1"}
+            available_on_public_internet = True
+
+        stub_server = StubServer()
+
+        async def fake_get_toolset_by_name_cached(prisma_client, toolset_name):
+            return StubToolset()
+
+        async def fake_apply_toolset_scope(user_api_key_auth, toolset_id):
+            scope_inputs.append(user_api_key_auth)
+            return scoped_auth
+
+        async def fake_get_allowed_mcp_servers(user_api_key_auth=None, **kwargs):
+            assert user_api_key_auth is scoped_auth
+            return ["toolset-server-1"]
+
+        async def fake_get_tools(server, server_auth_header, *args, **kwargs):
+            return ["toolset-tool-1"]
+
+        monkeypatch.setattr(
+            "litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp.MCPRequestHandler._reload_admitted_user",
+            record_reload,
+        )
+        monkeypatch.setattr(
+            "litellm.proxy.utils.get_prisma_client_or_throw",
+            lambda *args, **kwargs: MagicMock(),
+        )
+        monkeypatch.setattr(
+            rest_endpoints.global_mcp_server_manager,
+            "get_toolset_by_name_cached",
+            fake_get_toolset_by_name_cached,
+            raising=False,
+        )
+        monkeypatch.setattr(rest_endpoints, "_apply_toolset_scope", fake_apply_toolset_scope, raising=False)
+        monkeypatch.setattr(
+            rest_endpoints.global_mcp_server_manager,
+            "get_allowed_mcp_servers",
+            fake_get_allowed_mcp_servers,
+            raising=False,
+        )
+        monkeypatch.setattr(
+            rest_endpoints.global_mcp_server_manager,
+            "get_mcp_server_by_id",
+            lambda server_id: stub_server if server_id == "toolset-server-1" else None,
+            raising=False,
+        )
+        monkeypatch.setattr(rest_endpoints, "_get_tools_for_single_server", fake_get_tools, raising=False)
+
+        request = _build_request(path="/mcp-rest/tools/list", method="GET")
+        result = await rest_endpoints.list_tool_rest_api(
+            request,
+            server_id=None,
+            toolset_name="research_tools",
+            user_api_key_dict=session_auth,
+        )
+
+        assert result["tools"] == ["toolset-tool-1"]
+        assert scope_inputs == [session_auth]
+        assert reload_calls == []
+
     async def test_include_disabled_tools_is_admin_only(self, monkeypatch):
         """include_disabled_tools skips the allowlist filter only for PROXY_ADMIN;
         a non-admin passing it stays filtered so the REST endpoint can't be used