refactor(mcp): satisfy tightened immutability lint ceiling

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/_experimental/mcp_server/catalog.py

@@ -32,7 +32,7 @@ class CatalogSnapshot:
     servers: Mapping[str, MCPServer]
     identity: str
     tools: Mapping[str, MCPTool]
-    routing: dict[str, str]
+    routing: dict[str, str]  # mutable-ok: tool routes are remapped in place through this snapshot's live dict
 
 
 def _configuration_identity(server: MCPServer) -> str:
@@ -96,7 +96,7 @@ class TargetCatalog:
         scoped: Final = self._operation.get()
         return scoped[0] if scoped is not None and not scoped[1].is_set() else None
 
-    def routing(self) -> dict[str, str]:
+    def routing(self) -> dict[str, str]:  # mutable-ok: returns the live routing dict that callers remap in place
         staged: Final = self._staged_routing.get()
         if staged is not None and not staged[1].is_set():
             return staged[0]
@@ -361,7 +361,9 @@ class TargetCatalog:
             closed.set()
             self._staged_routing.reset(routing_token)
 
-    async def _stage_servers(self, rows: Sequence[BaseModel], *, reuse_unchanged: bool) -> dict[str, MCPServer]:
+    async def _stage_servers(
+        self, rows: Sequence[BaseModel], *, reuse_unchanged: bool
+    ) -> dict[str, MCPServer]:  # mutable-ok: assign_unique_short_prefix requires a dict registry
         from litellm.proxy._experimental.mcp_server.db import LiteLLM_MCPServerTable
         from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
             carry_forward_resolved_oauth_endpoints,

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -2415,7 +2415,9 @@ class MCPServerManager:
         return self.catalog.registry()
 
     @property
-    def tool_name_to_mcp_server_name_mapping(self) -> dict[str, str]:
+    def tool_name_to_mcp_server_name_mapping(
+        self,
+    ) -> dict[str, str]:  # mutable-ok: callers remap tool routes through this dict
         return self.catalog.routing()
 
     @tool_name_to_mcp_server_name_mapping.setter

litellm/proxy/_experimental/mcp_server/tool_registry.py

@@ -25,22 +25,24 @@ class MCPToolRegistry:
 
     def __init__(self):
         # Registry to store all registered tools
-        self.published_tools: dict[str, MCPTool] = {}
+        self.published_tools: dict[str, MCPTool] = {}  # mutable-ok: register_tool writes entries through .tools
         self._catalog_tools: ContextVar[tuple[dict[str, MCPTool], asyncio.Event] | None] = ContextVar(
             "mcp_catalog_tools", default=None
         )
 
     @property
-    def tools(self) -> dict[str, MCPTool]:
+    def tools(self) -> dict[str, MCPTool]:  # mutable-ok: register_tool mutates the returned mapping
         scoped: Final = self._catalog_tools.get()
         return scoped[0] if scoped is not None and not scoped[1].is_set() else self.published_tools
 
     @tools.setter
-    def tools(self, tools: dict[str, MCPTool]) -> None:
+    def tools(self, tools: dict[str, MCPTool]) -> None:  # mutable-ok: stored dict is mutated by register_tool
         self.published_tools = tools
 
     @contextmanager
-    def catalog_scope(self, tools: Mapping[str, MCPTool]) -> Generator[dict[str, MCPTool]]:
+    def catalog_scope(
+        self, tools: Mapping[str, MCPTool]
+    ) -> Generator[dict[str, MCPTool]]:  # mutable-ok: yields the mutable staged copy
         detached: Final = dict(tools)
         closed: Final = asyncio.Event()
         token: Final = self._catalog_tools.set((detached, closed))

litellm/proxy/guardrails/guardrail_hooks/mcp_security/mcp_security_guardrail.py

@@ -5,6 +5,7 @@ Validates that MCP servers referenced in request tools are registered
 on the LiteLLM gateway. Blocks or alerts when unregistered servers are found.
 """
 
+from collections.abc import Mapping
 from typing import Any, Final, Literal
 
 from fastapi import HTTPException
@@ -90,15 +91,15 @@ class MCPSecurityGuardrail(CustomGuardrail):
         return server_names
 
     @staticmethod
-    async def _find_unregistered_mcp_servers(data: dict) -> set[str]:
+    async def _find_unregistered_mcp_servers(data: Mapping[str, object]) -> frozenset[str]:
         """Check tools in data against the MCP server registry. Returns set of unregistered server names."""
         tools: Final = data.get("tools")
         if not tools or not isinstance(tools, list):
-            return set()
+            return frozenset()
 
         requested_servers: Final = MCPSecurityGuardrail._extract_mcp_server_names_from_tools(tools)
         if not requested_servers:
-            return set()
+            return frozenset()
 
         from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
             global_mcp_server_manager,
@@ -108,4 +109,4 @@ class MCPSecurityGuardrail(CustomGuardrail):
             registry: Final = global_mcp_server_manager.get_registry()
             registered_names: Final = set(registry.keys())
 
-            return requested_servers - registered_names
+            return frozenset(requested_servers - registered_names)

litellm/proxy/management_endpoints/mcp_management_endpoints.py

@@ -2366,7 +2366,7 @@ if MCP_AVAILABLE:
     ):
         async with _oauth_server_operation(server_id, user_api_key_dict, request=request) as mcp_server:
             request_data: Final = await _read_request_body(request=request)
-            data: Final[dict] = {**request_data}
+            data: Final[Mapping[str, object]] = {**request_data}
             client_redirect_uris: Final = client_supplied_redirect_uris(data.get("redirect_uris"))
 
             return await register_client_with_server(

litellm/proxy/proxy_server.py

@@ -20323,18 +20323,11 @@ async def _resolve_mcp_csv_tokens(csv_segment: str, client_ip: str | None) -> li
             global_mcp_server_manager,
         )
 
-        seen: Final[set] = set()
-        deduped: Final[list[str]] = []
-        for raw in csv_segment.split(","):
-            token = raw.strip()
-            if not token or token in seen:
-                continue
-            seen.add(token)
-            deduped.append(token)
-            if len(deduped) >= DEFAULT_MCP_NAMESPACE_CSV_MAX_TOKENS:
-                break
+        deduped: Final = tuple(
+            token for token in dict.fromkeys(raw.strip() for raw in csv_segment.split(",")) if token
+        )[:DEFAULT_MCP_NAMESPACE_CSV_MAX_TOKENS]
 
-        resolved: Final[list[str]] = []
+        resolved: Final[list[str]] = []  # mutable-ok: sequential await per token cannot live in a comprehension
         for token in deduped:
             if global_mcp_server_manager.get_mcp_server_by_name(token, client_ip=client_ip):
                 resolved.append(token)