ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-24 06:24:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #361 from jlowin/streamable-http

Browse source 
Streamable HTTP support
This commit is contained in:
Jeremiah Lowin 2025-05-07 22:30:49 -04:00 • committed by GitHub
commit 61a1f334ae
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 888 additions and 214 deletions
Download patch file Download diff file Expand all files Collapse all files

15
docs/deployment/authentication.mdx Normal file
View file

@ -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.

192
docs/deployment/running-server.mdx Normal file
View file

@ -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.

13
docs/docs.json
View file

@ -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",

2
docs/getting-started/quickstart.mdx
View file

@ -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.

4
docs/patterns/proxy.mdx
View file

@ -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
---

165
docs/servers/fastmcp.mdx
View file

@ -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)
# To use a different transport, e.g., Streamable HTTP:
# mcp.run(transport="streamable-http", host="127.0.0.1", port=9000)
```
This pattern is important because:
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.
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
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.
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"
)
)
```
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:
```bash
# Run a server (defaults to stdio transport)
fastmcp run my_server.py:mcp
# 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,
)
from mcp.server.auth.settings import AuthSettings #, ... other auth imports
# from your_auth_implementation import MyOAuthServerProvider # Placeholder
# 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"],
),
)
# 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.

2
src/fastmcp/cli/cli.py
View file

@ -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[

18
src/fastmcp/client/transports.py
View file

@ -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"):

227
src/fastmcp/server/http.py
View file

@ -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] = []
# Get auth middleware and routes
middleware, auth_routes, required_scopes = setup_auth_middleware_and_routes(
auth_server_provider, auth_settings
)
# Handle authentication configuration
# Initialize routes with auth routes
routes: list[Route | Mount] = auth_routes.copy()
# Add SSE routes with or without auth
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 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,
)
)
# Add authenticated routes
# 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)
# Create and return the Starlette app with middleware
return Starlette(debug=debug, routes=routes, middleware=middleware)
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 app with lifespan
return create_base_app(routes, middleware, debug, lifespan)

75
src/fastmcp/server/server.py
View file

@ -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,

241
src/fastmcp/server/streamable_http_manager.py Normal file
View file

@ -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)

7
src/fastmcp/settings.py
View file

@ -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."""

12
test.py Normal file
View file

@ -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",
)

102
tests/client/test_streamable_http.py Normal file
View file

@ -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"

20
tests/server/test_http_dependencies.py
View file

@ -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)

3
tests/server/test_lifespan.py
View file

@ -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():