diff --git a/README.md b/README.md index 56aa3609a..a0d47606e 100644 --- a/README.md +++ b/README.md @@ -410,6 +410,43 @@ 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 diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index 1e5b69eba..24fe97105 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/src/mcp/server/fastmcp/server.py @@ -88,6 +88,7 @@ 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/" @@ -184,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 @@ -197,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.""" @@ -558,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, @@ -573,7 +579,33 @@ 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: @@ -585,10 +617,20 @@ def sse_app(self) -> Starlette: 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, message_dispatch=message_dispatch + normalized_message_endpoint, + message_dispatch=message_dispatch ) async def handle_sse(scope: Scope, receive: Receive, send: Send): diff --git a/tests/server/fastmcp/test_server.py b/tests/server/fastmcp/test_server.py index 772c41529..64700d959 100644 --- a/tests/server/fastmcp/test_server.py +++ b/tests/server/fastmcp/test_server.py @@ -1,9 +1,11 @@ import base64 from pathlib import Path from typing import TYPE_CHECKING +from unittest.mock import patch import pytest from pydantic import AnyUrl +from starlette.routing import Mount, Route from mcp.server.fastmcp import Context, FastMCP from mcp.server.fastmcp.prompts.base import EmbeddedResource, Message, UserMessage @@ -31,6 +33,97 @@ async def test_create_server(self): assert mcp.name == "FastMCP" assert mcp.instructions == "Server instructions" + @pytest.mark.anyio + async def test_normalize_path(self): + """Test path normalization for mount paths.""" + mcp = FastMCP() + + # Test root path + assert mcp._normalize_path("/", "/messages/") == "/messages/" + + # Test path with trailing slash + assert mcp._normalize_path("/github/", "/messages/") == "/github/messages/" + + # Test path without trailing slash + assert mcp._normalize_path("/github", "/messages/") == "/github/messages/" + + # Test endpoint without leading slash + assert mcp._normalize_path("/github", "messages/") == "/github/messages/" + + # Test both with trailing/leading slashes + assert mcp._normalize_path("/api/", "/v1/") == "/api/v1/" + + @pytest.mark.anyio + async def test_sse_app_with_mount_path(self): + """Test SSE app creation with different mount paths.""" + # Test with default mount path + mcp = FastMCP() + with patch.object( + mcp, "_normalize_path", return_value="/messages/" + ) as mock_normalize: + mcp.sse_app() + # Verify _normalize_path was called with correct args + mock_normalize.assert_called_once_with("/", "/messages/") + + # Test with custom mount path in settings + mcp = FastMCP() + mcp.settings.mount_path = "/custom" + with patch.object( + mcp, "_normalize_path", return_value="/custom/messages/" + ) as mock_normalize: + mcp.sse_app() + # Verify _normalize_path was called with correct args + mock_normalize.assert_called_once_with("/custom", "/messages/") + + # Test with mount_path parameter + mcp = FastMCP() + with patch.object( + mcp, "_normalize_path", return_value="/param/messages/" + ) as mock_normalize: + mcp.sse_app(mount_path="/param") + # Verify _normalize_path was called with correct args + mock_normalize.assert_called_once_with("/param", "/messages/") + + @pytest.mark.anyio + async def test_starlette_routes_with_mount_path(self): + """Test that Starlette routes are correctly configured with mount path.""" + # Test with mount path in settings + mcp = FastMCP() + mcp.settings.mount_path = "/api" + app = mcp.sse_app() + + # Find routes by type + sse_routes = [r for r in app.routes if isinstance(r, Route)] + mount_routes = [r for r in app.routes if isinstance(r, Mount)] + + # Verify routes exist + assert len(sse_routes) == 1, "Should have one SSE route" + assert len(mount_routes) == 1, "Should have one mount route" + + # Verify path values + assert sse_routes[0].path == "/sse", "SSE route path should be /sse" + assert ( + mount_routes[0].path == "/messages" + ), "Mount route path should be /messages" + + # Test with mount path as parameter + mcp = FastMCP() + app = mcp.sse_app(mount_path="/param") + + # Find routes by type + sse_routes = [r for r in app.routes if isinstance(r, Route)] + mount_routes = [r for r in app.routes if isinstance(r, Mount)] + + # Verify routes exist + assert len(sse_routes) == 1, "Should have one SSE route" + assert len(mount_routes) == 1, "Should have one mount route" + + # Verify path values + assert sse_routes[0].path == "/sse", "SSE route path should be /sse" + assert ( + mount_routes[0].path == "/messages" + ), "Mount route path should be /messages" + @pytest.mark.anyio async def test_non_ascii_description(self): """Test that FastMCP handles non-ASCII characters in descriptions correctly""" @@ -518,8 +611,6 @@ async def async_tool(x: int, ctx: Context) -> str: @pytest.mark.anyio async def test_context_logging(self): - from unittest.mock import patch - import mcp.server.session """Test that context logging methods work."""