Merge branch 'main' into ihrpr/auth-example

Author: ihrpr

README.md

@@ -410,8 +410,69 @@ app = Starlette(
 app.router.routes.append(Host('mcp.acme.corp', app=mcp.sse_app()))
 ```
 
+When mounting multiple MCP servers under different paths, you can configure the mount path in several ways:
+
+```python
+from starlette.applications import Starlette
+from starlette.routing import Mount
+from mcp.server.fastmcp import FastMCP
+
+# Create multiple MCP servers
+github_mcp = FastMCP("GitHub API")
+browser_mcp = FastMCP("Browser")
+curl_mcp = FastMCP("Curl")
+search_mcp = FastMCP("Search")
+
+# Method 1: Configure mount paths via settings (recommended for persistent configuration)
+github_mcp.settings.mount_path = "/github"
+browser_mcp.settings.mount_path = "/browser"
+
+# Method 2: Pass mount path directly to sse_app (preferred for ad-hoc mounting)
+# This approach doesn't modify the server's settings permanently
+
+# Create Starlette app with multiple mounted servers
+app = Starlette(
+    routes=[
+        # Using settings-based configuration
+        Mount("/github", app=github_mcp.sse_app()),
+        Mount("/browser", app=browser_mcp.sse_app()),
+        # Using direct mount path parameter
+        Mount("/curl", app=curl_mcp.sse_app("/curl")),
+        Mount("/search", app=search_mcp.sse_app("/search")),
+    ]
+)
+
+# Method 3: For direct execution, you can also pass the mount path to run()
+if __name__ == "__main__":
+    search_mcp.run(transport="sse", mount_path="/search")
+```
+
 For more information on mounting applications in Starlette, see the [Starlette documentation](https://www.starlette.io/routing/#submounting-routes).
 
+#### Message Dispatch Options
+
+By default, the SSE server uses an in-memory message dispatch system for incoming POST messages. For production deployments or distributed scenarios, you can use Redis or implement your own message dispatch system that conforms to the `MessageDispatch` protocol:
+
+```python
+# Using the built-in Redis message dispatch
+from mcp.server.fastmcp import FastMCP
+from mcp.server.message_queue import RedisMessageDispatch
+
+# Create a Redis message dispatch
+redis_dispatch = RedisMessageDispatch(
+    redis_url="redis://localhost:6379/0", prefix="mcp:pubsub:"
+)
+
+# Pass the message dispatch instance to the server
+mcp = FastMCP("My App", message_queue=redis_dispatch)
+```
+
+To use Redis, add the Redis dependency:
+
+```bash
+uv add "mcp[redis]"
+```
+
 ## Examples
 
 ### Echo Server

examples/servers/simple-prompt/mcp_simple_prompt/server.py

@@ -88,12 +88,15 @@ async def get_prompt(
         )
 
     if transport == "sse":
+        from mcp.server.message_queue.redis import RedisMessageDispatch
         from mcp.server.sse import SseServerTransport
         from starlette.applications import Starlette
         from starlette.responses import Response
         from starlette.routing import Mount, Route
 
-        sse = SseServerTransport("/messages/")
+        message_dispatch = RedisMessageDispatch("redis://localhost:6379/0")
+
+        sse = SseServerTransport("/messages/", message_dispatch=message_dispatch)
 
         async def handle_sse(request):
             async with sse.connect_sse(

pyproject.toml

@@ -37,6 +37,7 @@ dependencies = [
 rich = ["rich>=13.9.4"]
 cli = ["typer>=0.12.4", "python-dotenv>=1.0.0"]
 ws = ["websockets>=15.0.1"]
+redis = ["redis>=5.2.1", "types-redis>=4.6.0.20241004"]
 
 [project.scripts]
 mcp = "mcp.cli:app [cli]"
@@ -55,6 +56,7 @@ dev = [
     "pytest-xdist>=3.6.1",
     "pytest-examples>=0.0.14",
     "pytest-pretty>=1.2.0",
+    "fakeredis==2.28.1",
 ]
 docs = [
     "mkdocs>=1.6.1",

src/mcp/client/sse.py

@@ -98,7 +98,9 @@ async def sse_reader(
                                             await read_stream_writer.send(exc)
                                             continue
 
-                                        session_message = SessionMessage(message)
+                                        session_message = SessionMessage(
+                                            message=message
+                                        )
                                         await read_stream_writer.send(session_message)
                                     case _:
                                         logger.warning(
@@ -148,3 +150,5 @@ async def post_writer(endpoint_url: str):
         finally:
             await read_stream_writer.aclose()
             await write_stream.aclose()
+            await read_stream.aclose()
+            await write_stream_reader.aclose()

src/mcp/client/stdio/__init__.py

@@ -144,7 +144,7 @@ async def stdout_reader():
                             await read_stream_writer.send(exc)
                             continue
 
-                        session_message = SessionMessage(message)
+                        session_message = SessionMessage(message=message)
                         await read_stream_writer.send(session_message)
         except anyio.ClosedResourceError:
             await anyio.lowlevel.checkpoint()

src/mcp/client/streamable_http.py

@@ -153,7 +153,7 @@ async def _handle_sse_event(
                 ):
                     message.root.id = original_request_id
 
-                session_message = SessionMessage(message)
+                session_message = SessionMessage(message=message)
                 await read_stream_writer.send(session_message)
 
                 # Call resumption token callback if we have an ID
@@ -286,7 +286,7 @@ async def _handle_json_response(
         try:
             content = await response.aread()
             message = JSONRPCMessage.model_validate_json(content)
-            session_message = SessionMessage(message)
+            session_message = SessionMessage(message=message)
             await read_stream_writer.send(session_message)
         except Exception as exc:
             logger.error(f"Error parsing JSON response: {exc}")
@@ -333,7 +333,7 @@ async def _send_session_terminated_error(
             id=request_id,
             error=ErrorData(code=32600, message="Session terminated"),
         )
-        session_message = SessionMessage(JSONRPCMessage(jsonrpc_error))
+        session_message = SessionMessage(message=JSONRPCMessage(jsonrpc_error))
         await read_stream_writer.send(session_message)
 
     async def post_writer(

src/mcp/client/websocket.py

@@ -60,7 +60,7 @@ async def ws_reader():
                 async for raw_text in ws:
                     try:
                         message = types.JSONRPCMessage.model_validate_json(raw_text)
-                        session_message = SessionMessage(message)
+                        session_message = SessionMessage(message=message)
                         await read_stream_writer.send(session_message)
                     except ValidationError as exc:
                         # If JSON parse or model validation fails, send the exception

src/mcp/server/fastmcp/server.py

@@ -44,6 +44,7 @@
 from mcp.server.lowlevel.server import LifespanResultT
 from mcp.server.lowlevel.server import Server as MCPServer
 from mcp.server.lowlevel.server import lifespan as default_lifespan
+from mcp.server.message_queue import MessageDispatch
 from mcp.server.session import ServerSession, ServerSessionT
 from mcp.server.sse import SseServerTransport
 from mcp.server.stdio import stdio_server
@@ -87,9 +88,15 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
     # HTTP settings
     host: str = "0.0.0.0"
     port: int = 8000
+    mount_path: str = "/"  # Mount path (e.g. "/github", defaults to root path)
     sse_path: str = "/sse"
     message_path: str = "/messages/"
 
+    # SSE message queue settings
+    message_dispatch: MessageDispatch | None = Field(
+        None, description="Custom message dispatch instance"
+    )
+
     # resource settings
     warn_on_duplicate_resources: bool = True
 
@@ -178,11 +185,16 @@ def name(self) -> str:
     def instructions(self) -> str | None:
         return self._mcp_server.instructions
 
-    def run(self, transport: Literal["stdio", "sse"] = "stdio") -> None:
+    def run(
+        self,
+        transport: Literal["stdio", "sse"] = "stdio",
+        mount_path: str | None = None,
+    ) -> None:
         """Run the FastMCP server. Note this is a synchronous function.
 
         Args:
             transport: Transport protocol to use ("stdio" or "sse")
+            mount_path: Optional mount path for SSE transport
         """
         TRANSPORTS = Literal["stdio", "sse"]
         if transport not in TRANSPORTS.__args__:  # type: ignore
@@ -191,7 +203,7 @@ def run(self, transport: Literal["stdio", "sse"] = "stdio") -> None:
         if transport == "stdio":
             anyio.run(self.run_stdio_async)
         else:  # transport == "sse"
-            anyio.run(self.run_sse_async)
+            anyio.run(lambda: self.run_sse_async(mount_path))
 
     def _setup_handlers(self) -> None:
         """Set up core MCP protocol handlers."""
@@ -552,11 +564,11 @@ async def run_stdio_async(self) -> None:
                 self._mcp_server.create_initialization_options(),
             )
 
-    async def run_sse_async(self) -> None:
+    async def run_sse_async(self, mount_path: str | None = None) -> None:
         """Run the server using SSE transport."""
         import uvicorn
 
-        starlette_app = self.sse_app()
+        starlette_app = self.sse_app(mount_path)
 
         config = uvicorn.Config(
             starlette_app,
@@ -567,14 +579,59 @@ async def run_sse_async(self) -> None:
         server = uvicorn.Server(config)
         await server.serve()
 
-    def sse_app(self) -> Starlette:
+    def _normalize_path(self, mount_path: str, endpoint: str) -> str:
+        """
+        Combine mount path and endpoint to return a normalized path.
+
+        Args:
+            mount_path: The mount path (e.g. "/github" or "/")
+            endpoint: The endpoint path (e.g. "/messages/")
+
+        Returns:
+            Normalized path (e.g. "/github/messages/")
+        """
+        # Special case: root path
+        if mount_path == "/":
+            return endpoint
+
+        # Remove trailing slash from mount path
+        if mount_path.endswith("/"):
+            mount_path = mount_path[:-1]
+
+        # Ensure endpoint starts with slash
+        if not endpoint.startswith("/"):
+            endpoint = "/" + endpoint
+
+        # Combine paths
+        return mount_path + endpoint
+
+    def sse_app(self, mount_path: str | None = None) -> Starlette:
         """Return an instance of the SSE server app."""
+        message_dispatch = self.settings.message_dispatch
+        if message_dispatch is None:
+            from mcp.server.message_queue import InMemoryMessageDispatch
+
+            message_dispatch = InMemoryMessageDispatch()
+            logger.info("Using default in-memory message dispatch")
+
         from starlette.middleware import Middleware
         from starlette.routing import Mount, Route
 
+        # Update mount_path in settings if provided
+        if mount_path is not None:
+            self.settings.mount_path = mount_path
+
+        # Create normalized endpoint considering the mount path
+        normalized_message_endpoint = self._normalize_path(
+            self.settings.mount_path, self.settings.message_path
+        )
+        
         # Set up auth context and dependencies
 
-        sse = SseServerTransport(self.settings.message_path)
+        sse = SseServerTransport(
+          normalized_message_endpoint,
+          message_dispatch=message_dispatch
+        )
 
         async def handle_sse(scope: Scope, receive: Receive, send: Send):
             # Add client ID from auth context into request context if available
@@ -589,7 +646,14 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send):
                     streams[1],
                     self._mcp_server.create_initialization_options(),
                 )
-            return Response()
+                return Response()
+
+        @asynccontextmanager
+        async def lifespan(app: Starlette):
+            try:
+                yield
+            finally:
+                await message_dispatch.close()
 
         # Create routes
         routes: list[Route | Mount] = []
@@ -666,7 +730,10 @@ async def sse_endpoint(request: Request) -> None:
 
         # Create Starlette app with routes and middleware
         return Starlette(
-            debug=self.settings.debug, routes=routes, middleware=middleware
+            debug=self.settings.debug,
+            routes=routes,
+            middleware=middleware,
+            lifespan=lifespan,
         )
 
     async def list_prompts(self) -> list[MCPPrompt]:

src/mcp/server/message_queue/__init__.py

@@ -0,0 +1,16 @@
+"""
+Message Dispatch Module for MCP Server
+
+This module implements dispatch interfaces for handling
+messages between clients and servers.
+"""
+
+from mcp.server.message_queue.base import InMemoryMessageDispatch, MessageDispatch
+
+# Try to import Redis implementation if available
+try:
+    from mcp.server.message_queue.redis import RedisMessageDispatch
+except ImportError:
+    RedisMessageDispatch = None
+
+__all__ = ["MessageDispatch", "InMemoryMessageDispatch", "RedisMessageDispatch"]