Move sse app to http

src/fastmcp/server/http.py

@@ -3,11 +3,30 @@ from __future__ import annotations
 from collections.abc import Generator
 from contextlib import contextmanager
 from contextvars import ContextVar
+from typing import TYPE_CHECKING
 
+from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
+from mcp.server.auth.middleware.bearer_auth import (
+    BearerAuthBackend,
+    RequireAuthMiddleware,
+)
+from mcp.server.auth.provider import OAuthAuthorizationServerProvider
+from mcp.server.auth.routes import create_auth_routes
+from mcp.server.auth.settings import AuthSettings
+from mcp.server.sse import SseServerTransport
+from starlette.applications import Starlette
+from starlette.middleware import Middleware
+from starlette.middleware.authentication import AuthenticationMiddleware
 from starlette.requests import Request
+from starlette.responses import Response
+from starlette.routing import Mount, Route
+from starlette.types import Receive, Scope, Send
 
 from fastmcp.utilities.logging import get_logger
 
+if TYPE_CHECKING:
+    from fastmcp import FastMCP
+
 logger = get_logger(__name__)
 
 _current_http_request: ContextVar[Request | None] = ContextVar(
@@ -36,3 +55,122 @@ class RequestContextMiddleware:
     async def __call__(self, scope, receive, send):
         with set_http_request(Request(scope)):
             await self.app(scope, receive, send)
+
+
+def create_sse_app(
+    server: FastMCP,
+    message_path: str,
+    sse_path: str,
+    auth_server_provider: OAuthAuthorizationServerProvider | None = None,
+    auth_settings: AuthSettings | None = None,
+    debug: bool = False,
+    additional_routes: list[Route] | list[Mount] | list[Route | Mount] | None = None,
+) -> Starlette:
+    """Return an instance of the SSE server app.
+
+    Args:
+        server: The FastMCP server instance
+        message_path: Path for SSE messages
+        sse_path: Path for SSE connections
+        auth_server_provider: Optional auth provider
+        auth_settings: Optional auth settings
+        debug: Whether to enable debug mode
+        additional_routes: Optional list of custom routes
+
+    Returns:
+        A Starlette application configured for SSE
+    """
+
+    # Set up SSE transport
+    sse = SseServerTransport(message_path)
+
+    async def handle_sse(scope: Scope, receive: Receive, send: Send):
+        # Add client ID from auth context into request context if available
+        async with sse.connect_sse(
+            scope,
+            receive,
+            send,
+        ) as streams:
+            await server._mcp_server.run(
+                streams[0],
+                streams[1],
+                server._mcp_server.create_initialization_options(),
+            )
+        return Response()
+
+    # Create routes
+    routes: list[Route | Mount] = []
+    middleware: list[Middleware] = []
+    required_scopes = []
+
+    # Add auth endpoints if auth provider is configured
+    if auth_server_provider:
+        assert auth_settings
+
+        required_scopes = auth_settings.required_scopes or []
+
+        middleware = [
+            # extract auth info from request (but do not require it)
+            Middleware(
+                AuthenticationMiddleware,
+                backend=BearerAuthBackend(
+                    provider=auth_server_provider,
+                ),
+            ),
+            # Add the auth context middleware to store
+            # authenticated user in a contextvar
+            Middleware(AuthContextMiddleware),
+        ]
+        routes.extend(
+            create_auth_routes(
+                provider=auth_server_provider,
+                issuer_url=auth_settings.issuer_url,
+                service_documentation_url=auth_settings.service_documentation_url,
+                client_registration_options=auth_settings.client_registration_options,
+                revocation_options=auth_settings.revocation_options,
+            )
+        )
+
+    # When auth is not configured, we shouldn't require auth
+    if auth_server_provider:
+        # Auth is enabled, wrap the endpoints with RequireAuthMiddleware
+        routes.append(
+            Route(
+                sse_path,
+                endpoint=RequireAuthMiddleware(handle_sse, required_scopes),
+                methods=["GET"],
+            )
+        )
+        routes.append(
+            Mount(
+                message_path,
+                app=RequireAuthMiddleware(sse.handle_post_message, required_scopes),
+            )
+        )
+    else:
+        # Auth is disabled, no need for RequireAuthMiddleware
+        # Since handle_sse is an ASGI app, we need to create a compatible endpoint
+        async def sse_endpoint(request: Request) -> None:
+            # Convert the Starlette request to ASGI parameters
+            await handle_sse(request.scope, request.receive, request._send)  # type: ignore[reportPrivateUsage]
+
+        routes.append(
+            Route(
+                sse_path,
+                endpoint=sse_endpoint,
+                methods=["GET"],
+            )
+        )
+        routes.append(
+            Mount(
+                message_path,
+                app=sse.handle_post_message,
+            )
+        )
+
+    # mount custom routes last, so they have the lowest route matching precedence
+    if additional_routes:
+        routes.extend(additional_routes)
+
+    # Create Starlette app with routes and middleware
+    return Starlette(debug=debug, routes=routes, middleware=middleware)

src/fastmcp/server/server.py

@@ -16,16 +16,10 @@ import anyio
 import httpx
 import pydantic
 import uvicorn
-from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
-from mcp.server.auth.middleware.bearer_auth import (
-    BearerAuthBackend,
-    RequireAuthMiddleware,
-)
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider
 from mcp.server.lowlevel.helper_types import ReadResourceContents
 from mcp.server.lowlevel.server import LifespanResultT
 from mcp.server.lowlevel.server import Server as MCPServer
-from mcp.server.sse import SseServerTransport
 from mcp.server.stdio import stdio_server
 from mcp.types import (
     AnyFunction,
@@ -41,12 +35,9 @@ from mcp.types import ResourceTemplate as MCPResourceTemplate
 from mcp.types import Tool as MCPTool
 from pydantic import AnyUrl
 from starlette.applications import Starlette
-from starlette.middleware import Middleware
-from starlette.middleware.authentication import AuthenticationMiddleware
 from starlette.requests import Request
 from starlette.responses import Response
-from starlette.routing import Mount, Route
-from starlette.types import Receive, Scope, Send
+from starlette.routing import Route
 
 import fastmcp.server
 import fastmcp.settings
@@ -55,7 +46,7 @@ from fastmcp.prompts import Prompt, PromptManager
 from fastmcp.prompts.prompt import PromptResult
 from fastmcp.resources import Resource, ResourceManager
 from fastmcp.resources.template import ResourceTemplate
-from fastmcp.server.http import RequestContextMiddleware
+from fastmcp.server.http import RequestContextMiddleware, create_sse_app
 from fastmcp.tools import ToolManager
 from fastmcp.tools.tool import Tool
 from fastmcp.utilities.cache import TimedCache
@@ -155,7 +146,7 @@ class FastMCP(Generic[LifespanResultT]):
                 "is specified"
             )
         self._auth_server_provider = auth_server_provider
-        self._custom_starlette_routes: list[Route] = []
+        self._additional_http_routes: list[Route] = []
         self.dependencies = self.settings.dependencies
 
         # Set up MCP protocol handlers
@@ -294,7 +285,7 @@ class FastMCP(Generic[LifespanResultT]):
         def decorator(
             func: Callable[[Request], Awaitable[Response]],
         ) -> Callable[[Request], Awaitable[Response]]:
-            self._custom_starlette_routes.append(
+            self._additional_http_routes.append(
                 Route(
                     path,
                     endpoint=func,
@@ -742,104 +733,14 @@ class FastMCP(Generic[LifespanResultT]):
 
     def sse_app(self) -> Starlette:
         """Return an instance of the SSE server app."""
-        from starlette.middleware import Middleware
-        from starlette.routing import Mount, Route
-
-        # Set up auth context and dependencies
-
-        sse = SseServerTransport(self.settings.message_path)
-
-        async def handle_sse(scope: Scope, receive: Receive, send: Send):
-            # Add client ID from auth context into request context if available
-
-            async with sse.connect_sse(
-                scope,
-                receive,
-                send,
-            ) as streams:
-                await self._mcp_server.run(
-                    streams[0],
-                    streams[1],
-                    self._mcp_server.create_initialization_options(),
-                )
-            return Response()
-
-        # Create routes
-        routes: list[Route | Mount] = []
-        middleware: list[Middleware] = []
-        required_scopes = []
-
-        # Add 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
-
-            required_scopes = self.settings.auth.required_scopes or []
-
-            middleware = [
-                # extract auth info from request (but do not require it)
-                Middleware(
-                    AuthenticationMiddleware,
-                    backend=BearerAuthBackend(
-                        provider=self._auth_server_provider,
-                    ),
-                ),
-                # Add the auth context middleware to store
-                # authenticated user in a contextvar
-                Middleware(AuthContextMiddleware),
-            ]
-            routes.extend(
-                create_auth_routes(
-                    provider=self._auth_server_provider,
-                    issuer_url=self.settings.auth.issuer_url,
-                    service_documentation_url=self.settings.auth.service_documentation_url,
-                    client_registration_options=self.settings.auth.client_registration_options,
-                    revocation_options=self.settings.auth.revocation_options,
-                )
-            )
-
-        # When auth is not configured, we shouldn't require auth
-        if self._auth_server_provider:
-            # Auth is enabled, wrap the endpoints with RequireAuthMiddleware
-            routes.append(
-                Route(
-                    self.settings.sse_path,
-                    endpoint=RequireAuthMiddleware(handle_sse, required_scopes),
-                    methods=["GET"],
-                )
-            )
-            routes.append(
-                Mount(
-                    self.settings.message_path,
-                    app=RequireAuthMiddleware(sse.handle_post_message, required_scopes),
-                )
-            )
-        else:
-            # Auth is disabled, no need for RequireAuthMiddleware
-            # Since handle_sse is an ASGI app, we need to create a compatible endpoint
-            async def sse_endpoint(request: Request) -> None:
-                # Convert the Starlette request to ASGI parameters
-                await handle_sse(request.scope, request.receive, request._send)  # type: ignore[reportPrivateUsage]
-
-            routes.append(
-                Route(
-                    self.settings.sse_path,
-                    endpoint=sse_endpoint,
-                    methods=["GET"],
-                )
-            )
-            routes.append(
-                Mount(
-                    self.settings.message_path,
-                    app=sse.handle_post_message,
-                )
-            )
-        # mount these routes last, so they have the lowest route matching precedence
-        routes.extend(self._custom_starlette_routes)
-
-        # Create Starlette app with routes and middleware
-        return Starlette(
-            debug=self.settings.debug, routes=routes, middleware=middleware
+        return create_sse_app(
+            server=self,
+            message_path=self.settings.message_path,
+            sse_path=self.settings.sse_path,
+            auth_server_provider=self._auth_server_provider,
+            auth_settings=self.settings.auth,
+            debug=self.settings.debug,
+            additional_routes=self._additional_http_routes,
         )
 
     def mount(