Add SSE polling demo server and client examples (SEP-1699)

Demonstrates the SSE polling pattern with close_sse_stream():
- Server: process_batch tool that checkpoints periodically
- Client: auto-reconnects transparently with Last-Event-ID
- Shows priming events, retry interval, and event replay

Author: felixweinberger

examples/clients/sse-polling-client/README.md

@@ -0,0 +1,30 @@
+# MCP SSE Polling Demo Client
+
+Demonstrates client-side auto-reconnect for the SSE polling pattern (SEP-1699).
+
+## Features
+
+- Connects to SSE polling demo server
+- Automatically reconnects when server closes SSE stream
+- Resumes from Last-Event-ID to avoid missing messages
+- Respects server-provided retry interval
+
+## Usage
+
+```bash
+# First start the server:
+uv run mcp-sse-polling-demo --port 3000
+
+# Then run this client:
+uv run mcp-sse-polling-client --url http://localhost:3000/mcp
+
+# Custom options:
+uv run mcp-sse-polling-client --url http://localhost:3000/mcp --items 20 --checkpoint-every 5
+```
+
+## Options
+
+- `--url`: Server URL (default: http://localhost:3000/mcp)
+- `--items`: Number of items to process (default: 10)
+- `--checkpoint-every`: Checkpoint interval (default: 3)
+- `--log-level`: Logging level (default: DEBUG)

examples/clients/sse-polling-client/mcp_sse_polling_client/__init__.py

@@ -0,0 +1 @@
+"""SSE Polling Demo Client - demonstrates auto-reconnect for long-running tasks."""

examples/clients/sse-polling-client/mcp_sse_polling_client/main.py

@@ -0,0 +1,104 @@
+"""
+SSE Polling Demo Client
+
+Demonstrates the client-side auto-reconnect for SSE polling pattern.
+
+This client connects to the SSE Polling Demo server and calls process_batch,
+which triggers periodic server-side stream closes. The client automatically
+reconnects using Last-Event-ID and resumes receiving messages.
+
+Run with:
+    # First start the server:
+    uv run mcp-sse-polling-demo --port 3000
+
+    # Then run this client:
+    uv run mcp-sse-polling-client --url http://localhost:3000/mcp
+"""
+
+import asyncio
+import logging
+
+import click
+from mcp import ClientSession
+from mcp.client.streamable_http import streamablehttp_client
+
+logger = logging.getLogger(__name__)
+
+
+async def run_demo(url: str, items: int, checkpoint_every: int) -> None:
+    """Run the SSE polling demo."""
+    print(f"\n{'='*60}")
+    print("SSE Polling Demo Client")
+    print(f"{'='*60}")
+    print(f"Server URL: {url}")
+    print(f"Processing {items} items with checkpoints every {checkpoint_every}")
+    print(f"Watch DEBUG logs for reconnection messages")
+    print(f"{'='*60}\n")
+
+    async with streamablehttp_client(url) as (read_stream, write_stream, _):
+        async with ClientSession(read_stream, write_stream) as session:
+            # Initialize the connection
+            print("Initializing connection...")
+            await session.initialize()
+            print("Connected!\n")
+
+            # List available tools
+            tools = await session.list_tools()
+            print(f"Available tools: {[t.name for t in tools.tools]}\n")
+
+            # Call the process_batch tool
+            print(f"Calling process_batch(items={items}, checkpoint_every={checkpoint_every})...")
+            print("Watch for reconnections in DEBUG logs\n")
+            print("-" * 40)
+
+            result = await session.call_tool(
+                "process_batch",
+                {
+                    "items": items,
+                    "checkpoint_every": checkpoint_every,
+                },
+            )
+
+            print("-" * 40)
+            if result.content:
+                content = result.content[0]
+                text = getattr(content, "text", str(content))
+                print(f"\nResult: {text}")
+            else:
+                print("\nResult: No content")
+            print(f"{'='*60}\n")
+
+
+@click.command()
+@click.option(
+    "--url",
+    default="http://localhost:3000/mcp",
+    help="Server URL",
+)
+@click.option(
+    "--items",
+    default=10,
+    help="Number of items to process",
+)
+@click.option(
+    "--checkpoint-every",
+    default=3,
+    help="Checkpoint interval",
+)
+@click.option(
+    "--log-level",
+    default="DEBUG",
+    help="Logging level",
+)
+def main(url: str, items: int, checkpoint_every: int, log_level: str) -> None:
+    """Run the SSE Polling Demo client."""
+    logging.basicConfig(
+        level=getattr(logging, log_level.upper()),
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    )
+
+    asyncio.run(run_demo(url, items, checkpoint_every))
+
+
+if __name__ == "__main__":
+    main()

examples/clients/sse-polling-client/pyproject.toml

@@ -0,0 +1,36 @@
+[project]
+name = "mcp-sse-polling-client"
+version = "0.1.0"
+description = "Demo client for SSE polling with auto-reconnect"
+readme = "README.md"
+requires-python = ">=3.10"
+authors = [{ name = "Anthropic, PBC." }]
+keywords = ["mcp", "sse", "polling", "client"]
+license = { text = "MIT" }
+dependencies = ["click>=8.2.0", "mcp"]
+
+[project.scripts]
+mcp-sse-polling-client = "mcp_sse_polling_client.main:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["mcp_sse_polling_client"]
+
+[tool.pyright]
+include = ["mcp_sse_polling_client"]
+venvPath = "."
+venv = ".venv"
+
+[tool.ruff.lint]
+select = ["E", "F", "I"]
+ignore = []
+
+[tool.ruff]
+line-length = 120
+target-version = "py310"
+
+[dependency-groups]
+dev = ["pyright>=1.1.378", "pytest>=8.3.3", "ruff>=0.6.9"]

examples/servers/sse-polling-demo/README.md

@@ -0,0 +1,36 @@
+# MCP SSE Polling Demo Server
+
+Demonstrates the SSE polling pattern with server-initiated stream close for long-running tasks (SEP-1699).
+
+## Features
+
+- Priming events (automatic with EventStore)
+- Server-initiated stream close via `close_sse_stream()` callback
+- Client auto-reconnect with Last-Event-ID
+- Progress notifications during long-running tasks
+- Configurable retry interval
+
+## Usage
+
+```bash
+# Start server on default port
+uv run mcp-sse-polling-demo --port 3000
+
+# Custom retry interval (milliseconds)
+uv run mcp-sse-polling-demo --port 3000 --retry-interval 100
+```
+
+## Tool: process_batch
+
+Processes items with periodic checkpoints that trigger SSE stream closes:
+
+- `items`: Number of items to process (1-100, default: 10)
+- `checkpoint_every`: Close stream after this many items (1-20, default: 3)
+
+## Client
+
+Use the companion `mcp-sse-polling-client` to test:
+
+```bash
+uv run mcp-sse-polling-client --url http://localhost:3000/mcp
+```

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

@@ -0,0 +1 @@
+"""SSE Polling Demo Server - demonstrates close_sse_stream for long-running tasks."""

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

@@ -0,0 +1,6 @@
+"""Entry point for the SSE Polling Demo server."""
+
+from .server import main
+
+if __name__ == "__main__":
+    main()

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

@@ -0,0 +1,100 @@
+"""
+In-memory event store for demonstrating resumability functionality.
+
+This is a simple implementation intended for examples and testing,
+not for production use where a persistent storage solution would be more appropriate.
+"""
+
+import logging
+from collections import deque
+from dataclasses import dataclass
+from uuid import uuid4
+
+from mcp.server.streamable_http import EventCallback, EventId, EventMessage, EventStore, StreamId
+from mcp.types import JSONRPCMessage
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class EventEntry:
+    """Represents an event entry in the event store."""
+
+    event_id: EventId
+    stream_id: StreamId
+    message: JSONRPCMessage | None  # None for priming events
+
+
+class InMemoryEventStore(EventStore):
+    """
+    Simple in-memory implementation of the EventStore interface for resumability.
+    This is primarily intended for examples and testing, not for production use
+    where a persistent storage solution would be more appropriate.
+
+    This implementation keeps only the last N events per stream for memory efficiency.
+    """
+
+    def __init__(self, max_events_per_stream: int = 100):
+        """Initialize the event store.
+
+        Args:
+            max_events_per_stream: Maximum number of events to keep per stream
+        """
+        self.max_events_per_stream = max_events_per_stream
+        # for maintaining last N events per stream
+        self.streams: dict[StreamId, deque[EventEntry]] = {}
+        # event_id -> EventEntry for quick lookup
+        self.event_index: dict[EventId, EventEntry] = {}
+
+    async def store_event(self, stream_id: StreamId, message: JSONRPCMessage | None) -> EventId:
+        """Stores an event with a generated event ID.
+
+        Args:
+            stream_id: ID of the stream the event belongs to
+            message: The message to store, or None for priming events
+        """
+        event_id = str(uuid4())
+        event_entry = EventEntry(event_id=event_id, stream_id=stream_id, message=message)
+
+        # Get or create deque for this stream
+        if stream_id not in self.streams:
+            self.streams[stream_id] = deque(maxlen=self.max_events_per_stream)
+
+        # If deque is full, the oldest event will be automatically removed
+        # We need to remove it from the event_index as well
+        if len(self.streams[stream_id]) == self.max_events_per_stream:
+            oldest_event = self.streams[stream_id][0]
+            self.event_index.pop(oldest_event.event_id, None)
+
+        # Add new event
+        self.streams[stream_id].append(event_entry)
+        self.event_index[event_id] = event_entry
+
+        return event_id
+
+    async def replay_events_after(
+        self,
+        last_event_id: EventId,
+        send_callback: EventCallback,
+    ) -> StreamId | None:
+        """Replays events that occurred after the specified event ID."""
+        if last_event_id not in self.event_index:
+            logger.warning(f"Event ID {last_event_id} not found in store")
+            return None
+
+        # Get the stream and find events after the last one
+        last_event = self.event_index[last_event_id]
+        stream_id = last_event.stream_id
+        stream_events = self.streams.get(last_event.stream_id, deque())
+
+        # Events in deque are already in chronological order
+        found_last = False
+        for event in stream_events:
+            if found_last:
+                # Skip priming events (None messages) during replay
+                if event.message is not None:
+                    await send_callback(EventMessage(event.message, event.event_id))
+            elif event.event_id == last_event_id:
+                found_last = True
+
+        return stream_id