feat(mcp): parallelize tool fetching from multiple MCP servers (#18627)

* feat(mcp): parallelize tool fetching from multiple MCP servers

Replace sequential tool fetching with asyncio.gather() to reduce
client timeouts when using multiple MCP servers.

Changes:
- mcp_server_manager.py: list_tools() now fetches tools in parallel
- server.py: _get_tools_from_mcp_servers() now fetches tools in parallel

Real-world impact (7 MCP servers example):
- Sequential: ~4.5+ seconds (exceeds typical 5-second client timeouts)
- Parallel: ~1.2 seconds (max of all servers)

Fixes #18626

* fix: copy oauth2_headers to avoid shared dict mutation in parallel tasks

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -644,14 +644,14 @@ class MCPServerManager:
         """
         allowed_mcp_servers = await self.get_allowed_mcp_servers(user_api_key_auth)
 
-        list_tools_result: List[MCPTool] = []
         verbose_logger.debug("SERVER MANAGER LISTING TOOLS")
 
-        for server_id in allowed_mcp_servers:
+        async def _fetch_server_tools(server_id: str) -> List[MCPTool]:
+            """Fetch tools from a single server with error handling."""
             server = self.get_mcp_server_by_id(server_id)
             if server is None:
                 verbose_logger.warning(f"MCP Server {server_id} not found")
-                continue
+                return []
 
             # Get server-specific auth header if available
             server_auth_header = None
@@ -669,15 +669,21 @@ class MCPServerManager:
                     server=server,
                     mcp_auth_header=server_auth_header,
                 )
-                list_tools_result.extend(tools)
-                verbose_logger.info(
-                    f"Successfully fetched {len(tools)} tools from server {server.name}"
-                )
+                return tools
             except Exception as e:
                 verbose_logger.warning(
                     f"Failed to list tools from server {server.name}: {str(e)}. Continuing with other servers."
                 )
-                # Continue with other servers instead of failing completely
+                return []
+
+        # Fetch tools from all servers in parallel
+        tasks = [_fetch_server_tools(server_id) for server_id in allowed_mcp_servers]
+        results = await asyncio.gather(*tasks)
+
+        # Flatten results into single list
+        list_tools_result: List[MCPTool] = [
+            tool for tools in results for tool in tools
+        ]
 
         verbose_logger.info(
             f"Successfully fetched {len(list_tools_result)} tools total from all servers"

litellm/proxy/_experimental/mcp_server/server.py

@@ -709,7 +709,8 @@ if MCP_AVAILABLE:
 
         extra_headers: Optional[Dict[str, str]] = None
         if server.auth_type == MCPAuth.oauth2:
-            extra_headers = oauth2_headers
+            # Copy to avoid mutating the original dict (important for parallel fetching)
+            extra_headers = oauth2_headers.copy() if oauth2_headers else None
 
         if server.extra_headers and raw_headers:
             if extra_headers is None:
@@ -755,11 +756,10 @@ if MCP_AVAILABLE:
         # Decide whether to add prefix based on number of allowed servers
         add_prefix = not (len(allowed_mcp_servers) == 1)
 
-        # Get tools from each allowed server
-        all_tools = []
-        for server in allowed_mcp_servers:
+        async def _fetch_and_filter_server_tools(server: MCPServer) -> List[MCPTool]:
+            """Fetch and filter tools from a single server with error handling."""
             if server is None:
-                continue
+                return []
 
             server_auth_header, extra_headers = _prepare_mcp_server_headers(
                 server=server,
@@ -786,16 +786,24 @@ if MCP_AVAILABLE:
                     user_api_key_auth=user_api_key_auth,
                 )
 
-                all_tools.extend(filtered_tools)
-
                 verbose_logger.debug(
                     f"Successfully fetched {len(tools)} tools from server {server.name}, {len(filtered_tools)} after filtering"
                 )
+                return filtered_tools
             except Exception as e:
                 verbose_logger.exception(
                     f"Error getting tools from server {server.name}: {str(e)}"
                 )
-                # Continue with other servers instead of failing completely
+                return []
+
+        # Fetch tools from all servers in parallel
+        tasks = [
+            _fetch_and_filter_server_tools(server) for server in allowed_mcp_servers
+        ]
+        results = await asyncio.gather(*tasks)
+
+        # Flatten results into single list
+        all_tools: List[MCPTool] = [tool for tools in results for tool in tools]
 
         verbose_logger.info(
             f"Successfully fetched {len(all_tools)} tools total from all MCP servers"