Add base_path support for SSE endpoint subpath routing via nginx or other http middleware

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:

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,13 +845,15 @@ 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:
         """Create a Starlette app using the specified HTTP transport.
 
         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,

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"

tests/server/test_server.py

@@ -1,11 +1,13 @@
 from typing import Annotated
+from unittest.mock import patch
 
 import pytest
 from mcp.types import (
     TextContent,
     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"
+        )