refactor: move streamable HTTP app creation from FastMCP to lowlevel Server (#1899)

maxisbey · web-flow · commit 34e66a3812b9 · 2026-01-19T18:48:04.000+01:00
diff --git a/docs/migration.md b/docs/migration.md
@@ -282,7 +282,29 @@ The `ClientSession.read_resource()`, `subscribe_resource()`, and `unsubscribe_re
 
 ## New Features
 
-<!-- Add new features below -->
+### `streamable_http_app()` available on lowlevel Server
+
+The `streamable_http_app()` method is now available directly on the lowlevel `Server` class, not just `FastMCP`. This allows using the streamable HTTP transport without the FastMCP wrapper.
+
+```python
+from mcp.server.lowlevel.server import Server
+
+server = Server("my-server")
+
+# Register handlers...
+@server.list_tools()
+async def list_tools():
+    return [...]
+
+# Create a Starlette app for streamable HTTP
+app = server.streamable_http_app(
+    streamable_http_path="/mcp",
+    json_response=False,
+    stateless_http=False,
+)
+```
+
+The lowlevel `Server` also now exposes a `session_manager` property to access the `StreamableHTTPSessionManager` after calling `streamable_http_app()`.
 
 ## Need Help?
 
diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py
@@ -165,7 +165,6 @@ def __init__(
         if auth_server_provider and not token_verifier:  # pragma: no cover
             self._token_verifier = ProviderTokenVerifier(auth_server_provider)
         self._custom_starlette_routes: list[Route] = []
-        self._session_manager: StreamableHTTPSessionManager | None = None
 
         # Set up MCP protocol handlers
         self._setup_handlers()
@@ -211,14 +210,7 @@ def session_manager(self) -> StreamableHTTPSessionManager:
         Raises:
             RuntimeError: If called before streamable_http_app() has been called.
         """
-        if self._session_manager is None:  # pragma: no cover
-            raise RuntimeError(
-                "Session manager can only be accessed after"
-                "calling streamable_http_app()."
-                "The session manager is created lazily"
-                "to avoid unnecessary initialization."
-            )
-        return self._session_manager  # pragma: no cover
+        return self._mcp_server.session_manager  # pragma: no cover
 
     @overload
     def run(self, transport: Literal["stdio"] = ...) -> None: ...
@@ -929,107 +921,19 @@ def streamable_http_app(
         host: str = "127.0.0.1",
     ) -> Starlette:
         """Return an instance of the StreamableHTTP server app."""
-        from starlette.middleware import Middleware
-
-        # Auto-enable DNS rebinding protection for localhost (IPv4 and IPv6)
-        if transport_security is None and host in ("127.0.0.1", "localhost", "::1"):
-            transport_security = TransportSecuritySettings(
-                enable_dns_rebinding_protection=True,
-                allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
-                allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
-            )
-
-        # Create session manager on first call (lazy initialization)
-        if self._session_manager is None:  # pragma: no branch
-            self._session_manager = StreamableHTTPSessionManager(
-                app=self._mcp_server,
-                event_store=event_store,
-                retry_interval=retry_interval,
-                json_response=json_response,
-                stateless=stateless_http,
-                security_settings=transport_security,
-            )
-
-        # Create the ASGI handler
-        streamable_http_app = StreamableHTTPASGIApp(self._session_manager)
-
-        # Create routes
-        routes: list[Route | Mount] = []
-        middleware: list[Middleware] = []
-        required_scopes: list[str] = []
-
-        # Set up auth if configured
-        if self.settings.auth:  # pragma: no cover
-            required_scopes = self.settings.auth.required_scopes or []
-
-            # Add auth middleware if token verifier is available
-            if self._token_verifier:
-                middleware = [
-                    Middleware(
-                        AuthenticationMiddleware,
-                        backend=BearerAuthBackend(self._token_verifier),
-                    ),
-                    Middleware(AuthContextMiddleware),
-                ]
-
-            # Add auth endpoints if auth server provider is configured
-            if self._auth_server_provider:
-                from mcp.server.auth.routes import create_auth_routes
-
-                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,
-                    )
-                )
-
-        # Set up routes with or without auth
-        if self._token_verifier:  # pragma: no cover
-            # Determine resource metadata URL
-            resource_metadata_url = None
-            if self.settings.auth and self.settings.auth.resource_server_url:
-                from mcp.server.auth.routes import build_resource_metadata_url
-
-                # Build compliant metadata URL for WWW-Authenticate header
-                resource_metadata_url = build_resource_metadata_url(self.settings.auth.resource_server_url)
-
-            routes.append(
-                Route(
-                    streamable_http_path,
-                    endpoint=RequireAuthMiddleware(streamable_http_app, required_scopes, resource_metadata_url),
-                )
-            )
-        else:
-            # Auth is disabled, no wrapper needed
-            routes.append(
-                Route(
-                    streamable_http_path,
-                    endpoint=streamable_http_app,
-                )
-            )
-
-        # Add protected resource metadata endpoint if configured as RS
-        if self.settings.auth and self.settings.auth.resource_server_url:  # pragma: no cover
-            from mcp.server.auth.routes import create_protected_resource_routes
-
-            routes.extend(
-                create_protected_resource_routes(
-                    resource_url=self.settings.auth.resource_server_url,
-                    authorization_servers=[self.settings.auth.issuer_url],
-                    scopes_supported=self.settings.auth.required_scopes,
-                )
-            )
-
-        routes.extend(self._custom_starlette_routes)
-
-        return Starlette(
+        return self._mcp_server.streamable_http_app(
+            streamable_http_path=streamable_http_path,
+            json_response=json_response,
+            stateless_http=stateless_http,
+            event_store=event_store,
+            retry_interval=retry_interval,
+            transport_security=transport_security,
+            host=host,
+            auth=self.settings.auth,
+            token_verifier=self._token_verifier,
+            auth_server_provider=self._auth_server_provider,
+            custom_starlette_routes=self._custom_starlette_routes,
             debug=self.settings.debug,
-            routes=routes,
-            middleware=middleware,
-            lifespan=lambda app: self.session_manager.run(),
         )
 
     async def list_prompts(self) -> list[MCPPrompt]:
@@ -1071,16 +975,6 @@ async def get_prompt(self, name: str, arguments: dict[str, Any] | None = None) -
             raise ValueError(str(e))
 
 
-class StreamableHTTPASGIApp:
-    """ASGI application for Streamable HTTP server transport."""
-
-    def __init__(self, session_manager: StreamableHTTPSessionManager):
-        self.session_manager = session_manager
-
-    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:  # pragma: no cover
-        await self.session_manager.handle_request(scope, receive, send)
-
-
 class Context(BaseModel, Generic[ServerSessionT, LifespanContextT, RequestT]):
     """Context object providing access to MCP capabilities.
 
diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py
@@ -79,15 +79,27 @@ async def main():
 import anyio
 import jsonschema
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+from starlette.applications import Starlette
+from starlette.middleware import Middleware
+from starlette.middleware.authentication import AuthenticationMiddleware
+from starlette.routing import Mount, Route
 from typing_extensions import TypeVar
 
 import mcp.types as types
+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, TokenVerifier
+from mcp.server.auth.routes import build_resource_metadata_url, create_auth_routes, create_protected_resource_routes
+from mcp.server.auth.settings import AuthSettings
 from mcp.server.experimental.request_context import Experimental
 from mcp.server.lowlevel.experimental import ExperimentalHandlers
 from mcp.server.lowlevel.func_inspection import create_call_wrapper
 from mcp.server.lowlevel.helper_types import ReadResourceContents
 from mcp.server.models import InitializationOptions
 from mcp.server.session import ServerSession
+from mcp.server.streamable_http import EventStore
+from mcp.server.streamable_http_manager import StreamableHTTPASGIApp, StreamableHTTPSessionManager
+from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.context import RequestContext
 from mcp.shared.exceptions import McpError, UrlElicitationRequiredError
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
@@ -162,6 +174,7 @@ def __init__(
         self.notification_handlers: dict[type, Callable[..., Awaitable[None]]] = {}
         self._tool_cache: dict[str, types.Tool] = {}
         self._experimental_handlers: ExperimentalHandlers | None = None
+        self._session_manager: StreamableHTTPSessionManager | None = None
         logger.debug("Initializing server %r", name)
 
     def create_initialization_options(
@@ -258,6 +271,20 @@ def experimental(self) -> ExperimentalHandlers:
             self._experimental_handlers = ExperimentalHandlers(self, self.request_handlers, self.notification_handlers)
         return self._experimental_handlers
 
+    @property
+    def session_manager(self) -> StreamableHTTPSessionManager:
+        """Get the StreamableHTTP session manager.
+
+        Raises:
+            RuntimeError: If called before streamable_http_app() has been called.
+        """
+        if self._session_manager is None:  # pragma: no cover
+            raise RuntimeError(
+                "Session manager can only be accessed after calling streamable_http_app(). "
+                "The session manager is created lazily to avoid unnecessary initialization."
+            )
+        return self._session_manager  # pragma: no cover
+
     def list_prompts(self):
         def decorator(
             func: Callable[[], Awaitable[list[types.Prompt]]]
@@ -791,6 +818,118 @@ async def _handle_notification(self, notify: Any):
             except Exception:  # pragma: no cover
                 logger.exception("Uncaught exception in notification handler")
 
+    def streamable_http_app(
+        self,
+        *,
+        streamable_http_path: str = "/mcp",
+        json_response: bool = False,
+        stateless_http: bool = False,
+        event_store: EventStore | None = None,
+        retry_interval: int | None = None,
+        transport_security: TransportSecuritySettings | None = None,
+        host: str = "127.0.0.1",
+        auth: AuthSettings | None = None,
+        token_verifier: TokenVerifier | None = None,
+        auth_server_provider: (OAuthAuthorizationServerProvider[Any, Any, Any] | None) = None,
+        custom_starlette_routes: list[Route] | None = None,
+        debug: bool = False,
+    ) -> Starlette:
+        """Return an instance of the StreamableHTTP server app."""
+        # Auto-enable DNS rebinding protection for localhost (IPv4 and IPv6)
+        if transport_security is None and host in ("127.0.0.1", "localhost", "::1"):
+            transport_security = TransportSecuritySettings(
+                enable_dns_rebinding_protection=True,
+                allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
+                allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
+            )
+
+        session_manager = StreamableHTTPSessionManager(
+            app=self,
+            event_store=event_store,
+            retry_interval=retry_interval,
+            json_response=json_response,
+            stateless=stateless_http,
+            security_settings=transport_security,
+        )
+        self._session_manager = session_manager
+
+        # Create the ASGI handler
+        streamable_http_app = StreamableHTTPASGIApp(session_manager)
+
+        # Create routes
+        routes: list[Route | Mount] = []
+        middleware: list[Middleware] = []
+        required_scopes: list[str] = []
+
+        # Set up auth if configured
+        if auth:  # pragma: no cover
+            required_scopes = auth.required_scopes or []
+
+            # Add auth middleware if token verifier is available
+            if token_verifier:
+                middleware = [
+                    Middleware(
+                        AuthenticationMiddleware,
+                        backend=BearerAuthBackend(token_verifier),
+                    ),
+                    Middleware(AuthContextMiddleware),
+                ]
+
+            # Add auth endpoints if auth server provider is configured
+            if auth_server_provider:
+                routes.extend(
+                    create_auth_routes(
+                        provider=auth_server_provider,
+                        issuer_url=auth.issuer_url,
+                        service_documentation_url=auth.service_documentation_url,
+                        client_registration_options=auth.client_registration_options,
+                        revocation_options=auth.revocation_options,
+                    )
+                )
+
+        # Set up routes with or without auth
+        if token_verifier:  # pragma: no cover
+            # Determine resource metadata URL
+            resource_metadata_url = None
+            if auth and auth.resource_server_url:
+                # Build compliant metadata URL for WWW-Authenticate header
+                resource_metadata_url = build_resource_metadata_url(auth.resource_server_url)
+
+            routes.append(
+                Route(
+                    streamable_http_path,
+                    endpoint=RequireAuthMiddleware(streamable_http_app, required_scopes, resource_metadata_url),
+                )
+            )
+        else:
+            # Auth is disabled, no wrapper needed
+            routes.append(
+                Route(
+                    streamable_http_path,
+                    endpoint=streamable_http_app,
+                )
+            )
+
+        # Add protected resource metadata endpoint if configured as RS
+        if auth and auth.resource_server_url:  # pragma: no cover
+            routes.extend(
+                create_protected_resource_routes(
+                    resource_url=auth.resource_server_url,
+                    authorization_servers=[auth.issuer_url],
+                    scopes_supported=auth.required_scopes,
+                )
+            )
+
+        if custom_starlette_routes:  # pragma: no cover
+            routes.extend(custom_starlette_routes)
+
+        return Starlette(
+            debug=debug,
+            routes=routes,
+            middleware=middleware,
+            lifespan=lambda app: session_manager.run(),
+        )
+
 
 async def _ping_handler(request: types.PingRequest) -> types.ServerResult:
     return types.EmptyResult()
diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py
