Add tests and API stubs for SSE polling support (SEP-1699)

Sets up test infrastructure and API surface for SEP-1699 SSE polling.
Tests will fail until implementation is complete.

New APIs (stubbed):
- StreamableHTTPReconnectionOptions dataclass
- Server: _create_priming_event(), close_sse_stream(), retry_interval
- Client: resume_stream(), _get_next_reconnection_delay()
- RequestContext.close_sse_stream callback

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())