diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index e5b6c3acc..784d71e7c 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/src/mcp/server/fastmcp/server.py @@ -618,7 +618,7 @@ async def run_streamable_http_async(self) -> None: import uvicorn starlette_app = self.streamable_http_app() - + config = uvicorn.Config( starlette_app, host=self.settings.host, @@ -770,7 +770,7 @@ async def sse_endpoint(request: Request) -> Response: def streamable_http_app(self) -> Starlette: """Return an instance of the StreamableHTTP server app.""" from starlette.middleware import Middleware - from starlette.routing import Mount + from starlette.routing import Mount, Router # Create session manager on first call (lazy initialization) if self._session_manager is None: @@ -787,12 +787,26 @@ async def handle_streamable_http( ) -> None: await self.session_manager.handle_request(scope, receive, send) - # Create routes + async def streamable_http_endpoint(request: Request): + return await handle_streamable_http( + request.scope, request.receive, request._send + ) # type: ignore[reportPrivateUsage] + + # Normalize the main path (no trailing slash) + _main_path = self.settings.streamable_http_path.removesuffix("/") + + streamable_router = Router( + routes=[ + Route("/", endpoint=streamable_http_endpoint, methods=["GET", "POST"]), + ], + redirect_slashes=False, + ) + routes: list[Route | Mount] = [] middleware: list[Middleware] = [] required_scopes = [] - # Add auth endpoints if auth provider is configured + # Auth endpoints if auth provider is configured if self._auth_server_provider: assert self.settings.auth from mcp.server.auth.routes import create_auth_routes @@ -817,18 +831,19 @@ async def handle_streamable_http( revocation_options=self.settings.auth.revocation_options, ) ) + routes.append( Mount( - self.settings.streamable_http_path, - app=RequireAuthMiddleware(handle_streamable_http, required_scopes), + _main_path, + app=RequireAuthMiddleware(streamable_router, required_scopes), ) ) else: # Auth is disabled, no wrapper needed routes.append( Mount( - self.settings.streamable_http_path, - app=handle_streamable_http, + _main_path, + app=streamable_router, ) ) diff --git a/tests/server/fastmcp/test_server.py b/tests/server/fastmcp/test_server.py index b817761ea..a975974eb 100644 --- a/tests/server/fastmcp/test_server.py +++ b/tests/server/fastmcp/test_server.py @@ -124,6 +124,20 @@ async def test_starlette_routes_with_mount_path(self): mount_routes[0].path == "/messages" ), "Mount route path should be /messages" + mcp = FastMCP() + app = mcp.streamable_http_app() + + # Find routes by type + streamable_routes = [r for r in app.routes if isinstance(r, Mount)] + + # Verify routes exist + assert len(streamable_routes) == 1, "Should have two streamable routes" + + # Verify path values + assert ( + streamable_routes[0].path == "/mcp" + ), "Streamable route path should be /mcp" + @pytest.mark.anyio async def test_non_ascii_description(self): """Test that FastMCP handles non-ASCII characters in descriptions correctly"""