diff --git a/src/fastmcp/server/http.py b/src/fastmcp/server/http.py index 5257c24b7..0400d93c1 100644 --- a/src/fastmcp/server/http.py +++ b/src/fastmcp/server/http.py @@ -135,10 +135,36 @@ def create_base_app( ) +def _normalize_path(base_path: str, endpoint: str) -> str: + """ + Combine base path and endpoint to return a normalized path. + Args: + base_path: The base path (e.g. "/mcp" or "/") + endpoint: The endpoint path (e.g. "/messages/") + Returns: + Normalized path (e.g. "/mcp/messages/") + """ + # Special case: root path + if base_path == "/": + return endpoint + + # Remove trailing slash from mount path + if base_path.endswith("/"): + base_path = base_path[:-1] + + # Ensure endpoint starts with slash + if not endpoint.startswith("/"): + endpoint = "/" + endpoint + + # Combine paths + return base_path + endpoint + + def create_sse_app( server: FastMCP, message_path: str, sse_path: str, + base_path: str | None = None, auth_server_provider: OAuthAuthorizationServerProvider | None = None, auth_settings: AuthSettings | None = None, debug: bool = False, @@ -151,6 +177,7 @@ def create_sse_app( server: The FastMCP server instance message_path: Path for SSE messages sse_path: Path for SSE connections + base_path: Optional base path for SSE transport auth_server_provider: Optional auth provider auth_settings: Optional auth settings debug: Whether to enable debug mode @@ -163,8 +190,13 @@ def create_sse_app( server_routes: list[BaseRoute] = [] server_middleware: list[Middleware] = [] + normalized_message_endpoint = message_path + if base_path is not None: + # Create normalized endpoint considering the base path + normalized_message_endpoint = _normalize_path(base_path, message_path) + # Set up SSE transport - sse = SseServerTransport(message_path) + sse = SseServerTransport(normalized_message_endpoint) # Create handler for SSE connections async def handle_sse(scope: Scope, receive: Receive, send: Send) -> Response: diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py index 9dabc8735..ed0bbab84 100644 --- a/src/fastmcp/server/server.py +++ b/src/fastmcp/server/server.py @@ -203,6 +203,7 @@ def run( Args: transport: Transport protocol to use ("stdio", "sse", or "streamable-http") + mount_path: Optional mount path for SSE transport """ logger.info(f'Starting server "{self.name}"...') @@ -723,6 +724,7 @@ async def run_http_async( transport: Literal["streamable-http", "sse"] = "streamable-http", host: str | None = None, port: int | None = None, + base_path: str | None = None, log_level: str | None = None, path: str | None = None, uvicorn_config: dict | None = None, @@ -733,6 +735,7 @@ async def run_http_async( transport: Transport protocol to use - either "streamable-http" (default) or "sse" host: Host address to bind to (defaults to settings.host) port: Port to bind to (defaults to settings.port) + base_path: Optional base path for SSE transport log_level: Log level for the server (defaults to settings.log_level) path: Path for the endpoint (defaults to settings.streamable_http_path or settings.sse_path) uvicorn_config: Additional configuration for the Uvicorn server @@ -742,7 +745,11 @@ async def run_http_async( # lifespan is required for streamable http uvicorn_config["lifespan"] = "on" - app = self.http_app(path=path, transport=transport) + app = self.http_app( + base_path=base_path or self.settings.base_path, + path=path, + transport=transport, + ) config = uvicorn.Config( app, @@ -838,6 +845,7 @@ def streamable_http_app( def http_app( self, path: str | None = None, + base_path: str | None = None, middleware: list[Middleware] | None = None, transport: Literal["streamable-http", "sse"] = "streamable-http", ) -> Starlette: @@ -845,6 +853,7 @@ def http_app( Args: path: The path for the HTTP endpoint + base_path: Optional base path for SSE transport middleware: A list of middleware to apply to the app transport: Transport protocol to use - either "streamable-http" (default) or "sse" @@ -867,10 +876,15 @@ def http_app( middleware=middleware, ) elif transport == "sse": + # Update base_path in settings if provided + if base_path is not None: + self.settings.base_path = base_path + return create_sse_app( server=self, message_path=self.settings.message_path, sse_path=path or self.settings.sse_path, + base_path=self.settings.base_path, auth_server_provider=self._auth_server_provider, auth_settings=self.settings.auth, debug=self.settings.debug, diff --git a/src/fastmcp/settings.py b/src/fastmcp/settings.py index 21395ce7c..a44f32271 100644 --- a/src/fastmcp/settings.py +++ b/src/fastmcp/settings.py @@ -59,6 +59,7 @@ class ServerSettings(BaseSettings): # HTTP settings host: str = "127.0.0.1" port: int = 8000 + base_path: str = "/" sse_path: str = "/sse" message_path: str = "/messages/" streamable_http_path: str = "/mcp" diff --git a/tests/server/test_server.py b/tests/server/test_server.py index ead1dd673..b391ca189 100644 --- a/tests/server/test_server.py +++ b/tests/server/test_server.py @@ -1,4 +1,5 @@ from typing import Annotated +from unittest.mock import patch import pytest from mcp.types import ( @@ -6,6 +7,7 @@ TextResourceContents, ) from pydantic import Field +from starlette.routing import Mount, Route from fastmcp import Client, FastMCP from fastmcp.exceptions import ClientError, NotFoundError @@ -734,3 +736,82 @@ def sample_prompt() -> str: assert len(prompts_dict) == 1 prompt = prompts_dict["sample_prompt"] assert prompt.tags == {"example", "test-tag"} + + +class TestSSEUseBasePath: + def test_normalize(self): + from fastmcp.server.http import _normalize_path + + # Test root path + assert _normalize_path("/", "/messages/") == "/messages/" + + # Test path with trailing slash + assert _normalize_path("/mcp/", "/messages/") == "/mcp/messages/" + + # Test path without trailing slash + assert _normalize_path("/mcp", "/messages/") == "/mcp/messages/" + + # Test endpoint without leading slash + assert _normalize_path("/mcp", "messages/") == "/mcp/messages/" + + # Test both with trailing/leading slashes + assert _normalize_path("/api/", "/v1/") == "/api/v1/" + + async def test_http_app_base_path(self): + mcp = FastMCP() + with patch( + "fastmcp.server.http._normalize_path", return_value="/messages/" + ) as mock_normalize: + mcp.http_app(transport="sse") + mock_normalize.assert_called_once_with("/", "/messages/") + + mcp = FastMCP() + with patch( + "fastmcp.server.http._normalize_path", return_value="/mcp/messages/" + ) as mock_normalize: + mcp.http_app(base_path="/mcp", transport="sse") + mock_normalize.assert_called_once_with("/mcp", "/messages/") + + mcp = FastMCP() + mcp.settings.base_path = "/api" + with patch( + "fastmcp.server.http._normalize_path", return_value="/api/messages/" + ) as mock_normalize: + mcp.http_app(transport="sse") + mock_normalize.assert_called_once_with("/api", "/messages/") + + async def test_starlette_routes_not_change_with_base_path(self): + mcp = FastMCP() + app = mcp.http_app(base_path="/mcp", transport="sse") + + # 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" + ) + + mcp = FastMCP() + mcp.settings.base_path = "/api" + app = mcp.http_app(transport="sse") + + # 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" + )