refactor(examples): migrate all HTTP examples to streamable_http_app()

docs/experimental/tasks-server.md

@@ -408,16 +408,10 @@ For custom error messages, call `task.fail()` before raising.
 For web applications, use the Streamable HTTP transport:
 
 ```python
-from collections.abc import AsyncIterator
-from contextlib import asynccontextmanager
-
 import uvicorn
-from starlette.applications import Starlette
-from starlette.routing import Mount
 
 from mcp.server import Server
 from mcp.server.experimental.task_context import ServerTaskContext
-from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.types import (
     CallToolResult, CreateTaskResult, TextContent, Tool, ToolExecution, TASK_REQUIRED,
 )
@@ -462,22 +456,8 @@ async def handle_tool(name: str, arguments: dict) -> CallToolResult | CreateTask
     return CallToolResult(content=[TextContent(type="text", text=f"Unknown: {name}")], isError=True)
 
 
-def create_app():
-    session_manager = StreamableHTTPSessionManager(app=server)
-
-    @asynccontextmanager
-    async def lifespan(app: Starlette) -> AsyncIterator[None]:
-        async with session_manager.run():
-            yield
-
-    return Starlette(
-        routes=[Mount("/mcp", app=session_manager.handle_request)],
-        lifespan=lifespan,
-    )
-
-
 if __name__ == "__main__":
-    uvicorn.run(create_app(), host="127.0.0.1", port=8000)
+    uvicorn.run(server.streamable_http_app(), host="127.0.0.1", port=8000)
 ```
 
 ## Testing Task Servers

examples/servers/simple-pagination/README.md

@@ -4,14 +4,14 @@ A simple MCP server demonstrating pagination for tools, resources, and prompts u
 
 ## Usage
 
-Start the server using either stdio (default) or SSE transport:
+Start the server using either stdio (default) or Streamable HTTP transport:
 
 ```bash
 # Using stdio transport (default)
 uv run mcp-simple-pagination
 
-# Using SSE transport on custom port
-uv run mcp-simple-pagination --transport sse --port 8000
+# Using Streamable HTTP transport on custom port
+uv run mcp-simple-pagination --transport streamable-http --port 8000
 ```
 
 The server exposes:

examples/servers/simple-pagination/mcp_simple_pagination/server.py

@@ -10,7 +10,6 @@
 import click
 from mcp import types
 from mcp.server import Server, ServerRequestContext
-from starlette.requests import Request
 
 T = TypeVar("T")
 
@@ -143,10 +142,10 @@ async def handle_get_prompt(ctx: ServerRequestContext, params: types.GetPromptRe
 
 
 @click.command()
-@click.option("--port", default=8000, help="Port to listen on for SSE")
+@click.option("--port", default=8000, help="Port to listen on for HTTP")
 @click.option(
     "--transport",
-    type=click.Choice(["stdio", "sse"]),
+    type=click.Choice(["stdio", "streamable-http"]),
     default="stdio",
     help="Transport type",
 )
@@ -161,30 +160,10 @@ def main(port: int, transport: str) -> int:
         on_get_prompt=handle_get_prompt,
     )
 
-    if transport == "sse":
-        from mcp.server.sse import SseServerTransport
-        from starlette.applications import Starlette
-        from starlette.responses import Response
-        from starlette.routing import Mount, Route
-
-        sse = SseServerTransport("/messages/")
-
-        async def handle_sse(request: Request):
-            async with sse.connect_sse(request.scope, request.receive, request._send) as streams:  # type: ignore[reportPrivateUsage]
-                await app.run(streams[0], streams[1], app.create_initialization_options())
-            return Response()
-
-        starlette_app = Starlette(
-            debug=True,
-            routes=[
-                Route("/sse", endpoint=handle_sse, methods=["GET"]),
-                Mount("/messages/", app=sse.handle_post_message),
-            ],
-        )
-
+    if transport == "streamable-http":
         import uvicorn
 
-        uvicorn.run(starlette_app, host="127.0.0.1", port=port)
+        uvicorn.run(app.streamable_http_app(), host="127.0.0.1", port=port)
     else:
         from mcp.server.stdio import stdio_server
 

examples/servers/simple-prompt/README.md

@@ -4,14 +4,14 @@ A simple MCP server that exposes a customizable prompt template with optional co
 
 ## Usage
 
-Start the server using either stdio (default) or SSE transport:
+Start the server using either stdio (default) or Streamable HTTP transport:
 
 ```bash
 # Using stdio transport (default)
 uv run mcp-simple-prompt
 
-# Using SSE transport on custom port
-uv run mcp-simple-prompt --transport sse --port 8000
+# Using Streamable HTTP transport on custom port
+uv run mcp-simple-prompt --transport streamable-http --port 8000
 ```
 
 The server exposes a prompt named "simple" that accepts two optional arguments:

examples/servers/simple-prompt/mcp_simple_prompt/server.py

@@ -2,7 +2,6 @@
 import click
 from mcp import types
 from mcp.server import Server, ServerRequestContext
-from starlette.requests import Request
 
 
 def create_messages(context: str | None = None, topic: str | None = None) -> list[types.PromptMessage]:
@@ -69,10 +68,10 @@ async def handle_get_prompt(ctx: ServerRequestContext, params: types.GetPromptRe
 
 
 @click.command()
-@click.option("--port", default=8000, help="Port to listen on for SSE")
+@click.option("--port", default=8000, help="Port to listen on for HTTP")
 @click.option(
     "--transport",
-    type=click.Choice(["stdio", "sse"]),
+    type=click.Choice(["stdio", "streamable-http"]),
     default="stdio",
     help="Transport type",
 )
@@ -83,30 +82,10 @@ def main(port: int, transport: str) -> int:
         on_get_prompt=handle_get_prompt,
     )
 
-    if transport == "sse":
-        from mcp.server.sse import SseServerTransport
-        from starlette.applications import Starlette
-        from starlette.responses import Response
-        from starlette.routing import Mount, Route
-
-        sse = SseServerTransport("/messages/")
-
-        async def handle_sse(request: Request):
-            async with sse.connect_sse(request.scope, request.receive, request._send) as streams:  # type: ignore[reportPrivateUsage]
-                await app.run(streams[0], streams[1], app.create_initialization_options())
-            return Response()
-
-        starlette_app = Starlette(
-            debug=True,
-            routes=[
-                Route("/sse", endpoint=handle_sse),
-                Mount("/messages/", app=sse.handle_post_message),
-            ],
-        )
-
+    if transport == "streamable-http":
         import uvicorn
 
-        uvicorn.run(starlette_app, host="127.0.0.1", port=port)
+        uvicorn.run(app.streamable_http_app(), host="127.0.0.1", port=port)
     else:
         from mcp.server.stdio import stdio_server
 

examples/servers/simple-resource/README.md

@@ -4,14 +4,14 @@ A simple MCP server that exposes sample text files as resources.
 
 ## Usage
 
-Start the server using either stdio (default) or SSE transport:
+Start the server using either stdio (default) or Streamable HTTP transport:
 
 ```bash
 # Using stdio transport (default)
 uv run mcp-simple-resource
 
-# Using SSE transport on custom port
-uv run mcp-simple-resource --transport sse --port 8000
+# Using Streamable HTTP transport on custom port
+uv run mcp-simple-resource --transport streamable-http --port 8000
 ```
 
 The server exposes some basic text file resources that can be read by clients.

examples/servers/simple-resource/mcp_simple_resource/server.py

@@ -4,7 +4,6 @@
 import click
 from mcp import types
 from mcp.server import Server, ServerRequestContext
-from starlette.requests import Request
 
 SAMPLE_RESOURCES = {
     "greeting": {
@@ -62,10 +61,10 @@ async def handle_read_resource(
 
 
 @click.command()
-@click.option("--port", default=8000, help="Port to listen on for SSE")
+@click.option("--port", default=8000, help="Port to listen on for HTTP")
 @click.option(
     "--transport",
-    type=click.Choice(["stdio", "sse"]),
+    type=click.Choice(["stdio", "streamable-http"]),
     default="stdio",
     help="Transport type",
 )
@@ -76,30 +75,10 @@ def main(port: int, transport: str) -> int:
         on_read_resource=handle_read_resource,
     )
 
-    if transport == "sse":
-        from mcp.server.sse import SseServerTransport
-        from starlette.applications import Starlette
-        from starlette.responses import Response
-        from starlette.routing import Mount, Route
-
-        sse = SseServerTransport("/messages/")
-
-        async def handle_sse(request: Request):
-            async with sse.connect_sse(request.scope, request.receive, request._send) as streams:  # type: ignore[reportPrivateUsage]
-                await app.run(streams[0], streams[1], app.create_initialization_options())
-            return Response()
-
-        starlette_app = Starlette(
-            debug=True,
-            routes=[
-                Route("/sse", endpoint=handle_sse, methods=["GET"]),
-                Mount("/messages/", app=sse.handle_post_message),
-            ],
-        )
-
+    if transport == "streamable-http":
         import uvicorn
 
-        uvicorn.run(starlette_app, host="127.0.0.1", port=port)
+        uvicorn.run(app.streamable_http_app(), host="127.0.0.1", port=port)
     else:
         from mcp.server.stdio import stdio_server
 

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

@@ -1,17 +1,11 @@
-import contextlib
 import logging
-from collections.abc import AsyncIterator
 
 import anyio
 import click
 import uvicorn
 from mcp import types
 from mcp.server import Server, ServerRequestContext
-from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
-from starlette.applications import Starlette
 from starlette.middleware.cors import CORSMiddleware
-from starlette.routing import Mount
-from starlette.types import Receive, Scope, Send
 
 logger = logging.getLogger(__name__)
 
@@ -104,32 +98,10 @@ def main(
         on_call_tool=handle_call_tool,
     )
 
-    # Create the session manager with true stateless mode
-    session_manager = StreamableHTTPSessionManager(
-        app=app,
-        event_store=None,
+    starlette_app = app.streamable_http_app(
+        stateless_http=True,
         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(
         debug=True,
-        routes=[Mount("/mcp", app=handle_streamable_http)],
-        lifespan=lifespan,
     )
 
     # Wrap ASGI application with CORS middleware to expose Mcp-Session-Id header

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

@@ -1,16 +1,11 @@
-import contextlib
 import logging
-from collections.abc import AsyncIterator
 
 import anyio
 import click
+import uvicorn
 from mcp import types
 from mcp.server import Server, ServerRequestContext
-from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
-from starlette.applications import Starlette
 from starlette.middleware.cors import CORSMiddleware
-from starlette.routing import Mount
-from starlette.types import Receive, Scope, Send
 
 from .event_store import InMemoryEventStore
 
@@ -127,34 +122,10 @@ def main(
     # 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
+    starlette_app = app.streamable_http_app(
+        event_store=event_store,
         json_response=json_response,
-    )
-
-    # ASGI handler for streamable HTTP connections
-    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(
         debug=True,
-        routes=[
-            Mount("/mcp", app=handle_streamable_http),
-        ],
-        lifespan=lifespan,
     )
 
     # Wrap ASGI application with CORS middleware to expose Mcp-Session-Id header
@@ -166,8 +137,6 @@ async def lifespan(app: Starlette) -> AsyncIterator[None]:
         expose_headers=["Mcp-Session-Id"],
     )
 
-    import uvicorn
-
     uvicorn.run(starlette_app, host="127.0.0.1", port=port)
 
     return 0