StreamableHttp - update docs

README.md

@@ -66,7 +66,7 @@ The Model Context Protocol allows applications to provide context for LLMs in a
 
 - Build MCP clients that can connect to any MCP server
 - Create MCP servers that expose resources, prompts and tools
-- Use standard transports like stdio and SSE
+- Use standard transports like stdio, SSE, and Streamable HTTP
 - Handle all MCP protocol messages and lifecycle events
 
 ## Installation
@@ -387,8 +387,66 @@ python server.py
 mcp run server.py
 ```
 
+### Streamable HTTP Transport
+
+> **Note**: Streamable HTTP transport is superseding SSE transport for production deployments.
+
+```python
+from mcp.server.fastmcp import FastMCP
+
+# Stateful server (maintains session state)
+mcp = FastMCP("StatefulServer")
+
+# Stateless server (no session persistence)
+mcp = FastMCP("StatelessServer", stateless_http=True)
+
+# Run server with streamable_http transport
+mcp.run(transport="streamable-http")
+```
+
+You can mount multiple FastMCP servers in a FastAPI application:
+
+```python
+# echo.py
+from mcp.server.fastmcp import FastMCP
+
+mcp = FastMCP(name="EchoServer", stateless_http=True)
+
+
+@mcp.tool(description="A simple echo tool")
+def echo(message: str) -> str:
+    return f"Echo: {message}"
+```
+
+```python
+# main.py
+from fastapi import FastAPI
+from routers import echo
+
+app = FastAPI()
+
+# Use the session manager's lifespan
+app = FastAPI(lifespan=lambda app: echo.mcp.session_manager.run())
+app.mount("/echo", echo.mcp.streamable_http_app())
+```
+
+For low level server with Streamable HTTP implementations, see:
+- Stateful server: [`examples/servers/simple-streamablehttp/`](examples/servers/simple-streamablehttp/)
+- Stateless server: [`examples/servers/simple-streamablehttp-stateless/`](examples/servers/simple-streamablehttp-stateless/)
+
+
+
+The streamable HTTP transport supports:
+- Stateful and stateless operation modes
+- Resumability with event stores
+- JSON or SSE response formats  
+- Better scalability for multi-node deployments
+
+
 ### Mounting to an Existing ASGI Server
 
+> **Note**: SSE transport is being superseded by streamable HTTP transport. Consider using streamable HTTP for production deployments.
+
 You can mount the SSE server to an existing ASGI server using the `sse_app` method. This allows you to integrate the SSE server with other ASGI applications.
 
 ```python
@@ -621,7 +679,7 @@ if __name__ == "__main__":
 
 ### Writing MCP Clients
 
-The SDK provides a high-level client interface for connecting to MCP servers:
+The SDK provides a high-level client interface for connecting to MCP servers using various transports:
 
 ```python
 from mcp import ClientSession, StdioServerParameters, types
@@ -685,6 +743,28 @@ if __name__ == "__main__":
     asyncio.run(run())
 ```
 
+Clients can also connect using streamable HTTP transport:
+
+```python
+from mcp.client.streamable_http import streamablehttp_client
+from mcp import ClientSession
+
+
+async def main():
+    # Connect to a streamable HTTP server
+    async with streamablehttp_client("example/mcp") as (
+        read_stream,
+        write_stream,
+        _,
+    ):
+        # Create a session using the client streams
+        async with ClientSession(read_stream, write_stream) as session:
+            # Initialize the connection
+            await session.initialize()
+            # Call a tool
+            tool_result = await session.call_tool("echo", {"message": "hello"})
+```
+
 ### MCP Primitives
 
 The MCP protocol defines three core primitives that servers can implement:

examples/servers/simple-streamablehttp-stateless/mcp_simple_streamablehttp_stateless/server.py

@@ -1,37 +1,17 @@
 import contextlib
 import logging
+from collections.abc import AsyncIterator
 
 import anyio
 import click
 import mcp.types as types
 from mcp.server.lowlevel import Server
-from mcp.server.streamableHttp import (
-    StreamableHTTPServerTransport,
-)
+from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from starlette.applications import Starlette
 from starlette.routing import Mount
+from starlette.types import Receive, Scope, Send
 
 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()
@@ -122,35 +102,28 @@ async def list_tools() -> list[types.Tool]:
             )
         ]
 
-    # 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 the session manager with true stateless mode
+    session_manager = StreamableHTTPSessionManager(
+        app=app,
+        event_store=None,
+        json_response=json_response,
+        stateless=True,
+    )
+
+    async def handle_streamable_http(
+        scope: Scope, receive: Receive, send: Send
+    ) -> None:
+        await session_manager.handle_request(scope, receive, send)
+
+    @contextlib.asynccontextmanager
+    async def lifespan(app: Starlette) -> AsyncIterator[None]:
+        """Context manager for session manager."""
+        async with session_manager.run():
+            logger.info("Application started with StreamableHTTP session manager!")
+            try:
+                yield
+            finally:
+                logger.info("Application shutting down...")
 
     # Create an ASGI application using the transport
     starlette_app = Starlette(

examples/servers/simple-streamablehttp/mcp_simple_streamablehttp/server.py

@@ -1,58 +1,22 @@
 import contextlib
 import logging
-from http import HTTPStatus
-from uuid import uuid4
+from collections.abc import AsyncIterator
 
 import anyio
 import click
 import mcp.types as types
 from mcp.server.lowlevel import Server
-from mcp.server.streamable_http import (
-    MCP_SESSION_ID_HEADER,
-    StreamableHTTPServerTransport,
-)
+from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from pydantic import AnyUrl
 from starlette.applications import Starlette
-from starlette.requests import Request
-from starlette.responses import Response
 from starlette.routing import Mount
+from starlette.types import Receive, Scope, Send
 
 from .event_store import InMemoryEventStore
 
 # Configure logging
 logger = logging.getLogger(__name__)
 
-# Global task group that will be initialized in the lifespan
-task_group = None
-
-# Event store for resumability
-# The InMemoryEventStore enables resumability support for StreamableHTTP transport.
-# It stores SSE events with unique IDs, allowing clients to:
-#   1. Receive event IDs for each SSE message
-#   2. Resume streams by sending Last-Event-ID in GET requests
-#   3. Replay missed events after reconnection
-# Note: This in-memory implementation is for demonstration ONLY.
-# For production, use a persistent storage solution.
-event_store = InMemoryEventStore()
-
-
-@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")
@@ -156,60 +120,38 @@ async def list_tools() -> list[types.Tool]:
             )
         ]
 
-    # We need to store the server instances between requests
-    server_instances = {}
-    # Lock to prevent race conditions when creating new sessions
-    session_creation_lock = anyio.Lock()
+    # Create event store for resumability
+    # The InMemoryEventStore enables resumability support for StreamableHTTP transport.
+    # It stores SSE events with unique IDs, allowing clients to:
+    #   1. Receive event IDs for each SSE message
+    #   2. Resume streams by sending Last-Event-ID in GET requests
+    #   3. Replay missed events after reconnection
+    # Note: This in-memory implementation is for demonstration ONLY.
+    # For production, use a persistent storage solution.
+    event_store = InMemoryEventStore()
+
+    # Create the session manager with our app and event store
+    session_manager = StreamableHTTPSessionManager(
+        app=app,
+        event_store=event_store,  # Enable resumability
+        json_response=json_response,
+    )
 
     # ASGI handler for streamable HTTP connections
-    async def handle_streamable_http(scope, receive, send):
-        request = Request(scope, receive)
-        request_mcp_session_id = request.headers.get(MCP_SESSION_ID_HEADER)
-        if (
-            request_mcp_session_id is not None
-            and request_mcp_session_id in server_instances
-        ):
-            transport = server_instances[request_mcp_session_id]
-            logger.debug("Session already exists, handling request directly")
-            await transport.handle_request(scope, receive, send)
-        elif request_mcp_session_id is None:
-            # try to establish new session
-            logger.debug("Creating new transport")
-            # Use lock to prevent race conditions when creating new sessions
-            async with session_creation_lock:
-                new_session_id = uuid4().hex
-                http_transport = StreamableHTTPServerTransport(
-                    mcp_session_id=new_session_id,
-                    is_json_response_enabled=json_response,
-                    event_store=event_store,  # Enable resumability
-                )
-                server_instances[http_transport.mcp_session_id] = http_transport
-                logger.info(f"Created new transport with session ID: {new_session_id}")
-
-                async def run_server(task_status=None):
-                    async with http_transport.connect() as streams:
-                        read_stream, write_stream = streams
-                        if task_status:
-                            task_status.started()
-                        await app.run(
-                            read_stream,
-                            write_stream,
-                            app.create_initialization_options(),
-                        )
-
-                if not task_group:
-                    raise RuntimeError("Task group is not initialized")
-
-                await task_group.start(run_server)
-
-                # Handle the HTTP request and return the response
-                await http_transport.handle_request(scope, receive, send)
-        else:
-            response = Response(
-                "Bad Request: No valid session ID provided",
-                status_code=HTTPStatus.BAD_REQUEST,
-            )
-            await response(scope, receive, send)
+    async def handle_streamable_http(
+        scope: Scope, receive: Receive, send: Send
+    ) -> None:
+        await session_manager.handle_request(scope, receive, send)
+
+    @contextlib.asynccontextmanager
+    async def lifespan(app: Starlette) -> AsyncIterator[None]:
+        """Context manager for managing session manager lifecycle."""
+        async with session_manager.run():
+            logger.info("Application started with StreamableHTTP session manager!")
+            try:
+                yield
+            finally:
+                logger.info("Application shutting down...")
 
     # Create an ASGI application using the transport
     starlette_app = Starlette(