diff --git a/README.md b/README.md index 0ca039ae..be248b4b 100644 --- a/README.md +++ b/README.md @@ -383,6 +383,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). ## Examples diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index aa240da7..35e859d4 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/src/mcp/server/fastmcp/server.py @@ -71,6 +71,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/" @@ -143,11 +144,14 @@ 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: """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 @@ -156,7 +160,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.""" @@ -462,11 +466,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: """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, @@ -477,9 +481,42 @@ 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 = "") -> Starlette: """Return an instance of the SSE server app.""" - sse = SseServerTransport(self.settings.message_path) + # Update mount_path in settings if provided + if mount_path: + self.settings.mount_path = mount_path + # Create normalized endpoint considering the mount path + normalized_endpoint = self._normalize_path( + self.settings.mount_path, self.settings.message_path + ) + sse = SseServerTransport(normalized_endpoint) async def handle_sse(request: Request) -> None: async with sse.connect_sse( diff --git a/tests/server/fastmcp/test_server.py b/tests/server/fastmcp/test_server.py index e76e59c5..39f9e78c 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."""