StreamableHttp - stateless server support (#554)

This commit is contained in:
ihrpr
2025-05-02 13:40:21 +01:00
committed by GitHub
parent 78f0b11a09
commit 72b66a58b1
8 changed files with 305 additions and 4 deletions

View File

@@ -0,0 +1,41 @@
# MCP Simple StreamableHttp Stateless Server Example
A stateless MCP server example demonstrating the StreamableHttp transport without maintaining session state. This example is ideal for understanding how to deploy MCP servers in multi-node environments where requests can be routed to any instance.
## Features
- Uses the StreamableHTTP transport in stateless mode (mcp_session_id=None)
- Each request creates a new ephemeral connection
- No session state maintained between requests
- Task lifecycle scoped to individual requests
- Suitable for deployment in multi-node environments
## Usage
Start the server:
```bash
# Using default port 3000
uv run mcp-simple-streamablehttp-stateless
# Using custom port
uv run mcp-simple-streamablehttp-stateless --port 3000
# Custom logging level
uv run mcp-simple-streamablehttp-stateless --log-level DEBUG
# Enable JSON responses instead of SSE streams
uv run mcp-simple-streamablehttp-stateless --json-response
```
The server exposes a tool named "start-notification-stream" that accepts three arguments:
- `interval`: Time between notifications in seconds (e.g., 1.0)
- `count`: Number of notifications to send (e.g., 5)
- `caller`: Identifier string for the caller
## Client
You can connect to this server using an HTTP client. For now, only the TypeScript SDK has streamable HTTP client examples, or you can use [Inspector](https://github.com/modelcontextprotocol/inspector) for testing.

View File

@@ -0,0 +1,4 @@
from .server import main
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,168 @@
import contextlib
import logging
import anyio
import click
import mcp.types as types
from mcp.server.lowlevel import Server
from mcp.server.streamableHttp import (
StreamableHTTPServerTransport,
)
from starlette.applications import Starlette
from starlette.routing import Mount
logger = logging.getLogger(__name__)
# Global task group that will be initialized in the lifespan
task_group = None
@contextlib.asynccontextmanager
async def lifespan(app):
"""Application lifespan context manager for managing task group."""
global task_group
async with anyio.create_task_group() as tg:
task_group = tg
logger.info("Application started, task group initialized!")
try:
yield
finally:
logger.info("Application shutting down, cleaning up resources...")
if task_group:
tg.cancel_scope.cancel()
task_group = None
logger.info("Resources cleaned up successfully.")
@click.command()
@click.option("--port", default=3000, help="Port to listen on for HTTP")
@click.option(
"--log-level",
default="INFO",
help="Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)",
)
@click.option(
"--json-response",
is_flag=True,
default=False,
help="Enable JSON responses instead of SSE streams",
)
def main(
port: int,
log_level: str,
json_response: bool,
) -> int:
# Configure logging
logging.basicConfig(
level=getattr(logging, log_level.upper()),
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
app = Server("mcp-streamable-http-stateless-demo")
@app.call_tool()
async def call_tool(
name: str, arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
ctx = app.request_context
interval = arguments.get("interval", 1.0)
count = arguments.get("count", 5)
caller = arguments.get("caller", "unknown")
# Send the specified number of notifications with the given interval
for i in range(count):
await ctx.session.send_log_message(
level="info",
data=f"Notification {i+1}/{count} from caller: {caller}",
logger="notification_stream",
related_request_id=ctx.request_id,
)
if i < count - 1: # Don't wait after the last notification
await anyio.sleep(interval)
return [
types.TextContent(
type="text",
text=(
f"Sent {count} notifications with {interval}s interval"
f" for caller: {caller}"
),
)
]
@app.list_tools()
async def list_tools() -> list[types.Tool]:
return [
types.Tool(
name="start-notification-stream",
description=(
"Sends a stream of notifications with configurable count"
" and interval"
),
inputSchema={
"type": "object",
"required": ["interval", "count", "caller"],
"properties": {
"interval": {
"type": "number",
"description": "Interval between notifications in seconds",
},
"count": {
"type": "number",
"description": "Number of notifications to send",
},
"caller": {
"type": "string",
"description": (
"Identifier of the caller to include in notifications"
),
},
},
},
)
]
# ASGI handler for stateless HTTP connections
async def handle_streamable_http(scope, receive, send):
logger.debug("Creating new transport")
# Use lock to prevent race conditions when creating new sessions
http_transport = StreamableHTTPServerTransport(
mcp_session_id=None,
is_json_response_enabled=json_response,
)
async with http_transport.connect() as streams:
read_stream, write_stream = streams
if not task_group:
raise RuntimeError("Task group is not initialized")
async def run_server():
await app.run(
read_stream,
write_stream,
app.create_initialization_options(),
# Runs in standalone mode for stateless deployments
# where clients perform initialization with any node
standalone_mode=True,
)
# Start server task
task_group.start_soon(run_server)
# Handle the HTTP request and return the response
await http_transport.handle_request(scope, receive, send)
# Create an ASGI application using the transport
starlette_app = Starlette(
debug=True,
routes=[
Mount("/mcp", app=handle_streamable_http),
],
lifespan=lifespan,
)
import uvicorn
uvicorn.run(starlette_app, host="0.0.0.0", port=port)
return 0

View File

@@ -0,0 +1,36 @@
[project]
name = "mcp-simple-streamablehttp-stateless"
version = "0.1.0"
description = "A simple MCP server exposing a StreamableHttp transport in stateless mode"
readme = "README.md"
requires-python = ">=3.10"
authors = [{ name = "Anthropic, PBC." }]
keywords = ["mcp", "llm", "automation", "web", "fetch", "http", "streamable", "stateless"]
license = { text = "MIT" }
dependencies = ["anyio>=4.5", "click>=8.1.0", "httpx>=0.27", "mcp", "starlette", "uvicorn"]
[project.scripts]
mcp-simple-streamablehttp-stateless = "mcp_simple_streamablehttp_stateless.server:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["mcp_simple_streamablehttp_stateless"]
[tool.pyright]
include = ["mcp_simple_streamablehttp_stateless"]
venvPath = "."
venv = ".venv"
[tool.ruff.lint]
select = ["E", "F", "I"]
ignore = []
[tool.ruff]
line-length = 88
target-version = "py310"
[tool.uv]
dev-dependencies = ["pyright>=1.1.378", "pytest>=8.3.3", "ruff>=0.6.9"]

View File

@@ -479,11 +479,21 @@ class Server(Generic[LifespanResultT]):
# but also make tracing exceptions much easier during testing and when using
# in-process servers.
raise_exceptions: bool = False,
# When True, the server as stateless deployments where
# clients can perform initialization with any node. The client must still follow
# the initialization lifecycle, but can do so with any available node
# rather than requiring initialization for each connection.
stateless: bool = False,
):
async with AsyncExitStack() as stack:
lifespan_context = await stack.enter_async_context(self.lifespan(self))
session = await stack.enter_async_context(
ServerSession(read_stream, write_stream, initialization_options)
ServerSession(
read_stream,
write_stream,
initialization_options,
stateless=stateless,
)
)
async with anyio.create_task_group() as tg:

View File

@@ -85,11 +85,17 @@ class ServerSession(
read_stream: MemoryObjectReceiveStream[types.JSONRPCMessage | Exception],
write_stream: MemoryObjectSendStream[types.JSONRPCMessage],
init_options: InitializationOptions,
stateless: bool = False,
) -> None:
super().__init__(
read_stream, write_stream, types.ClientRequest, types.ClientNotification
)
self._initialization_state = InitializationState.NotInitialized
self._initialization_state = (
InitializationState.Initialized
if stateless
else InitializationState.NotInitialized
)
self._init_options = init_options
self._incoming_message_stream_writer, self._incoming_message_stream_reader = (
anyio.create_memory_object_stream[ServerRequestResponder](0)

40
uv.lock generated
View File

@@ -1,5 +1,4 @@
version = 1
revision = 1
requires-python = ">=3.10"
[options]
@@ -11,6 +10,7 @@ members = [
"mcp-simple-prompt",
"mcp-simple-resource",
"mcp-simple-streamablehttp",
"mcp-simple-streamablehttp-stateless",
"mcp-simple-tool",
]
@@ -547,7 +547,6 @@ requires-dist = [
{ name = "uvicorn", marker = "sys_platform != 'emscripten'", specifier = ">=0.23.1" },
{ name = "websockets", marker = "extra == 'ws'", specifier = ">=15.0.1" },
]
provides-extras = ["cli", "rich", "ws"]
[package.metadata.requires-dev]
dev = [
@@ -670,6 +669,43 @@ dev = [
{ name = "ruff", specifier = ">=0.6.9" },
]
[[package]]
name = "mcp-simple-streamablehttp-stateless"
version = "0.1.0"
source = { editable = "examples/servers/simple-streamablehttp-stateless" }
dependencies = [
{ name = "anyio" },
{ name = "click" },
{ name = "httpx" },
{ name = "mcp" },
{ name = "starlette" },
{ name = "uvicorn" },
]
[package.dev-dependencies]
dev = [
{ name = "pyright" },
{ name = "pytest" },
{ name = "ruff" },
]
[package.metadata]
requires-dist = [
{ name = "anyio", specifier = ">=4.5" },
{ name = "click", specifier = ">=8.1.0" },
{ name = "httpx", specifier = ">=0.27" },
{ name = "mcp", editable = "." },
{ name = "starlette" },
{ name = "uvicorn" },
]
[package.metadata.requires-dev]
dev = [
{ name = "pyright", specifier = ">=1.1.378" },
{ name = "pytest", specifier = ">=8.3.3" },
{ name = "ruff", specifier = ">=0.6.9" },
]
[[package]]
name = "mcp-simple-tool"
version = "0.1.0"