Merge pull request #361 from jlowin/streamable-http

docs/deployment/authentication.mdx

@@ -0,0 +1,15 @@
+---
+title: Authentication
+sidebarTitle: Authentication
+description: Secure your FastMCP server with authentication.
+icon: lock
+---
+import { VersionBadge } from '/snippets/version-badge.mdx'
+
+<VersionBadge version="2.2.7" />
+
+This document will cover how to implement authentication for your FastMCP servers.
+
+FastMCP leverages the OAuth 2.0 support provided by the underlying Model Context Protocol (MCP) SDK.
+
+For now, refer to the [MCP Server Authentication documentation](/servers/fastmcp#authentication) for initial details and the [official MCP SDK documentation](https://modelcontextprotocol.io/specification/2025-03-26/basic/authorization) for more.

docs/deployment/running-server.mdx

@@ -0,0 +1,192 @@
+---
+title: Running Your FastMCP Server
+sidebarTitle: Running the Server
+description: Learn how to run and deploy your FastMCP server using various transport protocols like STDIO, Streamable HTTP, and SSE.
+icon: circle-play
+---
+import { VersionBadge } from '/snippets/version-badge.mdx'
+
+
+FastMCP servers can be run in different ways depending on your application's needs, from local command-line tools to persistent web services. This guide covers the primary methods for running your server, focusing on the available transport protocols: STDIO, Streamable HTTP, and SSE.
+
+## The `run()` Method
+
+The main way to run a FastMCP server from a Python script is by calling the `run()` method on a `FastMCP` instance. 
+
+<Tip>
+For maximum compatibility, it's best practice to place the `run()` call within an `if __name__ == "__main__":` block. This ensures the server starts only when the script is executed directly, not when imported as a module.
+</Tip>
+
+```python {9-10} my_server.py
+from fastmcp import FastMCP
+
+mcp = FastMCP(name="MyServer")
+
+@mcp.tool()
+def hello(name: str) -> str:
+    return f"Hello, {name}!"
+
+if __name__ == "__main__":
+    mcp.run()
+```
+You can now run this MCP server by executing `python my_server.py`. 
+
+MCP servers can be run with a variety of different transport options, depending on your application's requirements. The `run()` method can take a `transport` argument and other transport-specific keyword arguments to configure how the server operates.
+
+## Transport Options
+
+Below is a comparison of available transport options to help you choose the right one for your needs:
+
+| Transport | Use Cases | Recommendation |
+| --------- | --------- | -------------- |
+| **STDIO** | Local tools, command-line scripts, and integrations with clients like Claude Desktop | Best for local tools and when clients manage server processes |
+| **Streamable HTTP** | Web-based deployments, microservices, exposing MCP over a network | Recommended choice for new web-based deployments |
+| **SSE** | Existing web-based deployments that rely on SSE | Suitable for compatibility with SSE clients; prefer Streamable HTTP for new projects |
+
+### STDIO
+
+The STDIO transport is the default and most widely compatible option for local MCP server execution. It is ideal for local tools, command-line integrations, and clients like Claude Desktop. However, it has the disadvantage of having to run the MCP code locally, which can introduce security concerns with third-party servers.
+
+STDIO is the default transport, so you don't need to specify it when calling `run()`. However, you can specify it explicitly to make your intent clear:
+
+```python {6}
+from fastmcp import FastMCP
+
+mcp = FastMCP()
+
+if __name__ == "__main__":
+    mcp.run(transport="stdio")
+```
+
+When using Stdio transport, you will typically *not* run the server yourself as a separate process. Rather, your *clients* will spin up a new server process for each session. As such, no additional configuration is required.
+
+### Streamable HTTP
+
+<VersionBadge version="2.3.0" />
+
+Streamable HTTP is a modern, efficient transport for exposing your MCP server via HTTP. It is generally recommended over SSE for new web-based deployments.
+
+To run a server using Streamable HTTP, you can use the `run()` method with the `transport` argument set to `"streamable-http"`. This will start a Uvicorn server on the default host (`127.0.0.1`), port (`8000`), and path (`/mcp`).
+<CodeGroup>
+```python {6} server.py
+from fastmcp import FastMCP
+
+mcp = FastMCP()
+
+if __name__ == "__main__":
+    mcp.run(transport="streamable-http")
+```
+```python {5} client.py
+import asyncio
+from fastmcp import Client
+
+async def example():
+    async with Client("http://127.0.0.1:8000/mcp") as client:
+        await client.ping()
+
+if __name__ == "__main__":
+    asyncio.run(example())
+```
+</CodeGroup>
+
+To customize the host, port, path, or log level, provide appropriate keyword arguments to the `run()` method.
+
+<CodeGroup>
+```python {8-11} server.py
+from fastmcp import FastMCP
+
+mcp = FastMCP()
+
+if __name__ == "__main__":
+    mcp.run(
+        transport="streamable-http",
+        host="127.0.0.1",
+        port=4200,
+        path="/my-custom-path",
+        log_level="debug",
+    )
+```
+```python {5} client.py
+import asyncio
+from fastmcp import Client
+
+async def example():
+    async with Client("http://127.0.0.1:4200/my-custom-path") as client:
+        await client.ping()
+
+if __name__ == "__main__":
+    asyncio.run(example())
+```
+</CodeGroup>
+
+
+### SSE
+
+Server-Sent Events (SSE) is an HTTP-based protocol for server-to-client streaming. While FastMCP supports SSE, Streamable HTTP is preferred for new projects.
+
+To run a server using SSE, you can use the `run()` method with the `transport` argument set to `"sse"`. This will start a Uvicorn server on the default host (`127.0.0.1`), port (`8000`), and with default SSE path (`/sse`) and message path (`/messages/`).
+
+<CodeGroup>
+```python {6} server.py
+from fastmcp import FastMCP
+
+mcp = FastMCP()
+
+if __name__ == "__main__":
+    mcp.run(transport="sse")
+```
+```python {3,7} client.py
+import asyncio
+from fastmcp import Client
+from fastmcp.client.transports import SSETransport
+
+async def example():
+    async with Client(
+        transport=SSETransport("http://127.0.0.1:8000/sse")
+    ) as client:
+        await client.ping()
+
+if __name__ == "__main__":
+    asyncio.run(example())
+```
+</CodeGroup>
+
+<Tip>
+Notice that the client in the above example uses an explicit `SSETransport` to connect to the server. FastMCP will attempt to infer the appropriate transport from the provided configuration, but HTTP URLs are assumed to be Streamable HTTP (as of FastMCP 2.3.0).
+</Tip>
+
+To customize the host, port, or log level, provide appropriate keyword arguments to the `run()` method. You can also adjust the SSE path (which clients should connect to) and the message POST endpoint (which clients use to send subsequent messages).
+
+<CodeGroup>
+```python {8-12} server.py
+from fastmcp import FastMCP
+
+mcp = FastMCP()
+
+if __name__ == "__main__":
+    mcp.run(
+        transport="sse",
+        host="127.0.0.1",
+        port=4200,
+        log_level="debug",
+        path="/my-custom-sse-path",
+        message_path="/my-custom-message-path/",
+    )
+```
+```python {7} client.py
+import asyncio
+from fastmcp import Client
+from fastmcp.client.transports import SSETransport
+
+async def example():
+    async with Client(
+        transport=SSETransport("http://127.0.0.1:4200/my-custom-sse-path")
+    ) as client:
+        await client.ping()
+
+if __name__ == "__main__":
+    asyncio.run(example())
+```
+</CodeGroup>
+
+Your client only needs to know the host, port, and "main" path; the message path will be transmitted to it as part of the connection handshake.

docs/docs.json

@@ -49,7 +49,16 @@
                     "servers/tools",
                     "servers/resources",
                     "servers/prompts",
-                    "servers/context"
+                    "servers/context",
+                    "patterns/proxy",
+                    "patterns/composition"
+                ]
+            },
+            {
+                "group": "Deployment",
+                "pages": [
+                    "deployment/running-server",
+                    "deployment/authentication"
                 ]
             },
             {
@@ -62,8 +71,6 @@
             {
                 "group": "Patterns",
                 "pages": [
-                    "patterns/proxy",
-                    "patterns/composition",
                     "patterns/decorating-methods",
                     "patterns/http-requests",
                     "patterns/openapi",

docs/getting-started/quickstart.mdx

@@ -1,6 +1,6 @@
 ---
 title: Quickstart
-icon: rocket
+icon: rocket-launch
 ---
 
 Welcome! This guide will help you quickly set up FastMCP and run your first MCP server.

docs/patterns/proxy.mdx

@@ -1,6 +1,6 @@
 ---
-title: Proxying Servers
-sidebarTitle: Proxying
+title: Proxy Servers
+sidebarTitle: Proxy Servers
 description: Use FastMCP to act as an intermediary or change transport for other MCP servers.
 icon: arrows-retweet
 ---

docs/servers/fastmcp.mdx

@@ -1,6 +1,6 @@
 ---
 title: The FastMCP Server
-sidebarTitle: FastMCP Server
+sidebarTitle: FastMCP Servers
 description: Learn about the core FastMCP server class and how to run it.
 icon: server
 ---
@@ -97,11 +97,7 @@ See [Prompts](/servers/prompts) for detailed documentation.
 
 ## Running the Server
 
-FastMCP servers need a transport mechanism to communicate with clients. In the MCP protocol, servers typically run as separate processes that clients connect to.
-
-### The `__main__` Block Pattern
-
-The standard way to make your server executable is to include a `run()` call inside an `if __name__ == "__main__":` block:
+FastMCP servers need a transport mechanism to communicate with clients. You typically start your server by calling the `mcp.run()` method on your `FastMCP` instance, often within an `if __name__ == "__main__":` block in your main server script. This pattern ensures compatibility with various MCP clients.
 
 ```python
 # my_server.py
@@ -115,118 +111,17 @@ def greet(name: str) -> str:
     return f"Hello, {name}!"
 
 if __name__ == "__main__":
-    # This code only runs when the file is executed directly
-    
-    # Basic run with default settings (stdio transport)
+    # This runs the server, defaulting to STDIO transport
     mcp.run()
     
-    # Or with specific transport and parameters
-    # mcp.run(transport="sse", host="127.0.0.1", port=9000)
-```
-
-This pattern is important because:
-
-1. **Client Compatibility**: Standard MCP clients (like Claude Desktop) expect to execute your server file directly with `python my_server.py`
-2. **Process Isolation**: Each server runs in its own process, allowing clients to manage multiple servers independently
-3. **Import Safety**: The main block prevents the server from running when the file is imported by other code
-
-While this pattern is technically optional when using FastMCP's CLI, it's considered a best practice for maximum compatibility with all MCP clients.
-
-### Transport Options
-
-FastMCP supports two transport mechanisms:
-
-#### STDIO Transport (Default)
-
-The standard input/output (STDIO) transport is the default and most widely compatible option:
-
-```python
-# Run with stdio (default)
-mcp.run()  # or explicitly: mcp.run(transport="stdio")
-```
-
-With STDIO:
-- The client starts a new server process for each session
-- Communication happens through standard input/output streams
-- The server process terminates when the client disconnects
-- This is ideal for integrations with tools like Claude Desktop, where each conversation gets its own server instance
-
-#### SSE Transport (Server-Sent Events)
-
-For long-running servers that serve multiple clients, FastMCP supports SSE:
-
-```python
-# Run with SSE on default host/port (0.0.0.0:8000)
-mcp.run(transport="sse")
-```
-
-With SSE:
-- The server runs as a persistent web server
-- Multiple clients can connect simultaneously
-- The server stays running until explicitly terminated
-- This is ideal for remote access to services
-
-You can configure transport parameters directly when running the server:
-
-```python
-# Configure with specific parameters
-mcp.run(
-    transport="sse", 
-    host="127.0.0.1",  # Override default host
-    port=8888,         # Override default port
-    log_level="debug"  # Set logging level
-)
-
-# You can also run asynchronously with the same parameters
-import asyncio
-asyncio.run(
-    mcp.run_sse_async(
-        host="127.0.0.1", 
-        port=8888, 
-        log_level="debug"
-    )
-)
+    # To use a different transport, e.g., Streamable HTTP:
+    # mcp.run(transport="streamable-http", host="127.0.0.1", port=9000)
 ```
 
-Transport parameters passed to `run()` or `run_sse_async()` override any settings defined when creating the FastMCP instance. The most common parameters for SSE transport are:
-
-- `host`: Host to bind to (default: "0.0.0.0")
-- `port`: Port to bind to (default: 8000)
-- `log_level`: Logging level (default: "INFO")
-
-#### Advanced Transport Configuration
-
-Under the hood, FastMCP's `run()` method accepts arbitrary keyword arguments (`**transport_kwargs`) that are passed to the transport-specific run methods:
-
-```python
-# For SSE transport, kwargs are passed to run_sse_async()
-mcp.run(transport="sse", **transport_kwargs)
-
-# For stdio transport, kwargs are passed to run_stdio_async()
-mcp.run(transport="stdio", **transport_kwargs)
-```
-
-This means that any future transport-specific options will be automatically available through the same interface without requiring changes to your code.
-
-### Using the FastMCP CLI
-
-The FastMCP CLI provides a convenient way to run servers:
+FastMCP supports several transport options like STDIO (default, for local tools), Streamable HTTP (recommended for web services), and SSE (legacy web transport). The server can also be run using the FastMCP CLI.
 
-```bash
-# Run a server (defaults to stdio transport)
-fastmcp run my_server.py:mcp
+For detailed information on each transport, how to configure them (host, port, paths), and when to use which, please refer to the [**Running Your FastMCP Server**](/deployment/running-server) guide.
 
-# Explicitly specify a transport
-fastmcp run my_server.py:mcp --transport sse
-
-# Configure SSE transport with host and port
-fastmcp run my_server.py:mcp --transport sse --host 127.0.0.1 --port 8888
-
-# With log level
-fastmcp run my_server.py:mcp --transport sse --log-level DEBUG
-```
-
-The CLI can dynamically find and run FastMCP server objects in your files, but including the `if __name__ == "__main__":` block ensures compatibility with all clients.
 
 ## Composing Servers
 
@@ -289,7 +184,7 @@ print(mcp.settings.on_duplicate_tools) # Output: "error"
 
 ### Key Configuration Options
 
-- **`host`**: Host address for SSE transport (default: "0.0.0.0")
+- **`host`**: Host address for SSE transport (default: "127.0.0.1")
 - **`port`**: Port number for SSE transport (default: 8000)
 - **`log_level`**: Logging level (default: "INFO")
 - **`on_duplicate_tools`**: How to handle duplicate tool registrations
@@ -336,36 +231,24 @@ If the serializer function raises an exception, the tool will fall back to the d
 
 <VersionBadge version="2.2.7" />
 
-FastMCP inherits support for OAuth 2.0 authentication from the MCP protocol, allowing servers to protect their tools and resources behind authentication.
-
-### OAuth 2.0 Support
-
-The `mcp.server.auth` module implements an OAuth 2.0 server interface that servers can use by providing an implementation of the `OAuthServerProvider` protocol.
+FastMCP supports OAuth 2.0 authentication, allowing servers to protect their tools and resources. This is configured by providing an `auth_server_provider` and `auth` settings during `FastMCP` initialization.
 
 ```python
 from fastmcp import FastMCP
-from mcp.server.auth.settings import (
-    RevocationOptions,
-    ClientRegistrationOptions,
-    AuthSettings,
-)
-
-
-# Create a server with authentication
-mcp = FastMCP(
-    name="SecureApp",
-    auth_provider=MyOAuthServerProvider(),
-    auth=AuthSettings(
-        issuer_url="https://myapp.com",
-        revocation_options=RevocationOptions(
-            enabled=True,
-        ),
-        client_registration_options=ClientRegistrationOptions(
-            enabled=True,
-            valid_scopes=["myscope", "myotherscope"],
-            default_scopes=["myscope"],
-        ),
-        required_scopes=["myscope"],
-    ),
-)
-```
+from mcp.server.auth.settings import AuthSettings #, ... other auth imports
+# from your_auth_implementation import MyOAuthServerProvider # Placeholder
+
+# Create a server with authentication (conceptual example)
+# mcp = FastMCP(
+#     name="SecureApp",
+#     auth_server_provider=MyOAuthServerProvider(),
+#     auth=AuthSettings(
+#         issuer_url="https://myapp.com",
+#         # ... other OAuth settings ...
+#         required_scopes=["myscope"],
+#     ),
+# )
+```
+Due to the low-level nature of the current MCP SDK's auth provider interface, detailed implementation is beyond a quick example. Refer to the [MCP SDK documentation](https://modelcontextprotocol.io/specification/2025-03-26/basic/authorization) for specifics on implementing an `OAuthAuthorizationServerProvider`. FastMCP integrates with this by passing the provider and settings to the underlying MCP server.
+
+A dedicated [Authentication guide](/deployment/authentication) will cover this in more detail once higher-level abstractions are available in FastMCP.

src/fastmcp/cli/cli.py

@@ -334,7 +334,7 @@ def run(
         str | None,
         typer.Option(
             "--host",
-            help="Host to bind to when using sse transport (default: 0.0.0.0)",
+            help="Host to bind to when using sse transport (default: 127.0.0.1)",
         ),
     ] = None,
     port: Annotated[

src/fastmcp/client/transports.py

@@ -1,9 +1,11 @@
 import abc
 import contextlib
 import datetime
+import inspect
 import os
 import shutil
 import sys
+import warnings
 from collections.abc import AsyncIterator
 from pathlib import Path
 from typing import Any, TypedDict
@@ -450,6 +452,8 @@ def infer_transport(
     This function attempts to infer the correct transport type from the provided
     argument, handling various input types and converting them to the appropriate
     ClientTransport subclass.
+
+    For HTTP URLs, they are assumed to be Streamable HTTP URLs unless they end in `/sse`.
     """
     # the transport is already a ClientTransport
     if isinstance(transport, ClientTransport):
@@ -470,7 +474,19 @@ def infer_transport(
 
     # the transport is an http(s) URL
     elif isinstance(transport, AnyUrl | str) and str(transport).startswith("http"):
-        return SSETransport(url=transport)
+        if str(transport).rstrip("/").endswith("/sse"):
+            warnings.warn(
+                inspect.cleandoc(
+                    """
+                    As of FastMCP 2.3.0, HTTP URLs are inferred to use Streamable HTTP.
+                    The provided URL ends in `/sse`, so you may encounter unexpected behavior.
+                    If you intended to use SSE, please use the `SSETransport` class directly.
+                    """
+                ),
+                category=UserWarning,
+                stacklevel=2,
+            )
+        return StreamableHttpTransport(url=transport)
 
     # the transport is a websocket URL
     elif isinstance(transport, AnyUrl | str) and str(transport).startswith("ws"):

src/fastmcp/server/http.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
-from collections.abc import Generator
-from contextlib import contextmanager
+from collections.abc import AsyncGenerator, Callable, Generator
+from contextlib import asynccontextmanager, contextmanager
 from contextvars import ContextVar
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
 from mcp.server.auth.middleware.bearer_auth import (
@@ -22,10 +22,12 @@ from starlette.responses import Response
 from starlette.routing import Mount, Route
 from starlette.types import Receive, Scope, Send
 
+# This import is vendored until it is finalized in the upstream SDK
+from fastmcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from fastmcp.utilities.logging import get_logger
 
 if TYPE_CHECKING:
-    from fastmcp import FastMCP
+    from fastmcp.server.server import FastMCP
 
 logger = get_logger(__name__)
 
@@ -53,10 +55,92 @@ class RequestContextMiddleware:
         self.app = app
 
     async def __call__(self, scope, receive, send):
-        with set_http_request(Request(scope)):
+        if scope["type"] == "http":
+            with set_http_request(Request(scope)):
+                await self.app(scope, receive, send)
+        else:
             await self.app(scope, receive, send)
 
 
+def setup_auth_middleware_and_routes(
+    auth_server_provider: OAuthAuthorizationServerProvider | None,
+    auth_settings: AuthSettings | None,
+) -> tuple[list[Middleware], list[Route | Mount], list[str]]:
+    """Set up authentication middleware and routes if auth is enabled.
+
+    Args:
+        auth_server_provider: The OAuth authorization server provider
+        auth_settings: The auth settings
+
+    Returns:
+        Tuple of (middleware, auth_routes, required_scopes)
+    """
+    middleware: list[Middleware] = []
+    auth_routes: list[Route | Mount] = []
+    required_scopes: list[str] = []
+
+    if auth_server_provider:
+        if not auth_settings:
+            raise ValueError(
+                "auth_settings must be provided when auth_server_provider is specified"
+            )
+
+        middleware = [
+            Middleware(
+                AuthenticationMiddleware,
+                backend=BearerAuthBackend(provider=auth_server_provider),
+            ),
+            Middleware(AuthContextMiddleware),
+        ]
+
+        required_scopes = auth_settings.required_scopes or []
+
+        auth_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,
+            )
+        )
+
+    return middleware, auth_routes, required_scopes
+
+
+def create_base_app(
+    routes: list[Route | Mount],
+    middleware: list[Middleware],
+    debug: bool,
+    lifespan: Callable | None = None,
+) -> Starlette:
+    """Create a base Starlette app with common middleware and routes.
+
+    Args:
+        routes: List of routes to include in the app
+        middleware: List of middleware to include in the app
+        debug: Whether to enable debug mode
+        lifespan: Optional lifespan manager for the app
+
+    Returns:
+        A Starlette application
+    """
+    # Always add RequestContextMiddleware as the outermost middleware
+    middleware.append(Middleware(RequestContextMiddleware))
+
+    # Create the app
+    app_kwargs = {
+        "debug": debug,
+        "routes": routes,
+        "middleware": middleware,
+    }
+
+    if lifespan:
+        app_kwargs["lifespan"] = lifespan
+
+    return Starlette(**app_kwargs)
+
+
 def create_sse_app(
     server: FastMCP,
     message_path: str,
@@ -93,42 +177,17 @@ def create_sse_app(
             )
         return Response()
 
-    # Configure routes and middleware
-    routes: list[Route | Mount] = []
-    middleware: list[Middleware] = []
-
-    # Handle authentication configuration
-    if auth_server_provider:
-        # Ensure auth settings are provided when auth provider is present
-        if not auth_settings:
-            raise ValueError(
-                "auth_settings must be provided when auth_server_provider is specified"
-            )
-
-        # Configure auth middleware
-        middleware = [
-            Middleware(
-                AuthenticationMiddleware,
-                backend=BearerAuthBackend(provider=auth_server_provider),
-            ),
-            Middleware(AuthContextMiddleware),
-        ]
+    # Get auth middleware and routes
+    middleware, auth_routes, required_scopes = setup_auth_middleware_and_routes(
+        auth_server_provider, auth_settings
+    )
 
-        # Get required scopes for authentication
-        required_scopes = auth_settings.required_scopes or []
-
-        # Add auth routes
-        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,
-            )
-        )
+    # Initialize routes with auth routes
+    routes: list[Route | Mount] = auth_routes.copy()
 
-        # Add authenticated routes
+    # Add SSE routes with or without auth
+    if auth_server_provider:
+        # Auth is enabled, wrap endpoints with RequireAuthMiddleware
         routes.append(
             Route(
                 sse_path,
@@ -143,7 +202,7 @@ def create_sse_app(
             )
         )
     else:
-        # No authentication required
+        # No auth required
         async def sse_endpoint(request: Request) -> Response:
             return await handle_sse(request.scope, request.receive, request._send)  # type: ignore[reportPrivateUsage]
 
@@ -163,10 +222,88 @@ def create_sse_app(
 
     # Add custom routes with lowest precedence
     if additional_routes:
-        routes.extend(additional_routes)
+        routes.extend(cast(list[Route | Mount], additional_routes))
 
-    # Add RequestContextMiddleware as the outermost middleware
-    middleware.append(Middleware(RequestContextMiddleware))
+    # Create and return the app
+    return create_base_app(routes, middleware, debug)
+
+
+def create_streamable_http_app(
+    server: FastMCP,
+    streamable_http_path: str,
+    event_store: None = None,
+    auth_server_provider: OAuthAuthorizationServerProvider | None = None,
+    auth_settings: AuthSettings | None = None,
+    json_response: bool = False,
+    stateless_http: bool = False,
+    debug: bool = False,
+    additional_routes: list[Route] | list[Mount] | list[Route | Mount] | None = None,
+) -> Starlette:
+    """Return an instance of the StreamableHTTP server app.
+
+    Args:
+        server: The FastMCP server instance
+        streamable_http_path: Path for StreamableHTTP connections
+        event_store: Optional event store for session management
+        auth_server_provider: Optional auth provider
+        auth_settings: Optional auth settings
+        json_response: Whether to use JSON response format
+        stateless_http: Whether to use stateless mode (new transport per request)
+        debug: Whether to enable debug mode
+        additional_routes: Optional list of custom routes
+
+    Returns:
+        A Starlette application with StreamableHTTP support
+    """
+    # Create session manager using the provided event store
+    session_manager = StreamableHTTPSessionManager(
+        app=server._mcp_server,
+        event_store=event_store,
+        json_response=json_response,
+        stateless=stateless_http,
+    )
+
+    # Create the ASGI handler
+    async def handle_streamable_http(
+        scope: Scope, receive: Receive, send: Send
+    ) -> None:
+        await session_manager.handle_request(scope, receive, send)
+
+    # Get auth middleware and routes
+    middleware, auth_routes, required_scopes = setup_auth_middleware_and_routes(
+        auth_server_provider, auth_settings
+    )
+
+    # Initialize routes with auth routes
+    routes: list[Route | Mount] = auth_routes.copy()
+
+    # Add StreamableHTTP routes with or without auth
+    if auth_server_provider:
+        # Auth is enabled, wrap endpoint with RequireAuthMiddleware
+        routes.append(
+            Mount(
+                streamable_http_path,
+                app=RequireAuthMiddleware(handle_streamable_http, required_scopes),
+            )
+        )
+    else:
+        # No auth required
+        routes.append(
+            Mount(
+                streamable_http_path,
+                app=handle_streamable_http,
+            )
+        )
+
+    # Add custom routes with lowest precedence
+    if additional_routes:
+        routes.extend(cast(list[Route | Mount], additional_routes))
+
+    # Create a lifespan manager to start and stop the session manager
+    @asynccontextmanager
+    async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
+        async with session_manager.run():
+            yield
 
-    # Create and return the Starlette app with middleware
-    return Starlette(debug=debug, routes=routes, middleware=middleware)
+    # Create and return the app with lifespan
+    return create_base_app(routes, middleware, debug, lifespan)

src/fastmcp/server/server.py

@@ -146,6 +146,7 @@ class FastMCP(Generic[LifespanResultT]):
                 "is specified"
             )
         self._auth_server_provider = auth_server_provider
+
         self._additional_http_routes: list[Route] = []
         self.dependencies = self.settings.dependencies
 
@@ -167,30 +168,36 @@ class FastMCP(Generic[LifespanResultT]):
         return self._mcp_server.instructions
 
     async def run_async(
-        self, transport: Literal["stdio", "sse"] | None = None, **transport_kwargs: Any
+        self,
+        transport: Literal["stdio", "sse", "streamable-http"] | None = None,
+        **transport_kwargs: Any,
     ) -> None:
         """Run the FastMCP server asynchronously.
 
         Args:
-            transport: Transport protocol to use ("stdio" or "sse")
+            transport: Transport protocol to use ("stdio", "sse", or "streamable-http")
         """
         if transport is None:
             transport = "stdio"
-        if transport not in ["stdio", "sse"]:
+        if transport not in ["stdio", "sse", "streamable-http"]:
             raise ValueError(f"Unknown transport: {transport}")
 
         if transport == "stdio":
             await self.run_stdio_async(**transport_kwargs)
-        else:  # transport == "sse"
+        elif transport == "sse":
             await self.run_sse_async(**transport_kwargs)
+        else:  # transport == "streamable-http"
+            await self.run_streamable_http_async(**transport_kwargs)
 
     def run(
-        self, transport: Literal["stdio", "sse"] | None = None, **transport_kwargs: Any
+        self,
+        transport: Literal["stdio", "sse", "streamable-http"] | None = None,
+        **transport_kwargs: Any,
     ) -> None:
         """Run the FastMCP server. Note this is a synchronous function.
 
         Args:
-            transport: Transport protocol to use ("stdio" or "sse")
+            transport: Transport protocol to use ("stdio", "sse", or "streamable-http")
         """
         logger.info(f'Starting server "{self.name}"...')
 
@@ -711,6 +718,8 @@ class FastMCP(Generic[LifespanResultT]):
         host: str | None = None,
         port: int | None = None,
         log_level: str | None = None,
+        path: str | None = None,
+        message_path: str | None = None,
         uvicorn_config: dict | None = None,
     ) -> None:
         """Run the server using SSE transport."""
@@ -719,7 +728,7 @@ class FastMCP(Generic[LifespanResultT]):
         # timeout to make it possible to close immediately. see
         # https://github.com/jlowin/fastmcp/issues/296
         uvicorn_config.setdefault("timeout_graceful_shutdown", 0)
-        app = self.sse_app()
+        app = self.sse_app(path=path, message_path=message_path)
 
         config = uvicorn.Config(
             app,
@@ -731,18 +740,64 @@ class FastMCP(Generic[LifespanResultT]):
         server = uvicorn.Server(config)
         await server.serve()
 
-    def sse_app(self) -> Starlette:
+    def sse_app(
+        self,
+        path: str | None = None,
+        message_path: str | None = None,
+    ) -> Starlette:
         """Return an instance of the SSE server app."""
         return create_sse_app(
             server=self,
-            message_path=self.settings.message_path,
-            sse_path=self.settings.sse_path,
+            message_path=message_path or self.settings.message_path,
+            sse_path=path or 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 streamable_http_app(self, path: str | None = None) -> Starlette:
+        """Return an instance of the StreamableHTTP server app."""
+        from fastmcp.server.http import create_streamable_http_app
+
+        return create_streamable_http_app(
+            server=self,
+            streamable_http_path=path or self.settings.streamable_http_path,
+            event_store=None,
             auth_server_provider=self._auth_server_provider,
             auth_settings=self.settings.auth,
+            json_response=self.settings.json_response,
+            stateless_http=self.settings.stateless_http,
             debug=self.settings.debug,
             additional_routes=self._additional_http_routes,
         )
 
+    async def run_streamable_http_async(
+        self,
+        host: str | None = None,
+        port: int | None = None,
+        log_level: str | None = None,
+        path: str | None = None,
+        uvicorn_config: dict | None = None,
+    ) -> None:
+        """Run the server using StreamableHTTP transport."""
+        uvicorn_config = uvicorn_config or {}
+        uvicorn_config.setdefault("timeout_graceful_shutdown", 0)
+
+        app = self.streamable_http_app(path=path)
+
+        config = uvicorn.Config(
+            app,
+            host=host or self.settings.host,
+            port=port or self.settings.port,
+            log_level=log_level or self.settings.log_level.lower(),
+            # lifespan is required for streamable http
+            lifespan="on",
+            **uvicorn_config,
+        )
+        server = uvicorn.Server(config)
+        await server.serve()
+
     def mount(
         self,
         prefix: str,

src/fastmcp/server/streamable_http_manager.py

@@ -0,0 +1,241 @@
+"""StreamableHTTP Session Manager for MCP servers."""
+
+# follows https://github.com/modelcontextprotocol/python-sdk/blob/ihrpr/shttp/src/mcp/server/streamable_http_manager.py
+# and can be removed once that spec is finalized
+
+from __future__ import annotations
+
+import contextlib
+import logging
+from collections.abc import AsyncIterator
+from http import HTTPStatus
+from typing import Any
+from uuid import uuid4
+
+import anyio
+from anyio.abc import TaskStatus
+from mcp.server.lowlevel.server import Server as MCPServer
+from mcp.server.streamable_http import (
+    MCP_SESSION_ID_HEADER,
+    EventStore,
+    StreamableHTTPServerTransport,
+)
+from starlette.requests import Request
+from starlette.responses import Response
+from starlette.types import Receive, Scope, Send
+
+logger = logging.getLogger(__name__)
+
+
+class StreamableHTTPSessionManager:
+    """
+    Manages StreamableHTTP sessions with optional resumability via event store.
+
+    This class abstracts away the complexity of session management, event storage,
+    and request handling for StreamableHTTP transports. It handles:
+
+    1. Session tracking for clients
+    2. Resumability via an optional event store
+    3. Connection management and lifecycle
+    4. Request handling and transport setup
+
+    Args:
+        app: The MCP server instance
+        event_store: Optional event store for resumability support.
+                     If provided, enables resumable connections where clients
+                     can reconnect and receive missed events.
+                     If None, sessions are still tracked but not resumable.
+        json_response: Whether to use JSON responses instead of SSE streams
+        stateless: If True, creates a completely fresh transport for each request
+                   with no session tracking or state persistence between requests.
+
+    """
+
+    def __init__(
+        self,
+        app: MCPServer[Any],
+        event_store: EventStore | None = None,
+        json_response: bool = False,
+        stateless: bool = False,
+    ):
+        self.app = app
+        self.event_store = event_store
+        self.json_response = json_response
+        self.stateless = stateless
+
+        # Session tracking (only used if not stateless)
+        self._session_creation_lock = anyio.Lock()
+        self._server_instances: dict[str, StreamableHTTPServerTransport] = {}
+
+        # The task group will be set during lifespan
+        self._task_group = None
+
+    @contextlib.asynccontextmanager
+    async def run(self) -> AsyncIterator[None]:
+        """
+        Run the session manager with proper lifecycle management.
+
+        This creates and manages the task group for all session operations.
+
+        Use this in the lifespan context manager of your Starlette app:
+
+        @contextlib.asynccontextmanager
+        async def lifespan(app: Starlette) -> AsyncIterator[None]:
+            async with session_manager.run():
+                yield
+        """
+        async with anyio.create_task_group() as tg:
+            # Store the task group for later use
+            self._task_group = tg
+            logger.info("StreamableHTTP session manager started")
+            try:
+                yield  # Let the application run
+            finally:
+                logger.info("StreamableHTTP session manager shutting down")
+                # Cancel task group to stop all spawned tasks
+                tg.cancel_scope.cancel()
+                self._task_group = None
+                # Clear any remaining server instances
+                self._server_instances.clear()
+
+    async def handle_request(
+        self,
+        scope: Scope,
+        receive: Receive,
+        send: Send,
+    ) -> None:
+        """
+        Process ASGI request with proper session handling and transport setup.
+
+        Dispatches to the appropriate handler based on stateless mode.
+
+        Args:
+            scope: ASGI scope
+            receive: ASGI receive function
+            send: ASGI send function
+        """
+        if self._task_group is None:
+            raise RuntimeError(
+                "Task group is not initialized. Make sure to use the run()."
+            )
+
+        # Dispatch to the appropriate handler
+        if self.stateless:
+            await self._handle_stateless_request(scope, receive, send)
+        else:
+            await self._handle_stateful_request(scope, receive, send)
+
+    async def _handle_stateless_request(
+        self,
+        scope: Scope,
+        receive: Receive,
+        send: Send,
+    ) -> None:
+        """
+        Process request in stateless mode - creating a new transport for each request.
+
+        Args:
+            scope: ASGI scope
+            receive: ASGI receive function
+            send: ASGI send function
+        """
+        logger.debug("Stateless mode: Creating new transport for this request")
+        # No session ID needed in stateless mode
+        http_transport = StreamableHTTPServerTransport(
+            mcp_session_id=None,  # No session tracking in stateless mode
+            is_json_response_enabled=self.json_response,
+            event_store=None,  # No event store in stateless mode
+        )
+
+        # Start server in a new task
+        async def run_stateless_server(
+            *, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED
+        ):
+            async with http_transport.connect() as streams:
+                read_stream, write_stream = streams
+                task_status.started()
+                await self.app.run(
+                    read_stream,
+                    write_stream,
+                    self.app.create_initialization_options(),
+                    stateless=True,
+                )
+
+        # Assert task group is not None for type checking
+        assert self._task_group is not None
+        # Start the server task
+        await self._task_group.start(run_stateless_server)
+
+        # Handle the HTTP request and return the response
+        await http_transport.handle_request(scope, receive, send)
+
+    async def _handle_stateful_request(
+        self,
+        scope: Scope,
+        receive: Receive,
+        send: Send,
+    ) -> None:
+        """
+        Process request in stateful mode - maintaining session state between requests.
+
+        Args:
+            scope: ASGI scope
+            receive: ASGI receive function
+            send: ASGI send function
+        """
+        request = Request(scope, receive)
+        request_mcp_session_id = request.headers.get(MCP_SESSION_ID_HEADER)
+
+        # Existing session case
+        if (
+            request_mcp_session_id is not None
+            and request_mcp_session_id in self._server_instances
+        ):
+            transport = self._server_instances[request_mcp_session_id]
+            logger.debug("Session already exists, handling request directly")
+            await transport.handle_request(scope, receive, send)
+            return
+
+        if request_mcp_session_id is None:
+            # New session case
+            logger.debug("Creating new transport")
+            async with self._session_creation_lock:
+                new_session_id = uuid4().hex
+                http_transport = StreamableHTTPServerTransport(
+                    mcp_session_id=new_session_id,
+                    is_json_response_enabled=self.json_response,
+                    event_store=self.event_store,  # May be None (no resumability)
+                )
+
+                assert http_transport.mcp_session_id is not None
+                self._server_instances[http_transport.mcp_session_id] = http_transport
+                logger.info(f"Created new transport with session ID: {new_session_id}")
+
+                # Define the server runner
+                async def run_server(
+                    *, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED
+                ) -> None:
+                    async with http_transport.connect() as streams:
+                        read_stream, write_stream = streams
+                        task_status.started()
+                        await self.app.run(
+                            read_stream,
+                            write_stream,
+                            self.app.create_initialization_options(),
+                            stateless=False,  # Stateful mode
+                        )
+
+                # Assert task group is not None for type checking
+                assert self._task_group is not None
+                # Start the server task
+                await self._task_group.start(run_server)
+
+                # Handle the HTTP request and return the response
+                await http_transport.handle_request(scope, receive, send)
+        else:
+            # Invalid session ID
+            response = Response(
+                "Bad Request: No valid session ID provided",
+                status_code=HTTPStatus.BAD_REQUEST,
+            )
+            await response(scope, receive, send)

src/fastmcp/settings.py

@@ -61,6 +61,7 @@ class ServerSettings(BaseSettings):
     port: int = 8000
     sse_path: str = "/sse"
     message_path: str = "/messages/"
+    streamable_http_path: str = "/mcp"
     debug: bool = False
 
     # resource settings
@@ -82,6 +83,12 @@ class ServerSettings(BaseSettings):
 
     auth: AuthSettings | None = None
 
+    # StreamableHTTP settings
+    json_response: bool = False
+    stateless_http: bool = (
+        False  # If True, uses true stateless mode (new transport per request)
+    )
+
 
 class ClientSettings(BaseSettings):
     """FastMCP client settings."""

test.py

@@ -0,0 +1,12 @@
+from fastmcp import FastMCP
+
+mcp = FastMCP()
+
+if __name__ == "__main__":
+    mcp.run(
+        transport="streamable-http",
+        host="127.0.0.1",
+        port=4200,
+        path="/my-custom-path/",
+        log_level="debug",
+    )

tests/client/test_streamable_http.py

@@ -0,0 +1,102 @@
+import json
+import sys
+from collections.abc import Generator
+
+import pytest
+import uvicorn
+from mcp.types import TextResourceContents
+
+from fastmcp.client import Client
+from fastmcp.client.transports import StreamableHttpTransport
+from fastmcp.server.dependencies import get_http_request
+from fastmcp.server.server import FastMCP
+from fastmcp.utilities.tests import run_server_in_process
+
+
+def fastmcp_server():
+    """Fixture that creates a FastMCP server with tools, resources, and prompts."""
+    server = FastMCP("TestServer")
+
+    # Add a tool
+    @server.tool()
+    def greet(name: str) -> str:
+        """Greet someone by name."""
+        return f"Hello, {name}!"
+
+    # Add a second tool
+    @server.tool()
+    def add(a: int, b: int) -> int:
+        """Add two numbers together."""
+        return a + b
+
+    # Add a resource
+    @server.resource(uri="data://users")
+    async def get_users():
+        return ["Alice", "Bob", "Charlie"]
+
+    # Add a resource template
+    @server.resource(uri="data://user/{user_id}")
+    async def get_user(user_id: str):
+        return {"id": user_id, "name": f"User {user_id}", "active": True}
+
+    @server.resource(uri="request://headers")
+    async def get_headers() -> dict[str, str]:
+        request = get_http_request()
+
+        return dict(request.headers)
+
+    # Add a prompt
+    @server.prompt()
+    def welcome(name: str) -> str:
+        """Example greeting prompt."""
+        return f"Welcome to FastMCP, {name}!"
+
+    return server
+
+
+def run_server(host: str, port: int) -> None:
+    try:
+        app = fastmcp_server().streamable_http_app()
+        server = uvicorn.Server(
+            config=uvicorn.Config(
+                app=app,
+                host=host,
+                port=port,
+                log_level="error",
+                lifespan="on",
+            )
+        )
+        server.run()
+    except Exception as e:
+        print(f"Server error: {e}")
+        sys.exit(1)
+    sys.exit(0)
+
+
+@pytest.fixture(scope="module")
+def streamable_http_server() -> Generator[str, None, None]:
+    with run_server_in_process(run_server) as url:
+        yield f"{url}/mcp"
+
+
+async def test_ping(streamable_http_server: str):
+    """Test pinging the server."""
+    async with Client(
+        transport=StreamableHttpTransport(streamable_http_server)
+    ) as client:
+        result = await client.ping()
+        assert result is True
+
+
+async def test_http_headers(streamable_http_server: str):
+    """Test getting HTTP headers from the server."""
+    async with Client(
+        transport=StreamableHttpTransport(
+            streamable_http_server, headers={"X-DEMO-HEADER": "ABC"}
+        )
+    ) as client:
+        raw_result = await client.read_resource("request://headers")
+        assert isinstance(raw_result[0], TextResourceContents)
+        json_result = json.loads(raw_result[0].text)
+        assert "x-demo-header" in json_result
+        assert json_result["x-demo-header"] == "ABC"

tests/server/test_http_dependencies.py

@@ -7,7 +7,7 @@ import uvicorn
 from mcp.types import TextContent, TextResourceContents
 
 from fastmcp.client import Client
-from fastmcp.client.transports import SSETransport
+from fastmcp.client.transports import StreamableHttpTransport
 from fastmcp.server.dependencies import get_http_request
 from fastmcp.server.server import FastMCP
 from fastmcp.utilities.tests import run_server_in_process
@@ -43,9 +43,15 @@ def fastmcp_server():
 
 def run_server(host: str, port: int) -> None:
     try:
-        app = fastmcp_server().sse_app()
+        app = fastmcp_server().streamable_http_app()
         server = uvicorn.Server(
-            config=uvicorn.Config(app=app, host=host, port=port, log_level="error")
+            config=uvicorn.Config(
+                app=app,
+                host=host,
+                port=port,
+                log_level="error",
+                lifespan="on",
+            )
         )
         server.run()
     except Exception as e:
@@ -57,13 +63,13 @@ def run_server(host: str, port: int) -> None:
 @pytest.fixture(autouse=True, scope="module")
 def sse_server() -> Generator[str, None, None]:
     with run_server_in_process(run_server) as url:
-        yield f"{url}/sse"
+        yield f"{url}/mcp"
 
 
 async def test_http_headers_resource(sse_server: str):
     """Test getting HTTP headers from the server."""
     async with Client(
-        transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
+        transport=StreamableHttpTransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
     ) as client:
         raw_result = await client.read_resource("request://headers")
         assert isinstance(raw_result[0], TextResourceContents)
@@ -75,7 +81,7 @@ async def test_http_headers_resource(sse_server: str):
 async def test_http_headers_tool(sse_server: str):
     """Test getting HTTP headers from the server."""
     async with Client(
-        transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
+        transport=StreamableHttpTransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
     ) as client:
         result = await client.call_tool("get_headers_tool")
         assert isinstance(result[0], TextContent)
@@ -87,7 +93,7 @@ async def test_http_headers_tool(sse_server: str):
 async def test_http_headers_prompt(sse_server: str):
     """Test getting HTTP headers from the server."""
     async with Client(
-        transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
+        transport=StreamableHttpTransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
     ) as client:
         result = await client.get_prompt("get_headers_prompt")
         assert isinstance(result.messages[0].content, TextContent)

tests/server/test_lifespan.py

@@ -5,7 +5,6 @@ from contextlib import asynccontextmanager
 
 import anyio
 import pytest
-from mcp.server.fastmcp import Context, FastMCP
 from mcp.server.lowlevel.server import NotificationOptions, Server
 from mcp.server.models import InitializationOptions
 from mcp.shared.message import SessionMessage
@@ -19,6 +18,8 @@ from mcp.types import (
 )
 from pydantic import TypeAdapter
 
+from fastmcp import Context, FastMCP
+
 
 @pytest.mark.anyio
 async def test_lowlevel_server_lifespan():