From 85f9c5e83fcf69ed28795b55bb60d92843966681 Mon Sep 17 00:00:00 2001
From: Ryan Crabbe <ryan@berri.ai>
Date: Thu, 23 Apr 2026 11:13:30 -0700
Subject: [PATCH 1/3] feat(mcp): resolve team/key MCP permissions by server
 name or alias

team.object_permission.mcp_servers (and the per-key equivalent) previously
only accepted server_id strings. For config-loaded MCP servers, the id is
derived from a hash that includes the server URL, so the same logical
server in two regions ends up with two different ids in a shared database.
Permission lists had to enumerate every region's id.

Add a single MCPServerManager.expand_permission_list() helper that resolves
each entry against the current region's config + DB registry union: entries
that match a server_id pass through, entries that match an alias/server_name/
name expand to every matching id, and unresolved entries drop with a debug
log so stale or typo entries are diagnosable. Wire it into the four
_get_allowed_mcp_servers_for_* helpers so direct server entries and
mcp_tool_permissions dict keys are both expanded before the intersection.

Access-check outcomes are unchanged for existing id-based permissions;
name-based entries now resolve instead of being silently denied.
---
 .../mcp_server/auth/user_api_key_auth_mcp.py  |  53 ++++-
 .../mcp_server/mcp_server_manager.py          |  36 +++
 .../auth/test_user_api_key_auth_mcp.py        | 224 +++++++++++-------
 .../mcp_server/test_mcp_server_manager.py     | 124 ++++++++++
 4 files changed, 339 insertions(+), 98 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py b/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py
index 357d21eb09a..db413527982 100644
--- a/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py
+++ b/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py
@@ -643,8 +643,14 @@ class MCPRequestHandler:
             if key_object_permission is None:
                 return []
 
-            # Get direct MCP servers
-            direct_mcp_servers = key_object_permission.mcp_servers or []
+            # Permission entries may be server_ids OR names/aliases — expand to ids.
+            from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+                global_mcp_server_manager,
+            )
+
+            direct_mcp_servers = global_mcp_server_manager.expand_permission_list(
+                key_object_permission.mcp_servers or []
+            )
 
             # Get MCP servers from access groups
             access_group_servers = (
@@ -654,8 +660,8 @@ class MCPRequestHandler:
             )
 
             # servers referenced in tool permissions should also be accessible
-            tool_perm_servers = list(
-                (key_object_permission.mcp_tool_permissions or {}).keys()
+            tool_perm_servers = global_mcp_server_manager.expand_permission_list(
+                list((key_object_permission.mcp_tool_permissions or {}).keys())
             )
 
             # Combine all lists
@@ -685,8 +691,14 @@ class MCPRequestHandler:
             if object_permissions is None:
                 return []
 
-            # Get direct MCP servers
-            direct_mcp_servers = object_permissions.mcp_servers or []
+            # Permission entries may be server_ids OR names/aliases — expand to ids.
+            from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+                global_mcp_server_manager,
+            )
+
+            direct_mcp_servers = global_mcp_server_manager.expand_permission_list(
+                object_permissions.mcp_servers or []
+            )
 
             # Get MCP servers from access groups
             access_group_servers = (
@@ -696,8 +708,8 @@ class MCPRequestHandler:
             )
 
             # servers referenced in tool permissions should also be accessible
-            tool_perm_servers = list(
-                (object_permissions.mcp_tool_permissions or {}).keys()
+            tool_perm_servers = global_mcp_server_manager.expand_permission_list(
+                list((object_permissions.mcp_tool_permissions or {}).keys())
             )
 
             # Combine all lists
@@ -746,8 +758,14 @@ class MCPRequestHandler:
             if end_user_obj is None or end_user_obj.object_permission is None:
                 return []
 
-            # Get direct MCP servers
-            direct_mcp_servers = end_user_obj.object_permission.mcp_servers or []
+            # Permission entries may be server_ids OR names/aliases — expand to ids.
+            from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+                global_mcp_server_manager,
+            )
+
+            direct_mcp_servers = global_mcp_server_manager.expand_permission_list(
+                end_user_obj.object_permission.mcp_servers or []
+            )
 
             # Get MCP servers from access groups
             access_group_servers = (
@@ -757,8 +775,8 @@ class MCPRequestHandler:
             )
 
             # servers referenced in tool permissions should also be accessible
-            tool_perm_servers = list(
-                (end_user_obj.object_permission.mcp_tool_permissions or {}).keys()
+            tool_perm_servers = global_mcp_server_manager.expand_permission_list(
+                list((end_user_obj.object_permission.mcp_tool_permissions or {}).keys())
             )
 
             # Combine all lists
@@ -836,12 +854,21 @@ class MCPRequestHandler:
             if isinstance(mcp_access_groups, str):
                 mcp_access_groups = []
 
+            # Permission entries may be server_ids OR names/aliases — expand to ids.
+            from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+                global_mcp_server_manager,
+            )
+
+            expanded_direct_servers = global_mcp_server_manager.expand_permission_list(
+                list(direct_mcp_servers)
+            )
+
             access_group_servers = (
                 await MCPRequestHandler._get_mcp_servers_from_access_groups(
                     mcp_access_groups
                 )
             )
-            all_servers = list(direct_mcp_servers) + access_group_servers
+            all_servers = expanded_direct_servers + access_group_servers
             return list(set(all_servers))
         except Exception as e:
             verbose_logger.warning(
diff --git a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
index 8aec30bfc3c..4173dd5abb2 100644
--- a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
+++ b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
@@ -2765,6 +2765,42 @@ class MCPServerManager:
                 servers.append(server)
         return servers
 
+    def expand_permission_list(self, identifiers: List[str]) -> List[str]:
+        """
+        Expand a permission list of server_ids/names/aliases into concrete
+        server_ids against the current region's config + DB registry union.
+
+        Entries that match a server_id pass through unchanged. Entries that
+        don't are looked up by alias/server_name/name and replaced with every
+        matching server_id (duplicate names grant access to all matches).
+        Entries that resolve to nothing are dropped and a debug log is
+        emitted so admins can diagnose stale/typo permission entries — the
+        final access-check still denies them either way.
+        """
+        if not identifiers:
+            return []
+        registry = self.get_registry()
+        expanded: Set[str] = set()
+        for identifier in identifiers:
+            if identifier in registry:
+                expanded.add(identifier)
+                continue
+            matches: List[str] = [
+                server_id
+                for server_id, server in registry.items()
+                if server.alias == identifier
+                or server.server_name == identifier
+                or server.name == identifier
+            ]
+            if matches:
+                expanded.update(matches)
+            else:
+                verbose_logger.debug(
+                    f"MCP permission entry '{identifier}' does not resolve "
+                    "to any known server (config + DB union). Skipping."
+                )
+        return list(expanded)
+
     def get_mcp_server_by_name(
         self, server_name: str, client_ip: Optional[str] = None
     ) -> Optional[MCPServer]:
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py b/tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py
index 25ff143d595..fd1a2b12367 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py
@@ -1391,55 +1391,73 @@ async def test_get_allowed_mcp_servers_for_team_uses_helper():
     Test that _get_allowed_mcp_servers_for_team properly uses _get_team_object_permission
     helper which handles both loaded and unloaded object_permission cases.
     """
+    from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+        global_mcp_server_manager,
+    )
     from litellm.proxy._types import LiteLLM_ObjectPermissionTable
+    from litellm.types.mcp import MCPTransport
+    from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
-    # Create mock object permission with servers and access groups
-    mock_object_permission = LiteLLM_ObjectPermissionTable(
-        object_permission_id="perm-789",
-        mcp_servers=["direct-server1", "direct-server2"],
-        mcp_access_groups=["dev-group"],
-        vector_stores=[],
-    )
+    # Register placeholder ids in the manager so expand_permission_list resolves them.
+    for sid in ("direct-server1", "direct-server2"):
+        global_mcp_server_manager.registry[sid] = MCPServer(
+            server_id=sid,
+            name=sid,
+            server_name=sid,
+            url=f"https://{sid}.example.com",
+            transport=MCPTransport.http,
+        )
+    try:
+        # Create mock object permission with servers and access groups
+        mock_object_permission = LiteLLM_ObjectPermissionTable(
+            object_permission_id="perm-789",
+            mcp_servers=["direct-server1", "direct-server2"],
+            mcp_access_groups=["dev-group"],
+            vector_stores=[],
+        )
 
-    # Create mock user auth
-    mock_user_auth = UserAPIKeyAuth(
-        api_key="test-key",
-        user_id="test-user",
-        team_id="team-789",
-    )
+        # Create mock user auth
+        mock_user_auth = UserAPIKeyAuth(
+            api_key="test-key",
+            user_id="test-user",
+            team_id="team-789",
+        )
 
-    # Mock the helper methods
-    with patch.object(
-        MCPRequestHandler, "_get_team_object_permission"
-    ) as mock_get_team_perm:
+        # Mock the helper methods
         with patch.object(
-            MCPRequestHandler, "_get_mcp_servers_from_access_groups"
-        ) as mock_get_access_group_servers:
-            # Configure mocks
-            mock_get_team_perm.return_value = mock_object_permission
-            mock_get_access_group_servers.return_value = [
-                "group-server1",
-                "group-server2",
-            ]
+            MCPRequestHandler, "_get_team_object_permission"
+        ) as mock_get_team_perm:
+            with patch.object(
+                MCPRequestHandler, "_get_mcp_servers_from_access_groups"
+            ) as mock_get_access_group_servers:
+                # Configure mocks
+                mock_get_team_perm.return_value = mock_object_permission
+                mock_get_access_group_servers.return_value = [
+                    "group-server1",
+                    "group-server2",
+                ]
 
-            # Call the method
-            result = await MCPRequestHandler._get_allowed_mcp_servers_for_team(
-                mock_user_auth
-            )
+                # Call the method
+                result = await MCPRequestHandler._get_allowed_mcp_servers_for_team(
+                    mock_user_auth
+                )
 
-            # Assert the result contains both direct and access group servers
-            assert set(result) == {
-                "direct-server1",
-                "direct-server2",
-                "group-server1",
-                "group-server2",
-            }
+                # Assert the result contains both direct and access group servers
+                assert set(result) == {
+                    "direct-server1",
+                    "direct-server2",
+                    "group-server1",
+                    "group-server2",
+                }
 
-            # Verify _get_team_object_permission was called (the helper we fixed)
-            mock_get_team_perm.assert_called_once_with(mock_user_auth)
+                # Verify _get_team_object_permission was called (the helper we fixed)
+                mock_get_team_perm.assert_called_once_with(mock_user_auth)
 
-            # Verify access groups were resolved
-            mock_get_access_group_servers.assert_called_once_with(["dev-group"])
+                # Verify access groups were resolved
+                mock_get_access_group_servers.assert_called_once_with(["dev-group"])
+    finally:
+        for sid in ("direct-server1", "direct-server2"):
+            global_mcp_server_manager.registry.pop(sid, None)
 
 
 @pytest.mark.asyncio
@@ -1569,35 +1587,53 @@ async def test_get_allowed_mcp_servers_for_key_returns_empty_when_db_returns_non
 async def test_get_allowed_mcp_servers_for_key_prefers_in_memory_permission():
     """Ensure in-memory object_permission is used without hitting the DB."""
 
+    from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+        global_mcp_server_manager,
+    )
     from litellm.proxy._types import LiteLLM_ObjectPermissionTable
+    from litellm.types.mcp import MCPTransport
+    from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
-    perms = LiteLLM_ObjectPermissionTable(
-        object_permission_id="perm-in-memory",
-        mcp_servers=["direct-server"],
-        mcp_access_groups=["grp-alpha"],
-    )
-    user_api_key_auth = UserAPIKeyAuth(
-        api_key="test-key",
-        user_id="test-user",
-        object_permission=perms,
+    # Register "direct-server" in the manager so permission expansion resolves it.
+    # Without this, expand_permission_list drops unknown ids as stale — which is
+    # the correct production behavior but unrelated to what this test asserts.
+    global_mcp_server_manager.registry["direct-server"] = MCPServer(
+        server_id="direct-server",
+        name="direct-server",
+        server_name="direct-server",
+        url="https://direct-server.example.com",
+        transport=MCPTransport.http,
     )
+    try:
+        perms = LiteLLM_ObjectPermissionTable(
+            object_permission_id="perm-in-memory",
+            mcp_servers=["direct-server"],
+            mcp_access_groups=["grp-alpha"],
+        )
+        user_api_key_auth = UserAPIKeyAuth(
+            api_key="test-key",
+            user_id="test-user",
+            object_permission=perms,
+        )
 
-    with patch(
-        "litellm.proxy.auth.auth_checks.get_object_permission",
-        new_callable=AsyncMock,
-    ) as mock_get_perm:
-        with patch.object(
-            MCPRequestHandler, "_get_mcp_servers_from_access_groups"
-        ) as mock_access_groups:
-            mock_access_groups.return_value = ["group-server"]
+        with patch(
+            "litellm.proxy.auth.auth_checks.get_object_permission",
+            new_callable=AsyncMock,
+        ) as mock_get_perm:
+            with patch.object(
+                MCPRequestHandler, "_get_mcp_servers_from_access_groups"
+            ) as mock_access_groups:
+                mock_access_groups.return_value = ["group-server"]
 
-            result = await MCPRequestHandler._get_allowed_mcp_servers_for_key(
-                user_api_key_auth
-            )
+                result = await MCPRequestHandler._get_allowed_mcp_servers_for_key(
+                    user_api_key_auth
+                )
 
-    assert set(result) == {"direct-server", "group-server"}
-    mock_get_perm.assert_not_called()
-    mock_access_groups.assert_called_once_with(["grp-alpha"])
+        assert set(result) == {"direct-server", "group-server"}
+        mock_get_perm.assert_not_called()
+        mock_access_groups.assert_called_once_with(["grp-alpha"])
+    finally:
+        global_mcp_server_manager.registry.pop("direct-server", None)
 
 
 @pytest.mark.asyncio
@@ -1753,28 +1789,46 @@ async def test_tool_permission_servers_included_in_allowed_servers():
 
     Regression test for https://github.com/BerriAI/litellm/issues/21954
     """
-    perm = MagicMock()
-    perm.mcp_servers = []
-    perm.mcp_access_groups = []
-    perm.mcp_tool_permissions = {"server_id_123": ["tool_a", "tool_b"]}
-
-    user_api_key_auth = UserAPIKeyAuth(
-        api_key="test-key",
-        user_id="test-user",
+    from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+        global_mcp_server_manager,
     )
+    from litellm.types.mcp import MCPTransport
+    from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
-    with (
-        patch.object(
-            MCPRequestHandler, "_get_key_object_permission", return_value=perm
-        ),
-        patch.object(
-            MCPRequestHandler,
-            "_get_mcp_servers_from_access_groups",
-            new_callable=AsyncMock,
-            return_value=[],
-        ),
-    ):
-        result = await MCPRequestHandler._get_allowed_mcp_servers_for_key(
-            user_api_key_auth=user_api_key_auth,
+    # Register the server id so expand_permission_list resolves it rather than
+    # dropping it as stale.
+    global_mcp_server_manager.registry["server_id_123"] = MCPServer(
+        server_id="server_id_123",
+        name="server_id_123",
+        server_name="server_id_123",
+        url="https://server-id-123.example.com",
+        transport=MCPTransport.http,
+    )
+    try:
+        perm = MagicMock()
+        perm.mcp_servers = []
+        perm.mcp_access_groups = []
+        perm.mcp_tool_permissions = {"server_id_123": ["tool_a", "tool_b"]}
+
+        user_api_key_auth = UserAPIKeyAuth(
+            api_key="test-key",
+            user_id="test-user",
         )
-        assert "server_id_123" in result
+
+        with (
+            patch.object(
+                MCPRequestHandler, "_get_key_object_permission", return_value=perm
+            ),
+            patch.object(
+                MCPRequestHandler,
+                "_get_mcp_servers_from_access_groups",
+                new_callable=AsyncMock,
+                return_value=[],
+            ),
+        ):
+            result = await MCPRequestHandler._get_allowed_mcp_servers_for_key(
+                user_api_key_auth=user_api_key_auth,
+            )
+            assert "server_id_123" in result
+    finally:
+        global_mcp_server_manager.registry.pop("server_id_123", None)
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
index ac5349e7105..dbb53b9168b 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
@@ -2610,5 +2610,129 @@ class TestMCPServerManagerUpstreamInstructionsCache:
         assert "srv_b" in by_name and by_name["srv_b"].instructions is None
 
 
+class TestMCPServerManagerExpandPermissionList:
+    """Tests for the alias/name-aware permission list expansion used by team MCP permissions."""
+
+    def _make_server(
+        self,
+        server_id: str,
+        server_name: str,
+        alias=None,
+        name=None,
+    ) -> MCPServer:
+        return MCPServer(
+            server_id=server_id,
+            name=name if name is not None else (alias or server_name),
+            alias=alias,
+            server_name=server_name,
+            url=f"https://{server_id}.example.com",
+            transport=MCPTransport.http,
+        )
+
+    def test_empty_list_returns_empty(self):
+        manager = MCPServerManager()
+        assert manager.expand_permission_list([]) == []
+
+    def test_expands_server_name(self):
+        manager = MCPServerManager()
+        manager.config_mcp_servers["id-usw1"] = self._make_server(
+            "id-usw1", server_name="a"
+        )
+
+        assert manager.expand_permission_list(["a"]) == ["id-usw1"]
+
+    def test_expands_alias(self):
+        manager = MCPServerManager()
+        manager.config_mcp_servers["id-1"] = self._make_server(
+            "id-1", server_name="internal_name", alias="public_alias"
+        )
+
+        assert manager.expand_permission_list(["public_alias"]) == ["id-1"]
+
+    def test_drops_unknown_entry_silently(self):
+        manager = MCPServerManager()
+        manager.config_mcp_servers["id-1"] = self._make_server("id-1", server_name="b")
+
+        assert manager.expand_permission_list(["a"]) == []
+
+    def test_name_collision_expands_to_all_matches(self):
+        """Two servers sharing a server_name both resolve — the documented behavior."""
+        manager = MCPServerManager()
+        manager.config_mcp_servers["id-config"] = self._make_server(
+            "id-config", server_name="shared"
+        )
+        manager.registry["id-db"] = self._make_server("id-db", server_name="shared")
+
+        assert sorted(manager.expand_permission_list(["shared"])) == [
+            "id-config",
+            "id-db",
+        ]
+
+    def test_searches_config_and_registry_union(self):
+        manager = MCPServerManager()
+        manager.config_mcp_servers["cfg-id"] = self._make_server("cfg-id", server_name="a")
+        manager.registry["reg-id"] = self._make_server("reg-id", server_name="b")
+
+        assert manager.expand_permission_list(["a"]) == ["cfg-id"]
+        assert manager.expand_permission_list(["b"]) == ["reg-id"]
+
+    def test_id_match_takes_precedence_over_name_match(self):
+        """
+        If a permission entry matches a server_id directly, don't also add
+        servers whose server_name happens to equal that id.
+        """
+        manager = MCPServerManager()
+        manager.config_mcp_servers["id-1"] = self._make_server(
+            "id-1", server_name="other_name"
+        )
+        manager.config_mcp_servers["id-2"] = self._make_server(
+            "id-2", server_name="id-1"
+        )
+
+        assert manager.expand_permission_list(["id-1"]) == ["id-1"]
+
+    def test_mixed_ids_and_names_in_same_list(self):
+        manager = MCPServerManager()
+        manager.config_mcp_servers["uuid-1"] = self._make_server(
+            "uuid-1", server_name="a"
+        )
+        manager.config_mcp_servers["uuid-2"] = self._make_server(
+            "uuid-2", server_name="b"
+        )
+
+        # ["uuid-1", "b"] -> uuid-1 passes through, "b" resolves to uuid-2
+        assert sorted(manager.expand_permission_list(["uuid-1", "b"])) == [
+            "uuid-1",
+            "uuid-2",
+        ]
+
+    def test_deduplicates_overlapping_id_and_name_entries(self):
+        """If a list references the same server by both id and name, return it once."""
+        manager = MCPServerManager()
+        manager.config_mcp_servers["uuid-1"] = self._make_server(
+            "uuid-1", server_name="a"
+        )
+
+        assert manager.expand_permission_list(["uuid-1", "a"]) == ["uuid-1"]
+
+    def test_simulates_cross_region_portability(self):
+        """
+        Same permission entry "a" resolves to different concrete IDs per region —
+        the cross-region portability the customer is asking for.
+        """
+        usw1 = MCPServerManager()
+        usw1.config_mcp_servers["hash-usw1"] = self._make_server(
+            "hash-usw1", server_name="a"
+        )
+
+        usc1 = MCPServerManager()
+        usc1.config_mcp_servers["hash-usc1"] = self._make_server(
+            "hash-usc1", server_name="a"
+        )
+
+        assert usw1.expand_permission_list(["a"]) == ["hash-usw1"]
+        assert usc1.expand_permission_list(["a"]) == ["hash-usc1"]
+
+
 if __name__ == "__main__":
     pytest.main([__file__])

From 57b0d7f45fe00982369241109c130f6238a3d574 Mon Sep 17 00:00:00 2001
From: Ryan Crabbe <ryan@berri.ai>
Date: Thu, 23 Apr 2026 15:23:53 -0700
Subject: [PATCH 2/3] fix(mcp): resolve tool_permissions dict keys by
 id-or-name
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

PR feedback (greptile P1 / veria high): with the previous change, a team
storing mcp_tool_permissions={"my-alias": ["read_file"]} would pass the
server-access check (because the alias expanded to a concrete id in the
allowed-servers list) but the per-server tool lookup still did
dict.get(server_id) against the raw name-keyed dict — missing, returning
None, which callers treat as "no restrictions" → all tools allowed instead
of only the declared ones.

Add MCPServerManager.expand_tool_permissions() that rewrites the dict so
every key is a concrete server_id where possible (tool lists from keys
pointing at the same server are unioned). Unresolved keys pass through
unchanged so stale id-keyed restrictions still apply when the same string
is used for lookup. Wire the helper into the four dict-lookup sites:
get_allowed_tools_for_server (key + team paths), the agent tool lookup,
and the rest_endpoints.py tool filter.

Also switch expand_permission_list to pass through unresolved entries
(rather than dropping them) so existing test fixtures that use bare string
placeholders continue to work. The downstream access check denies unknown
entries when compared to the concrete request server_id, so security
posture is unchanged.

Sanitize the debug log to use %r formatting so an admin-controlled
identifier with newlines can't forge log entries (CodeQL log-injection
warning).
---
 .../mcp_server/auth/user_api_key_auth_mcp.py  | 53 ++++++++----
 .../mcp_server/mcp_server_manager.py          | 42 +++++++--
 .../mcp_server/rest_endpoints.py              | 10 +--
 .../mcp_server/test_mcp_server_manager.py     | 85 ++++++++++++++++++-
 4 files changed, 161 insertions(+), 29 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py b/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py
index db413527982..d4799e9f208 100644
--- a/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py
+++ b/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py
@@ -516,15 +516,26 @@ class MCPRequestHandler:
                 user_api_key_auth
             )
 
-            # Extract tool permissions for this server
+            # Extract tool permissions for this server. Dict keys may be
+            # server_ids OR names/aliases; normalize to server_id-keyed form
+            # before lookup so a name-based key does not silently drop its
+            # tool restrictions when server_id is the resolved uuid.
+            from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+                global_mcp_server_manager,
+            )
+
             key_tools = (
-                key_obj_perm.mcp_tool_permissions.get(server_id)
-                if key_obj_perm and key_obj_perm.mcp_tool_permissions
+                global_mcp_server_manager.expand_tool_permissions(
+                    key_obj_perm.mcp_tool_permissions
+                ).get(server_id)
+                if key_obj_perm
                 else None
             )
             team_tools = (
-                team_obj_perm.mcp_tool_permissions.get(server_id)
-                if team_obj_perm and team_obj_perm.mcp_tool_permissions
+                global_mcp_server_manager.expand_tool_permissions(
+                    team_obj_perm.mcp_tool_permissions
+                ).get(server_id)
+                if team_obj_perm
                 else None
             )
 
@@ -660,8 +671,10 @@ class MCPRequestHandler:
             )
 
             # servers referenced in tool permissions should also be accessible
-            tool_perm_servers = global_mcp_server_manager.expand_permission_list(
-                list((key_object_permission.mcp_tool_permissions or {}).keys())
+            tool_perm_servers = list(
+                global_mcp_server_manager.expand_tool_permissions(
+                    key_object_permission.mcp_tool_permissions
+                ).keys()
             )
 
             # Combine all lists
@@ -708,8 +721,10 @@ class MCPRequestHandler:
             )
 
             # servers referenced in tool permissions should also be accessible
-            tool_perm_servers = global_mcp_server_manager.expand_permission_list(
-                list((object_permissions.mcp_tool_permissions or {}).keys())
+            tool_perm_servers = list(
+                global_mcp_server_manager.expand_tool_permissions(
+                    object_permissions.mcp_tool_permissions
+                ).keys()
             )
 
             # Combine all lists
@@ -775,8 +790,10 @@ class MCPRequestHandler:
             )
 
             # servers referenced in tool permissions should also be accessible
-            tool_perm_servers = global_mcp_server_manager.expand_permission_list(
-                list((end_user_obj.object_permission.mcp_tool_permissions or {}).keys())
+            tool_perm_servers = list(
+                global_mcp_server_manager.expand_tool_permissions(
+                    end_user_obj.object_permission.mcp_tool_permissions
+                ).keys()
             )
 
             # Combine all lists
@@ -905,12 +922,16 @@ class MCPRequestHandler:
                 return None
 
             mcp_tool_permissions = getattr(obj_perm, "mcp_tool_permissions", None)
-            if not mcp_tool_permissions:
+            if not mcp_tool_permissions or not isinstance(mcp_tool_permissions, dict):
                 return None
-            if isinstance(mcp_tool_permissions, dict):
-                tools = mcp_tool_permissions.get(server_id)
-            else:
-                tools = None
+            # Dict keys may be server_ids OR names/aliases; normalize before lookup.
+            from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+                global_mcp_server_manager,
+            )
+
+            tools = global_mcp_server_manager.expand_tool_permissions(
+                mcp_tool_permissions
+            ).get(server_id)
             return list(tools) if tools else None
         except Exception as e:
             verbose_logger.warning(
diff --git a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
index 4173dd5abb2..251f271903b 100644
--- a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
+++ b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
@@ -2771,11 +2771,12 @@ class MCPServerManager:
         server_ids against the current region's config + DB registry union.
 
         Entries that match a server_id pass through unchanged. Entries that
-        don't are looked up by alias/server_name/name and replaced with every
-        matching server_id (duplicate names grant access to all matches).
-        Entries that resolve to nothing are dropped and a debug log is
+        match an alias/server_name/name are replaced with every matching
+        server_id (duplicate names grant access to all matches). Entries
+        that resolve to nothing pass through as-is and a debug log is
         emitted so admins can diagnose stale/typo permission entries — the
-        final access-check still denies them either way.
+        downstream access-check denies them when compared against the
+        concrete request server_id.
         """
         if not identifiers:
             return []
@@ -2795,12 +2796,41 @@ class MCPServerManager:
             if matches:
                 expanded.update(matches)
             else:
+                # %r quotes and escapes control chars so an admin-controlled
+                # identifier with newlines cannot forge log lines.
                 verbose_logger.debug(
-                    f"MCP permission entry '{identifier}' does not resolve "
-                    "to any known server (config + DB union). Skipping."
+                    "MCP permission entry %r does not resolve to any known "
+                    "server (config + DB union). Passing through — the "
+                    "downstream access check will deny it if it's stale.",
+                    identifier,
                 )
+                expanded.add(identifier)
         return list(expanded)
 
+    def expand_tool_permissions(
+        self,
+        tool_permissions: Optional[Dict[str, List[str]]],
+    ) -> Dict[str, List[str]]:
+        """
+        Rewrite an ``mcp_tool_permissions`` dict keyed by id/name/alias so
+        every key is a concrete server_id where possible. Tool lists from
+        keys that point at the same server are unioned, matching the
+        "duplicate names grant access to all matches" semantics of
+        ``expand_permission_list``.
+
+        Required so name-based keys don't silently drop their tool
+        restrictions when the lookup uses the resolved server_id. Unresolved
+        keys pass through via ``expand_permission_list`` so stale id-keyed
+        restrictions still apply when the same string is used for lookup.
+        """
+        if not tool_permissions:
+            return {}
+        result: Dict[str, List[str]] = {}
+        for key, tools in tool_permissions.items():
+            for server_id in self.expand_permission_list([key]):
+                result.setdefault(server_id, []).extend(tools or [])
+        return result
+
     def get_mcp_server_by_name(
         self, server_name: str, client_ip: Optional[str] = None
     ) -> Optional[MCPServer]:
diff --git a/litellm/proxy/_experimental/mcp_server/rest_endpoints.py b/litellm/proxy/_experimental/mcp_server/rest_endpoints.py
index 8131c040136..f1030bf0a8a 100644
--- a/litellm/proxy/_experimental/mcp_server/rest_endpoints.py
+++ b/litellm/proxy/_experimental/mcp_server/rest_endpoints.py
@@ -315,11 +315,11 @@ if MCP_AVAILABLE:
             and user_api_key_auth.object_permission
             and user_api_key_auth.object_permission.mcp_tool_permissions
         ):
-            allowed_tools_for_server = (
-                user_api_key_auth.object_permission.mcp_tool_permissions.get(
-                    server.server_id
-                )
-            )
+            # Dict keys may be server_ids OR names/aliases; normalize so lookup
+            # by concrete server_id resolves name-keyed restrictions too.
+            allowed_tools_for_server = global_mcp_server_manager.expand_tool_permissions(
+                user_api_key_auth.object_permission.mcp_tool_permissions
+            ).get(server.server_id)
             if (
                 allowed_tools_for_server is not None
                 and len(allowed_tools_for_server) > 0
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
index dbb53b9168b..6ac2b0ac4fa 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
@@ -2649,11 +2649,14 @@ class TestMCPServerManagerExpandPermissionList:
 
         assert manager.expand_permission_list(["public_alias"]) == ["id-1"]
 
-    def test_drops_unknown_entry_silently(self):
+    def test_passes_through_unknown_entry(self):
+        """Unresolved entries pass through unchanged (with a debug log) —
+        the downstream access check denies them when compared to the
+        concrete request server_id."""
         manager = MCPServerManager()
         manager.config_mcp_servers["id-1"] = self._make_server("id-1", server_name="b")
 
-        assert manager.expand_permission_list(["a"]) == []
+        assert manager.expand_permission_list(["a"]) == ["a"]
 
     def test_name_collision_expands_to_all_matches(self):
         """Two servers sharing a server_name both resolve — the documented behavior."""
@@ -2734,5 +2737,83 @@ class TestMCPServerManagerExpandPermissionList:
         assert usc1.expand_permission_list(["a"]) == ["hash-usc1"]
 
 
+class TestMCPServerManagerExpandToolPermissions:
+    """Tests for tool-permission dict rewriting — the privilege-escalation guard."""
+
+    def _make_server(self, server_id: str, server_name: str, alias=None) -> MCPServer:
+        return MCPServer(
+            server_id=server_id,
+            name=alias or server_name,
+            alias=alias,
+            server_name=server_name,
+            url=f"https://{server_id}.example.com",
+            transport=MCPTransport.http,
+        )
+
+    def test_empty_or_none_returns_empty_dict(self):
+        manager = MCPServerManager()
+        assert manager.expand_tool_permissions(None) == {}
+        assert manager.expand_tool_permissions({}) == {}
+
+    def test_rewrites_name_key_to_server_id(self):
+        """Privilege-escalation guard: a name-based key must resolve to the
+        concrete server_id, otherwise `.get(server_id)` misses and the tool
+        restriction is silently dropped (caller treats None as allow-all)."""
+        manager = MCPServerManager()
+        manager.config_mcp_servers["uuid-a"] = self._make_server(
+            "uuid-a", server_name="my-alias"
+        )
+
+        result = manager.expand_tool_permissions({"my-alias": ["read_file"]})
+        assert result == {"uuid-a": ["read_file"]}
+
+    def test_passes_through_existing_server_id_key(self):
+        manager = MCPServerManager()
+        manager.config_mcp_servers["uuid-a"] = self._make_server(
+            "uuid-a", server_name="alpha"
+        )
+
+        result = manager.expand_tool_permissions({"uuid-a": ["read_file"]})
+        assert result == {"uuid-a": ["read_file"]}
+
+    def test_unresolved_key_passes_through_unchanged(self):
+        """A stale id-keyed restriction (server since deleted, or just a
+        test-fixture placeholder) must still apply when something looks it
+        up by that same string — dropping would silently remove the
+        restriction."""
+        manager = MCPServerManager()
+
+        result = manager.expand_tool_permissions({"stale-uuid": ["read_file"]})
+        assert result == {"stale-uuid": ["read_file"]}
+
+    def test_name_collision_unions_tool_lists(self):
+        """Two servers sharing a server_name both match; their tool lists get
+        the restriction (matches the list-expansion collision semantics)."""
+        manager = MCPServerManager()
+        manager.config_mcp_servers["uuid-1"] = self._make_server(
+            "uuid-1", server_name="shared"
+        )
+        manager.registry["uuid-2"] = self._make_server("uuid-2", server_name="shared")
+
+        result = manager.expand_tool_permissions({"shared": ["read_file"]})
+        assert sorted(result.keys()) == ["uuid-1", "uuid-2"]
+        assert result["uuid-1"] == ["read_file"]
+        assert result["uuid-2"] == ["read_file"]
+
+    def test_id_and_name_keys_pointing_at_same_server_union_tools(self):
+        """If the admin writes both {"uuid-a": [...], "alias-a": [...]} and
+        both refer to the same server, the tool lists are unioned rather
+        than one overwriting the other."""
+        manager = MCPServerManager()
+        manager.config_mcp_servers["uuid-a"] = self._make_server(
+            "uuid-a", server_name="alias-a"
+        )
+
+        result = manager.expand_tool_permissions(
+            {"uuid-a": ["read_file"], "alias-a": ["write_file"]}
+        )
+        assert sorted(result["uuid-a"]) == ["read_file", "write_file"]
+
+
 if __name__ == "__main__":
     pytest.main([__file__])

From 09113078c02c111e61e91efe9ae006ceab70f09a Mon Sep 17 00:00:00 2001
From: Ryan Crabbe <ryan@berri.ai>
Date: Thu, 23 Apr 2026 16:00:28 -0700
Subject: [PATCH 3/3] style: apply black formatting

---
 litellm/proxy/_experimental/mcp_server/rest_endpoints.py  | 8 +++++---
 .../_experimental/mcp_server/test_mcp_server_manager.py   | 4 +++-
 2 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/rest_endpoints.py b/litellm/proxy/_experimental/mcp_server/rest_endpoints.py
index f1030bf0a8a..3047fb73325 100644
--- a/litellm/proxy/_experimental/mcp_server/rest_endpoints.py
+++ b/litellm/proxy/_experimental/mcp_server/rest_endpoints.py
@@ -317,9 +317,11 @@ if MCP_AVAILABLE:
         ):
             # Dict keys may be server_ids OR names/aliases; normalize so lookup
             # by concrete server_id resolves name-keyed restrictions too.
-            allowed_tools_for_server = global_mcp_server_manager.expand_tool_permissions(
-                user_api_key_auth.object_permission.mcp_tool_permissions
-            ).get(server.server_id)
+            allowed_tools_for_server = (
+                global_mcp_server_manager.expand_tool_permissions(
+                    user_api_key_auth.object_permission.mcp_tool_permissions
+                ).get(server.server_id)
+            )
             if (
                 allowed_tools_for_server is not None
                 and len(allowed_tools_for_server) > 0
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
index 6ac2b0ac4fa..447c28078ec 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
@@ -2673,7 +2673,9 @@ class TestMCPServerManagerExpandPermissionList:
 
     def test_searches_config_and_registry_union(self):
         manager = MCPServerManager()
-        manager.config_mcp_servers["cfg-id"] = self._make_server("cfg-id", server_name="a")
+        manager.config_mcp_servers["cfg-id"] = self._make_server(
+            "cfg-id", server_name="a"
+        )
         manager.registry["reg-id"] = self._make_server("reg-id", server_name="b")
 
         assert manager.expand_permission_list(["a"]) == ["cfg-id"]