fulfill the router with profile (#33)

* router wip

* Refactor clients

* return warning for owner or repo name mismatch (#9)

* return warning

* fix

* Router working for uvx servers

npx server still has issues with env

* fix: issues when starting server

* Support SSE client and add resource templates support to router

* Add file watcher and update router to support dynamic server config updates

* Refactor client connection management and add server lifespan handling

* Refactor router to support profile-based server management and add profile-aware SSE transport

* Refactor router to use ProfileConfigManager and ServerConfig instead of ConnectionDetails

* Update router README with new server config schema and namespace conventions

---------

Co-authored-by: Chen Nie <[email protected]>
Co-authored-by: Jeremy Dai <[email protected]>
Co-authored-by: Jonathan Wang <[email protected]>
Co-authored-by: calmini <[email protected]>

Author: 4 people

pyproject.toml

@@ -29,6 +29,8 @@ dependencies = [
     "boto3>=1.37.25",
     "loguru>=0.7.3",
     "ruamel-yaml>=0.18.10",
+    "aiohttp>=3.11.16",
+    "watchfiles>=1.0.4",
 ]
 
 [project.urls]

src/mcpm/router/README.md

@@ -0,0 +1,69 @@
+# MCPM Router
+
+The MCPM Router is a module that allows you to aggregate multiple MCP servers (both SSE and STDIO) and expose them as a single SSE server.
+
+## Features
+
+- Aggregate multiple MCP servers as a single server
+- Support both SSE and STDIO connections to underlying servers
+- Namespace capabilities from different servers
+- Expose a unified SSE server interface
+
+## Usage
+
+### Basic Usage
+
+```python
+import asyncio
+from mcpm.router import MCPRouter
+from mcpm.schema.server_config import STDIOServerConfig, SSEServerConfig
+
+async def main():
+    # Create a router
+    router = MCPRouter()
+
+    # Add a STDIO server
+    await router.add_server(
+        "example1",
+        STDIOServerConfig(
+            command="python",
+            args=["-m", "mcp.examples.simple_server"]
+        )
+    )
+
+    # Add an SSE server
+    await router.add_server(
+        "example2",
+        SSEServerConfig(
+            url="http://localhost:3000/sse"
+        )
+    )
+
+    # Start the SSE server
+    await router.start_sse_server(host="localhost", port=8080)
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+### Configuration File
+by default we use the configure from file `~/.config/mcpm/profile.json` to manage the servers.
+
+## Implementation Details
+
+The router works by:
+
+1. Connecting to each downstream server (SSE or STDIO)
+2. Collecting capabilities, tools, prompts, and resources from each server
+3. Namespacing these capabilities to avoid conflicts
+4. Creating an aggregated server that routes requests to the appropriate downstream server
+5. Exposing this aggregated server as an SSE server
+
+## Namespacing
+
+The router uses the following namespacing conventions:
+
+- Tools: `{server_name}_t_{tool_name}`
+- Prompts: `{server_name}_p_{prompt_name}`
+- Resources: `{server_name}:{resource_uri}`
+
+This allows the router to route requests to the appropriate server based on the namespaced identifier.

src/mcpm/router/__init__.py

@@ -0,0 +1,7 @@
+"""MCP Router Package"""
+
+from .router import MCPRouter
+
+__all__ = [
+    "MCPRouter",
+]

src/mcpm/router/client_connection.py

@@ -0,0 +1,69 @@
+import asyncio
+import logging
+from typing import Optional, cast
+
+from mcp import ClientSession, InitializeResult, StdioServerParameters, stdio_client
+from mcp.client.sse import sse_client
+
+from mcpm.schemas.server_config import ServerConfig, SSEServerConfig, STDIOServerConfig
+
+logger = logging.getLogger(__name__)
+
+
+def _stdio_transport_context(server_config: ServerConfig):
+    server_config = cast(STDIOServerConfig, server_config)
+    server_params = StdioServerParameters(command=server_config.command, args=server_config.args, env=server_config.env)
+    return stdio_client(server_params)
+
+
+def _sse_transport_context(server_config: ServerConfig):
+    server_config = cast(SSEServerConfig, server_config)
+    return sse_client(server_config.url, headers=server_config.headers)
+
+
+class ServerConnection:
+    def __init__(self, server_config: ServerConfig) -> None:
+        self.session: Optional[ClientSession] = None
+        self.session_initialized_response: Optional[InitializeResult] = None
+        self._initialized = False
+        self.server_config = server_config
+        self._initialized_event = asyncio.Event()
+        self._shutdown_event = asyncio.Event()
+
+        self._transport_context_factory = (
+            _stdio_transport_context if isinstance(server_config, STDIOServerConfig) else _sse_transport_context
+        )
+
+        self._server_task = asyncio.create_task(self._server_lifespan_cycle())
+
+    def healthy(self) -> bool:
+        return self.session is not None and self._initialized
+
+    # block until client session is initialized
+    async def wait_for_initialization(self):
+        await self._initialized_event.wait()
+
+    # request for client session to gracefully close
+    async def request_for_shutdown(self):
+        self._shutdown_event.set()
+
+    # block until client session is shutdown
+    async def wait_for_shutdown_request(self):
+        await self._shutdown_event.wait()
+
+    async def _server_lifespan_cycle(self):
+        try:
+            async with self._transport_context_factory(self.server_config) as (read, write):
+                async with ClientSession(read, write) as session:
+                    self.session_initialized_response = await session.initialize()
+
+                    self.session = session
+                    self._initialized = True
+                    self._initialized_event.set()
+                    # block here so that the session will not be closed after exit scope
+                    # we could retrieve alive session through self.session
+                    await self.wait_for_shutdown_request()
+        except Exception as e:
+            logger.error(f"Failed to connect to server {self.server_config.name}: {e}")
+            self._initialized_event.set()
+            self._shutdown_event.set()

src/mcpm/router/design.md

@@ -0,0 +1,158 @@
+# MCPM Router Design
+
+## Overview
+
+The MCPM Router serves as an intermediary layer within the MCPM (Model Context Protocol Manager) project. The router acts in a dual role:
+
+1. **As an MCP Client**: Connects to multiple downstream MCP servers
+2. **As an MCP Server**: Provides a unified interface to upstream MCP clients
+
+This design allows for aggregation of capabilities from multiple MCP servers while providing a single, stable connection point for clients.
+
+## Architecture
+
+```mermaid
+flowchart TB
+    subgraph clients[Upstream Clients]
+        c1[MCP Client 1]
+        c2[MCP Client 2]
+        cn[MCP Client N]
+    end
+    
+    subgraph router[MCPM Router]
+        r[MCPRouter]
+        ch[ClientHandler]
+        sh[ServerHandler]
+        cm[ConnectionManager]
+        r --> ch
+        r --> sh
+        ch --> cm
+        sh --> cm
+    end
+    
+    subgraph servers[Downstream Servers]
+        s1[MCP Server 1]
+        s2[MCP Server 2]
+        sm[MCP Server M]
+    end
+    
+    c1 <--SSE--> ch
+    c2 <--SSE--> ch
+    cn <--SSE--> ch
+    
+    sh <--STDIO/SSE--> s1
+    sh <--STDIO/SSE--> s2
+    sh <--STDIO/SSE--> sm
+    
+    classDef default fill:#f9f9f9,stroke:#333,stroke-width:1px;
+    classDef routerClass fill:#e1f5fe,stroke:#0277bd,stroke-width:2px;
+    class router routerClass;
+```
+
+## Key Components
+
+### `MCPRouter`
+
+The main orchestrator class that provides a unified API for the application:
+- Initializes and coordinates all internal components
+- Provides methods for connecting to downstream servers
+- Handles server and client routing through appropriate handlers
+- Manages namespacing of capabilities across different servers
+- Offers a clean, high-level interface for application code
+
+### `ConnectionManager`
+
+Maintains a registry of connections to downstream servers, providing methods to:
+- Add, retrieve, and remove downstream server connections
+- Lookup downstream servers by ID
+
+### `ServerHandler`
+
+Manages connections to downstream MCP servers:
+- Establishes and maintains connections using stdio or SSE transports
+- Aggregates capabilities from all connected servers
+- Routes requests from upstream clients to the appropriate downstream server
+- Handles notifications from downstream servers
+
+### `ClientHandler`
+
+Serves upstream MCP clients:
+- Provides an SSE server endpoint for clients to connect
+- Handles client connections/disconnections transparently
+- Routes client requests to the `ServerHandler`
+- Delivers responses and notifications back to clients
+
+### `ConnectionDetails` and `ConnectionType`
+
+Defines the configuration for connecting to downstream servers:
+- Supports multiple transport protocols (STDIO, SSE)
+- Validates configuration based on the connection type
+- Stores necessary connection parameters (command, args, env, URL)
+
+## Communication Flow
+
+### Downstream Connections (Router as Client)
+
+1. Router creates persistent connections to downstream MCP servers using STDIO or SSE
+2. Connections are maintained regardless of upstream client presence
+3. Server capabilities are fetched and aggregated with namespacing
+4. Notifications from servers are routed to appropriate upstream clients
+
+### Upstream Connections (Router as Server)
+
+1. Router provides an SSE server interface for upstream clients
+2. Clients can connect/disconnect at will without affecting downstream connections
+3. Client requests are routed to appropriate downstream servers
+4. Responses and notifications are delivered back to clients
+
+## Request Routing
+
+```mermaid
+sequenceDiagram
+    participant Client as MCP Client
+    participant CH as ClientHandler
+    participant Router as MCPRouter
+    participant SH as ServerHandler
+    participant Server as MCP Server
+    
+    Client->>CH: Request
+    CH->>Router: Forward request
+    Router->>Router: Parse namespaced ID
+    Router->>SH: Route to appropriate server
+    SH->>Server: Forward request with denormalized ID
+    Server->>SH: Response
+    SH->>Router: Forward response
+    Router->>CH: Route to originating client
+    CH->>Client: Deliver response
+```
+
+## Capability Aggregation
+
+1. Upon connection to a downstream server, all capabilities are fetched
+2. Capabilities (tools, resources, prompts) are namespaced with server ID
+3. Namespaced capabilities are added to aggregate pool
+4. Clients see all capabilities from all servers as a unified collection
+5. When a server disconnects, its capabilities are removed from the pool
+
+## Error Handling
+
+1. Connection errors are isolated to affected servers
+2. Standard JSON-RPC error responses for client requests
+3. Proper error propagation from downstream servers to clients
+4. Graceful handling of client and server disconnections
+
+## Benefits of This Design
+
+1. **Decoupling**: Upstream clients are decoupled from downstream servers
+2. **Resilience**: Client disconnections don't affect server connections
+3. **Aggregation**: Multiple capabilities from different servers appear as one
+4. **Flexibility**: Supports different transport protocols (STDIO, SSE)
+5. **Scalability**: Can manage multiple clients and servers simultaneously
+6. **Clean API**: The `MCPRouter` provides a simple, unified interface for applications
+
+## Implementation Notes
+
+- All communication follows the MCP protocol specification
+- Asynchronous operation using Python's asyncio
+- Type-safe interfaces using Python type hints
+- Clean separation of concerns between components 

src/mcpm/router/example.py

@@ -0,0 +1,55 @@
+"""
+Example script demonstrating how to use the MCPRouter to aggregate multiple MCP servers.
+"""
+
+import argparse
+import asyncio
+import logging
+from typing import List
+
+from .router import MCPRouter
+
+# Configure logging
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+logger = logging.getLogger(__name__)
+
+
+async def main(host: str, port: int, allow_origins: List[str] = None):
+    """
+    Main function to run the router example.
+
+    Args:
+        host: Host to bind the SSE server to
+        port: Port to bind the SSE server to
+        allow_origins: List of allowed origins for CORS
+    """
+    router = MCPRouter()
+
+    logger.info(f"Starting MCPRouter - will expose SSE server on http://{host}:{port}")
+
+    # Start the SSE server
+    try:
+        logger.info(f"Starting SSE server on http://{host}:{port}")
+        if allow_origins:
+            logger.info(f"CORS enabled for origins: {allow_origins}")
+        await router.start_sse_server(host, port, allow_origins)
+    except KeyboardInterrupt:
+        logger.info("Shutting down...")
+    except Exception as e:
+        logger.error(f"Error starting SSE server: {e}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="MCP Router Example")
+    parser.add_argument("--host", type=str, default="localhost", help="Host to bind the SSE server to")
+    parser.add_argument("--port", type=int, default=8080, help="Port to bind the SSE server to")
+    parser.add_argument("--cors", type=str, help="Comma-separated list of allowed origins for CORS")
+
+    args = parser.parse_args()
+
+    # Parse CORS origins
+    allow_origins = None
+    if args.cors:
+        allow_origins = [origin.strip() for origin in args.cors.split(",")]
+
+    asyncio.run(main(args.host, args.port, allow_origins))