refactor(MCP): Replace MCPRouter with FastMCP Proxy (#8877)

Co-authored-by: openhands <[email protected]>

Author: xingyaoww

.github/dependabot.yml

@@ -16,7 +16,6 @@ updates:
       mcp-packages:
         patterns:
           - "mcp"
-          - "mcpm"
       security-all:
         applies-to: "security-updates"
         patterns:

openhands/runtime/action_execution_server.py

@@ -9,7 +9,6 @@
 import asyncio
 import base64
 import json
-import logging
 import mimetypes
 import os
 import shutil
@@ -26,8 +25,6 @@
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import FileResponse, JSONResponse
 from fastapi.security import APIKeyHeader
-from mcpm import MCPRouter, RouterConfig
-from mcpm.router.router import logger as mcp_router_logger
 from openhands_aci.editor.editor import OHEditor
 from openhands_aci.editor.exceptions import ToolError
 from openhands_aci.editor.results import ToolResult
@@ -37,6 +34,7 @@
 from starlette.exceptions import HTTPException as StarletteHTTPException
 from uvicorn import run
 
+from openhands.core.config.mcp_config import MCPStdioServerConfig
 from openhands.core.exceptions import BrowserUnavailableException
 from openhands.core.logger import openhands_logger as logger
 from openhands.events.action import (
@@ -63,20 +61,18 @@
 from openhands.runtime.browser import browse
 from openhands.runtime.browser.browser_env import BrowserEnv
 from openhands.runtime.file_viewer_server import start_file_viewer_server
+
+# Import our custom MCP Proxy Manager
+from openhands.runtime.mcp.proxy import MCPProxyManager
 from openhands.runtime.plugins import ALL_PLUGINS, JupyterPlugin, Plugin, VSCodePlugin
 from openhands.runtime.utils import find_available_tcp_port
 from openhands.runtime.utils.bash import BashSession
 from openhands.runtime.utils.files import insert_lines, read_lines
-from openhands.runtime.utils.log_capture import capture_logs
 from openhands.runtime.utils.memory_monitor import MemoryMonitor
 from openhands.runtime.utils.runtime_init import init_user_and_working_directory
 from openhands.runtime.utils.system_stats import get_system_stats
 from openhands.utils.async_utils import call_sync_from_async, wait_all
 
-# Set MCP router logger to the same level as the main logger
-mcp_router_logger.setLevel(logger.getEffectiveLevel())
-
-
 if sys.platform == 'win32':
     from openhands.runtime.utils.windows_bash import WindowsPowershellSession
 
@@ -471,7 +467,7 @@ async def read(self, action: FileReadAction) -> Observation:
         filepath = self._resolve_path(action.path, working_dir)
         try:
             if filepath.lower().endswith(('.png', '.jpg', '.jpeg', '.bmp', '.gif')):
-                with open(filepath, 'rb') as file:  # noqa: ASYNC101
+                with open(filepath, 'rb') as file:
                     image_data = file.read()
                     encoded_image = base64.b64encode(image_data).decode('utf-8')
                     mime_type, _ = mimetypes.guess_type(filepath)
@@ -481,13 +477,13 @@ async def read(self, action: FileReadAction) -> Observation:
 
                 return FileReadObservation(path=filepath, content=encoded_image)
             elif filepath.lower().endswith('.pdf'):
-                with open(filepath, 'rb') as file:  # noqa: ASYNC101
+                with open(filepath, 'rb') as file:
                     pdf_data = file.read()
                     encoded_pdf = base64.b64encode(pdf_data).decode('utf-8')
                     encoded_pdf = f'data:application/pdf;base64,{encoded_pdf}'
                 return FileReadObservation(path=filepath, content=encoded_pdf)
             elif filepath.lower().endswith(('.mp4', '.webm', '.ogg')):
-                with open(filepath, 'rb') as file:  # noqa: ASYNC101
+                with open(filepath, 'rb') as file:
                     video_data = file.read()
                     encoded_video = base64.b64encode(video_data).decode('utf-8')
                     mime_type, _ = mimetypes.guess_type(filepath)
@@ -497,7 +493,7 @@ async def read(self, action: FileReadAction) -> Observation:
 
                 return FileReadObservation(path=filepath, content=encoded_video)
 
-            with open(filepath, 'r', encoding='utf-8') as file:  # noqa: ASYNC101
+            with open(filepath, 'r', encoding='utf-8') as file:
                 lines = read_lines(file.readlines(), action.start, action.end)
         except FileNotFoundError:
             return ErrorObservation(
@@ -530,7 +526,7 @@ async def write(self, action: FileWriteAction) -> Observation:
 
         mode = 'w' if not file_exists else 'r+'
         try:
-            with open(filepath, mode, encoding='utf-8') as file:  # noqa: ASYNC101
+            with open(filepath, mode, encoding='utf-8') as file:
                 if mode != 'w':
                     all_lines = file.readlines()
                     new_file = insert_lines(insert, all_lines, action.start, action.end)
@@ -654,14 +650,11 @@ def close(self):
             plugins_to_load.append(ALL_PLUGINS[plugin]())  # type: ignore
 
     client: ActionExecutor | None = None
-    mcp_router: MCPRouter | None = None
-    MCP_ROUTER_PROFILE_PATH = os.path.join(
-        os.path.dirname(__file__), 'mcp', 'config.json'
-    )
+    mcp_proxy_manager: MCPProxyManager | None = None
 
     @asynccontextmanager
     async def lifespan(app: FastAPI):
-        global client, mcp_router
+        global client, mcp_proxy_manager
         logger.info('Initializing ActionExecutor...')
         client = ActionExecutor(
             plugins_to_load,
@@ -676,63 +669,36 @@ async def lifespan(app: FastAPI):
         # Check if we're on Windows
         is_windows = sys.platform == 'win32'
 
-        # Initialize and mount MCP Router (skip on Windows)
+        # Initialize and mount MCP Proxy Manager (skip on Windows)
         if is_windows:
-            logger.info('Skipping MCP Router initialization on Windows')
-            mcp_router = None
+            logger.info('Skipping MCP Proxy initialization on Windows')
+            mcp_proxy_manager = None
         else:
-            logger.info('Initializing MCP Router...')
-            mcp_router = MCPRouter(
-                profile_path=MCP_ROUTER_PROFILE_PATH,
-                router_config=RouterConfig(
-                    api_key=SESSION_API_KEY,
-                    auth_enabled=bool(SESSION_API_KEY),
-                ),
+            logger.info('Initializing MCP Proxy Manager...')
+            # Create a MCP Proxy Manager
+            mcp_proxy_manager = MCPProxyManager(
+                auth_enabled=bool(SESSION_API_KEY),
+                api_key=SESSION_API_KEY,
+                logger_level=logger.getEffectiveLevel(),
             )
+            mcp_proxy_manager.initialize()
+            # Mount the proxy to the app
             allowed_origins = ['*']
-            sse_app = await mcp_router.get_sse_server_app(
-                allow_origins=allowed_origins, include_lifespan=False
-            )
-
-        # Only mount SSE app if MCP Router is initialized (not on Windows)
-        if mcp_router is not None:
-            # Check for route conflicts before mounting
-            main_app_routes = {route.path for route in app.routes}
-            sse_app_routes = {route.path for route in sse_app.routes}
-            conflicting_routes = main_app_routes.intersection(sse_app_routes)
-
-            if conflicting_routes:
-                logger.error(f'Route conflicts detected: {conflicting_routes}')
-                raise RuntimeError(
-                    f'Cannot mount SSE app - conflicting routes found: {conflicting_routes}'
-                )
-
-            app.mount('/', sse_app)
-            logger.info(
-                f'Mounted MCP Router SSE app at root path with allowed origins: {allowed_origins}'
-            )
-
-            # Additional debug logging
-            if logger.isEnabledFor(logging.DEBUG):
-                logger.debug('Main app routes:')
-                for route in main_app_routes:
-                    logger.debug(f'  {route}')
-                logger.debug('MCP SSE server app routes:')
-                for route in sse_app_routes:
-                    logger.debug(f'  {route}')
+            try:
+                await mcp_proxy_manager.mount_to_app(app, allowed_origins)
+            except Exception as e:
+                logger.error(f'Error mounting MCP Proxy: {e}', exc_info=True)
+                raise RuntimeError(f'Cannot mount MCP Proxy: {e}')
 
         yield
 
         # Clean up & release the resources
-        logger.info('Shutting down MCP Router...')
-        if mcp_router:
-            try:
-                await mcp_router.shutdown()
-                logger.info('MCP Router shutdown successfully.')
-            except Exception as e:
-                logger.error(f'Error shutting down MCP Router: {e}', exc_info=True)
+        logger.info('Shutting down MCP Proxy Manager...')
+        if mcp_proxy_manager:
+            del mcp_proxy_manager
+            mcp_proxy_manager = None
         else:
-            logger.info('MCP Router instance not found for shutdown.')
+            logger.info('MCP Proxy Manager instance not found for shutdown.')
 
         logger.info('Closing ActionExecutor...')
         if client:
@@ -824,6 +790,9 @@ async def update_mcp_server(request: Request):
         # Check if we're on Windows
         is_windows = sys.platform == 'win32'
 
+        # Access the global mcp_proxy_manager variable
+        global mcp_proxy_manager
+
         if is_windows:
             # On Windows, just return a success response without doing anything
             logger.info(
@@ -838,49 +807,28 @@ async def update_mcp_server(request: Request):
             )
 
         # Non-Windows implementation
-        assert mcp_router is not None
-        assert os.path.exists(MCP_ROUTER_PROFILE_PATH)
-
-        # Use synchronous file operations outside of async function
-        def read_profile():
-            with open(MCP_ROUTER_PROFILE_PATH, 'r') as f:
-                return json.load(f)
-
-        current_profile = read_profile()
-        assert 'default' in current_profile
-        assert isinstance(current_profile['default'], list)
+        if mcp_proxy_manager is None:
+            raise HTTPException(
+                status_code=500, detail='MCP Proxy Manager is not initialized'
+            )
 
         # Get the request body
         mcp_tools_to_sync = await request.json()
         if not isinstance(mcp_tools_to_sync, list):
             raise HTTPException(
                 status_code=400, detail='Request must be a list of MCP tools to sync'
             )
-
         logger.info(
-            f'Updating MCP server to: {json.dumps(mcp_tools_to_sync, indent=2)}.\nPrevious profile: {json.dumps(current_profile, indent=2)}'
+            f'Updating MCP server with tools: {json.dumps(mcp_tools_to_sync, indent=2)}'
         )
-        current_profile['default'] = mcp_tools_to_sync
-
-        # Use synchronous file operations outside of async function
-        def write_profile(profile):
-            with open(MCP_ROUTER_PROFILE_PATH, 'w') as f:
-                json.dump(profile, f)
-
-        write_profile(current_profile)
-
-        # Manually reload the profile and update the servers
-        mcp_router.profile_manager.reload()
-        servers_wait_for_update = mcp_router.get_unique_servers()
-        async with capture_logs('mcpm.router.router') as log_capture:
-            await mcp_router.update_servers(servers_wait_for_update)
-        router_error_log = log_capture.getvalue()
-
-        logger.info(
-            f'MCP router updated successfully with unique servers: {servers_wait_for_update}'
-        )
-        if router_error_log:
-            logger.warning(f'Some MCP servers failed to be added: {router_error_log}')
+        mcp_tools_to_sync = [MCPStdioServerConfig(**tool) for tool in mcp_tools_to_sync]
+        try:
+            await mcp_proxy_manager.update_and_remount(app, mcp_tools_to_sync, ['*'])
+            logger.info('MCP Proxy Manager updated and remounted successfully')
+            router_error_log = ''
+        except Exception as e:
+            logger.error(f'Error updating MCP Proxy Manager: {e}', exc_info=True)
+            router_error_log = str(e)
 
         return JSONResponse(
             status_code=200,
@@ -915,7 +863,7 @@ async def upload_file(
                     )
 
                 zip_path = os.path.join(full_dest_path, file.filename)
-                with open(zip_path, 'wb') as buffer:  # noqa: ASYNC101
+                with open(zip_path, 'wb') as buffer:
                     shutil.copyfileobj(file.file, buffer)
 
                 # Extract the zip file
@@ -928,7 +876,7 @@ async def upload_file(
             else:
                 # For single file uploads
                 file_path = os.path.join(full_dest_path, file.filename)
-                with open(file_path, 'wb') as buffer:  # noqa: ASYNC101
+                with open(file_path, 'wb') as buffer:
                     shutil.copyfileobj(file.file, buffer)
                 logger.debug(f'Uploaded file {file.filename} to {destination}')
 

openhands/runtime/impl/action_execution/action_execution_client.py

@@ -435,7 +435,7 @@ def get_mcp_config(
             # We should always include the runtime as an MCP server whenever there's > 0 stdio servers
             updated_mcp_config.sse_servers.append(
                 MCPSSEServerConfig(
-                    url=self.action_execution_server_url.rstrip('/') + '/sse',
+                    url=self.action_execution_server_url.rstrip('/') + '/mcp/sse',
                     api_key=self.session_api_key,
                 )
             )

openhands/runtime/mcp/config.json

@@ -1,3 +1,6 @@
 {
-    "default": []
+    "mcpServers": {
+        "default": {}
+    },
+    "tools": []
 }

openhands/runtime/mcp/proxy/README.md

@@ -0,0 +1,71 @@
+# MCP Proxy Manager
+
+This module provides a manager class for handling FastMCP proxy instances in OpenHands, including initialization, configuration, and mounting to FastAPI applications.
+
+## Overview
+
+The `MCPProxyManager` class encapsulates all the functionality related to creating, configuring, and managing FastMCP proxy instances. It simplifies the process of:
+
+1. Initializing a FastMCP proxy
+2. Configuring the proxy with tools
+3. Mounting the proxy to a FastAPI application
+4. Updating the proxy configuration
+5. Shutting down the proxy
+
+## Usage
+
+### Basic Usage
+
+```python
+from openhands.runtime.mcp.proxy import MCPProxyManager
+from fastapi import FastAPI
+
+# Create a FastAPI app
+app = FastAPI()
+
+# Create a proxy manager
+proxy_manager = MCPProxyManager(
+    name="MyProxyServer",
+    auth_enabled=True,
+    api_key="my-api-key"
+)
+
+# Initialize the proxy
+proxy_manager.initialize()
+
+# Mount the proxy to the app
+await proxy_manager.mount_to_app(app, allow_origins=["*"])
+
+# Update the tools configuration
+tools = [
+    {
+        "name": "my_tool",
+        "description": "My tool description",
+        "parameters": {...}
+    }
+]
+proxy_manager.update_tools(tools)
+
+# Update and remount the proxy
+await proxy_manager.update_and_remount(app, tools, allow_origins=["*"])
+
+# Shutdown the proxy
+await proxy_manager.shutdown()
+```
+
+### In-Memory Configuration
+
+The `MCPProxyManager` maintains the configuration in-memory, eliminating the need for file-based configuration. This makes it easier to update the configuration and reduces the complexity of the code.
+
+## Benefits
+
+1. **Simplified API**: The `MCPProxyManager` provides a simple and intuitive API for managing FastMCP proxies.
+2. **In-Memory Configuration**: Configuration is maintained in-memory, eliminating the need for file I/O operations.
+3. **Improved Error Handling**: The manager provides better error handling and logging for proxy operations.
+4. **Cleaner Code**: By encapsulating proxy-related functionality in a dedicated class, the code is more maintainable and easier to understand.
+
+## Implementation Details
+
+The `MCPProxyManager` uses the `FastMCP.as_proxy()` method to create a proxy server. It manages the lifecycle of the proxy, including initialization, configuration updates, and shutdown.
+
+When updating the tools configuration, the manager creates a new proxy with the updated configuration and remounts it to the FastAPI application, ensuring that the proxy is always up-to-date with the latest configuration.

openhands/runtime/mcp/proxy/__init__.py

@@ -0,0 +1,7 @@
+"""
+MCP Proxy module for OpenHands.
+"""
+
+from openhands.runtime.mcp.proxy.manager import MCPProxyManager
+
+__all__ = ['MCPProxyManager']