Add SSE polling support (SEP-1699)

Implements SEP-1699 which enables servers to disconnect SSE connections
at will by sending priming events and retry fields. This enables more
efficient resource management on the server side while maintaining
resumability.

Key changes:
- Server sends priming event (empty data with event ID) on SSE stream
- Server can call close_sse_stream() to close stream while gathering events
- Client auto-reconnects using server-provided retryInterval or exponential backoff
- Added e2e integration tests and example server/client

Github-Issue:#1654

Author: felixweinberger

examples/servers/everything-server/mcp_everything_server/server.py

@@ -14,20 +14,29 @@
 from mcp.server.fastmcp import Context, FastMCP
 from mcp.server.fastmcp.prompts.base import UserMessage
 from mcp.server.session import ServerSession
+from mcp.server.streamable_http import (
+    EventCallback,
+    EventId,
+    EventMessage,
+    EventStore,
+    StreamId,
+)
 from mcp.types import (
     AudioContent,
     Completion,
     CompletionArgument,
     CompletionContext,
     EmbeddedResource,
     ImageContent,
+    JSONRPCMessage,
     PromptReference,
     ResourceTemplateReference,
     SamplingMessage,
     TextContent,
     TextResourceContents,
 )
 from pydantic import AnyUrl, BaseModel, Field
+from starlette.requests import Request
 
 logger = logging.getLogger(__name__)
 
@@ -39,8 +48,47 @@
 resource_subscriptions: set[str] = set()
 watched_resource_content = "Watched resource content"
 
+
+# Simple in-memory event store for SSE polling resumability (SEP-1699)
+class SimpleEventStore(EventStore):
+    """Simple in-memory event store for testing resumability."""
+
+    def __init__(self) -> None:
+        self._events: list[tuple[StreamId, EventId, JSONRPCMessage]] = []
+        self._event_id_counter = 0
+
+    async def store_event(self, stream_id: StreamId, message: JSONRPCMessage) -> EventId:
+        """Store an event and return its ID."""
+        self._event_id_counter += 1
+        event_id = str(self._event_id_counter)
+        self._events.append((stream_id, event_id, message))
+        return event_id
+
+    async def replay_events_after(
+        self,
+        last_event_id: EventId,
+        send_callback: EventCallback,
+    ) -> StreamId | None:
+        """Replay events after the specified ID."""
+        target_stream_id = None
+        found = False
+        for stream_id, event_id, message in self._events:
+            if event_id == last_event_id:
+                target_stream_id = stream_id
+                found = True
+                continue
+            if found and stream_id == target_stream_id:
+                await send_callback(EventMessage(message=message, event_id=event_id))
+        return target_stream_id
+
+
+# Create event store for resumability
+event_store = SimpleEventStore()
+
 mcp = FastMCP(
     name="mcp-conformance-test-server",
+    event_store=event_store,
+    sse_retry_interval=3000,  # 3 seconds
 )
 
 
@@ -257,6 +305,33 @@ async def test_elicitation_sep1330_enums(ctx: Context[ServerSession, None]) -> s
         return f"Elicitation not supported or error: {str(e)}"
 
 
+@mcp.tool()
+async def test_reconnection(ctx: Context[ServerSession, None]) -> str:
+    """Tests SSE polling via server-initiated disconnect (SEP-1699)
+
+    This tool closes the SSE stream mid-call, requiring the client to reconnect
+    with Last-Event-ID to receive the remaining events.
+    """
+    # Send notification before disconnect
+    await ctx.info("Notification before disconnect")
+
+    # Get session_id from request headers
+    request = ctx.request_context.request
+    if isinstance(request, Request):
+        session_id = request.headers.get("mcp-session-id")
+        if session_id:
+            # Trigger server-initiated SSE disconnect
+            await mcp.session_manager.close_sse_stream(session_id, ctx.request_id)
+
+    # Wait for client to reconnect
+    await asyncio.sleep(0.2)
+
+    # Send notification after disconnect (will be replayed via event store)
+    await ctx.info("Notification after disconnect")
+
+    return "Reconnection test completed successfully"
+
+
 @mcp.tool()
 def test_error_handling() -> str:
     """Tests error response handling"""

examples/snippets/clients/sse_polling_client.py

@@ -0,0 +1,103 @@
+"""
+SSE Polling Example Client
+
+Demonstrates client-side behavior during server-initiated SSE disconnect.
+
+Key features:
+- Automatic reconnection when server closes SSE stream
+- Event replay via Last-Event-ID header (handled internally by the transport)
+- Progress notifications via logging callback
+
+This client connects to the SSE polling server and calls the `long-task` tool.
+The server disconnects at 50% progress, and the client automatically reconnects
+to receive remaining progress updates.
+
+Run:
+    # First start the server:
+    uv run examples/snippets/servers/sse_polling_server.py
+
+    # Then run this client:
+    uv run examples/snippets/clients/sse_polling_client.py
+"""
+
+import asyncio
+import logging
+
+from mcp import ClientSession
+from mcp.client.streamable_http import StreamableHTTPReconnectionOptions, streamablehttp_client
+from mcp.types import LoggingMessageNotificationParams, TextContent
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+async def main() -> None:
+    print("SSE Polling Example Client")
+    print("=" * 50)
+    print()
+
+    # Track notifications received via the logging callback
+    notifications_received: list[str] = []
+
+    async def logging_callback(params: LoggingMessageNotificationParams) -> None:
+        """Called when a log message notification is received from the server."""
+        data = params.data
+        if data:
+            data_str = str(data)
+            notifications_received.append(data_str)
+            print(f"[Progress] {data_str}")
+
+    # Configure reconnection behavior
+    reconnection_options = StreamableHTTPReconnectionOptions(
+        initial_reconnection_delay=1.0,  # Start with 1 second
+        max_reconnection_delay=30.0,  # Cap at 30 seconds
+        reconnection_delay_grow_factor=1.5,  # Exponential backoff
+        max_retries=5,  # Try up to 5 times
+    )
+
+    print("[Client] Connecting to server...")
+
+    async with streamablehttp_client(
+        "http://localhost:3001/mcp",
+        reconnection_options=reconnection_options,
+    ) as (read_stream, write_stream, get_session_id):
+        # Create session with logging callback to receive progress notifications
+        async with ClientSession(
+            read_stream,
+            write_stream,
+            logging_callback=logging_callback,
+        ) as session:
+            # Initialize the session
+            await session.initialize()
+            session_id = get_session_id()
+            print(f"[Client] Connected! Session ID: {session_id}")
+
+            # List available tools
+            tools = await session.list_tools()
+            tool_names = [t.name for t in tools.tools]
+            print(f"[Client] Available tools: {tool_names}")
+            print()
+
+            # Call the long-running task
+            print("[Client] Calling long-task tool...")
+            print("[Client] The server will disconnect at 50% and we'll auto-reconnect")
+            print()
+
+            # Call the tool
+            result = await session.call_tool("long-task", {})
+
+            print()
+            print("[Client] Task completed!")
+            if result.content and isinstance(result.content[0], TextContent):
+                print(f"[Result] {result.content[0].text}")
+            else:
+                print("[Result] No content")
+            print()
+            print(f"[Summary] Received {len(notifications_received)} progress notifications")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())