refactor(examples): migrate task/polling examples to streamable_http_app()

simple-task, simple-task-interactive, and sse-polling-demo demonstrate
tasks and SSE polling — transport wiring is incidental. Migrate them to
Server.streamable_http_app(), which handles session manager creation,
lifespan, routing, and DNS rebinding auto-enable in one call.

simple-streamablehttp and simple-streamablehttp-stateless stay on the
low-level API since demonstrating manual StreamableHTTPSessionManager
wiring is their purpose.

Author: maxisbey

examples/servers/simple-task-interactive/mcp_simple_task_interactive/server.py

@@ -6,19 +6,13 @@
 - ServerTaskContext.elicit() and ServerTaskContext.create_message() queue requests properly
 """
 
-from collections.abc import AsyncIterator
-from contextlib import asynccontextmanager
 from typing import Any
 
 import click
 import uvicorn
 from mcp import types
 from mcp.server import Server, ServerRequestContext
 from mcp.server.experimental.task_context import ServerTaskContext
-from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
-from mcp.server.transport_security import TransportSecuritySettings
-from starlette.applications import Starlette
-from starlette.routing import Mount
 
 
 async def handle_list_tools(
@@ -135,29 +129,10 @@ async def handle_call_tool(
 server.experimental.enable_tasks()
 
 
-def create_app(session_manager: StreamableHTTPSessionManager) -> Starlette:
-    @asynccontextmanager
-    async def app_lifespan(app: Starlette) -> AsyncIterator[None]:
-        async with session_manager.run():
-            yield
-
-    return Starlette(
-        routes=[Mount("/mcp", app=session_manager.handle_request)],
-        lifespan=app_lifespan,
-    )
-
-
 @click.command()
 @click.option("--port", default=8000, help="Port to listen on")
 def main(port: int) -> int:
-    session_manager = StreamableHTTPSessionManager(
-        app=server,
-        security_settings=TransportSecuritySettings(
-            allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
-            allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
-        ),
-    )
-    starlette_app = create_app(session_manager)
+    starlette_app = server.streamable_http_app()
     print(f"Starting server on http://localhost:{port}/mcp")
     uvicorn.run(starlette_app, host="127.0.0.1", port=port)
     return 0

examples/servers/simple-task/mcp_simple_task/server.py

@@ -1,18 +1,11 @@
 """Simple task server demonstrating MCP tasks over streamable HTTP."""
 
-from collections.abc import AsyncIterator
-from contextlib import asynccontextmanager
-
 import anyio
 import click
 import uvicorn
 from mcp import types
 from mcp.server import Server, ServerRequestContext
 from mcp.server.experimental.task_context import ServerTaskContext
-from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
-from mcp.server.transport_security import TransportSecuritySettings
-from starlette.applications import Starlette
-from starlette.routing import Mount
 
 
 async def handle_list_tools(
@@ -70,23 +63,7 @@ async def work(task: ServerTaskContext) -> types.CallToolResult:
 @click.command()
 @click.option("--port", default=8000, help="Port to listen on")
 def main(port: int) -> int:
-    session_manager = StreamableHTTPSessionManager(
-        app=server,
-        security_settings=TransportSecuritySettings(
-            allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
-            allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
-        ),
-    )
-
-    @asynccontextmanager
-    async def app_lifespan(app: Starlette) -> AsyncIterator[None]:
-        async with session_manager.run():
-            yield
-
-    starlette_app = Starlette(
-        routes=[Mount("/mcp", app=session_manager.handle_request)],
-        lifespan=app_lifespan,
-    )
+    starlette_app = server.streamable_http_app()
 
     print(f"Starting server on http://localhost:{port}/mcp")
     uvicorn.run(starlette_app, host="127.0.0.1", port=port)

examples/servers/sse-polling-demo/mcp_sse_polling_demo/server.py

@@ -12,19 +12,13 @@
     uv run mcp-sse-polling-demo --port 3000
 """
 
-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 mcp.server.transport_security import TransportSecuritySettings
-from starlette.applications import Starlette
-from starlette.routing import Mount
-from starlette.types import Receive, Scope, Send
 
 from .event_store import InMemoryEventStore
 
@@ -150,41 +144,14 @@ def main(port: int, log_level: str, retry_interval: int) -> int:
         on_call_tool=handle_call_tool,
     )
 
-    # Create event store for resumability
-    event_store = InMemoryEventStore()
-
-    # Create session manager with event store and retry interval
-    session_manager = StreamableHTTPSessionManager(
-        app=app,
-        event_store=event_store,
+    starlette_app = app.streamable_http_app(
+        event_store=InMemoryEventStore(),
         retry_interval=retry_interval,
-        security_settings=TransportSecuritySettings(
-            allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
-            allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
-        ),
-    )
-
-    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(starlette_app: Starlette) -> AsyncIterator[None]:
-        async with session_manager.run():
-            logger.info(f"SSE Polling Demo server started on port {port}")
-            logger.info("Try: POST /mcp with tools/call for 'process_batch'")
-            yield
-            logger.info("Server shutting down...")
-
-    starlette_app = Starlette(
         debug=True,
-        routes=[
-            Mount("/mcp", app=handle_streamable_http),
-        ],
-        lifespan=lifespan,
     )
 
-    import uvicorn
-
+    logger.info(f"SSE Polling Demo server starting on port {port}")
+    logger.info("Try: POST /mcp with tools/call for 'process_batch'")
     uvicorn.run(starlette_app, host="127.0.0.1", port=port)
     return 0