support for resumability - server

Author: ihrpr

examples/servers/simple-streamablehttp/README.md

@@ -9,6 +9,7 @@ A simple MCP server example demonstrating the StreamableHttp transport, which en
 - Task management with anyio task groups
 - Ability to send multiple notifications over time to the client
 - Proper resource cleanup and lifespan management
+- Resumability support via InMemoryEventStore
 
 ## Usage
 
@@ -32,6 +33,23 @@ The server exposes a tool named "start-notification-stream" that accepts three a
 - `count`: Number of notifications to send (e.g., 5)
 - `caller`: Identifier string for the caller
 
+## Resumability Support
+
+This server includes resumability support through the InMemoryEventStore. This enables clients to:
+
+- Reconnect to the server after a disconnection
+- Resume event streaming from where they left off using the Last-Event-ID header
+
+
+The server will:
+- Generate unique event IDs for each SSE message
+- Store events in memory for later replay
+- Replay missed events when a client reconnects with a Last-Event-ID header
+
+Note: The InMemoryEventStore is designed for demonstration purposes only. For production use, consider implementing a persistent storage solution.
+
+
+
 ## Client
 
-You can connect to this server using an HTTP client, for now only Typescript SDK has streamable HTTP client examples or you can use (Inspector)[https://github.com/modelcontextprotocol/inspector]
+You can connect to this server using an HTTP client, for now only Typescript SDK has streamable HTTP client examples or you can use [Inspector](https://github.com/modelcontextprotocol/inspector)

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

@@ -0,0 +1,75 @@
+"""
+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
+import time
+from collections.abc import Awaitable, Callable
+from uuid import uuid4
+
+from mcp.server.streamableHttp import EventId, EventStore, StreamId
+from mcp.types import JSONRPCMessage
+
+logger = logging.getLogger(__name__)
+
+
+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.
+    """
+
+    def __init__(self):
+        self.events: dict[
+            str, tuple[str, JSONRPCMessage, float]
+        ] = {}  # event_id -> (stream_id, message, timestamp)
+
+    async def store_event(
+        self, stream_id: StreamId, message: JSONRPCMessage
+    ) -> EventId:
+        """Stores an event with a generated event ID."""
+        event_id = str(uuid4())
+        self.events[event_id] = (stream_id, message, time.time())
+        return event_id
+
+    async def replay_events_after(
+        self,
+        last_event_id: EventId,
+        send_callback: Callable[[EventId, JSONRPCMessage], Awaitable[None]],
+    ) -> StreamId:
+        """Replays events that occurred after the specified event ID."""
+        logger.debug(f"Attempting to replay events after {last_event_id}")
+        logger.debug(f"Total events in store: {len(self.events)}")
+        logger.debug(f"Event IDs in store: {list(self.events.keys())}")
+
+        if not last_event_id or last_event_id not in self.events:
+            logger.warning(f"Event ID {last_event_id} not found in store")
+            return ""
+
+        # Get the stream ID and timestamp from the last event
+        stream_id, _, last_timestamp = self.events[last_event_id]
+
+        # Find all events for this stream after the last event
+        events_to_replay = [
+            (event_id, message)
+            for event_id, (sid, message, timestamp) in self.events.items()
+            if sid == stream_id and timestamp > last_timestamp
+        ]
+
+        # Sort by timestamp to ensure chronological order
+        events_to_replay.sort(key=lambda x: self.events[x[0]][2])
+
+        logger.debug(f"Found {len(events_to_replay)} events to replay")
+        logger.debug(
+            f"Events to replay: {[event_id for event_id, _ in events_to_replay]}"
+        )
+
+        # Send all events in order
+        for event_id, message in events_to_replay:
+            await send_callback(event_id, message)
+
+        return stream_id

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

@@ -17,12 +17,24 @@
 from starlette.responses import Response
 from starlette.routing import Mount
 
+from .event_store import InMemoryEventStore
+
 # Configure logging
 logger = logging.getLogger(__name__)
 
 # Global task group that will be initialized in the lifespan
 task_group = None
 
+# Event store for resumability
+# The InMemoryEventStore enables resumability support for StreamableHTTP transport.
+# It stores SSE events with unique IDs, allowing clients to:
+#   1. Receive event IDs for each SSE message
+#   2. Resume streams by sending Last-Event-ID in GET requests
+#   3. Replay missed events after reconnection
+# Note: This in-memory implementation is for demonstration ONLY.
+# For production, use a persistent storage solution.
+event_store = InMemoryEventStore()
+
 
 @contextlib.asynccontextmanager
 async def lifespan(app):
@@ -79,9 +91,14 @@ async def call_tool(
 
         # Send the specified number of notifications with the given interval
         for i in range(count):
+            # Include more detailed message for resumability demonstration
+            notification_msg = (
+                f"[{i+1}/{count}] Event from '{caller}' - "
+                f"Use Last-Event-ID to resume if disconnected"
+            )
             await ctx.session.send_log_message(
                 level="info",
-                data=f"Notification {i+1}/{count} from caller: {caller}",
+                data=notification_msg,
                 logger="notification_stream",
                 # Associates this notification with the original request
                 # Ensures notifications are sent to the correct response stream
@@ -90,6 +107,7 @@ async def call_tool(
                 # - nowhere (if GET request isn't supported)
                 related_request_id=ctx.request_id,
             )
+            logger.debug(f"Sent notification {i+1}/{count} for caller: {caller}")
             if i < count - 1:  # Don't wait after the last notification
                 await anyio.sleep(interval)
 
@@ -163,8 +181,10 @@ async def handle_streamable_http(scope, receive, send):
                 http_transport = StreamableHTTPServerTransport(
                     mcp_session_id=new_session_id,
                     is_json_response_enabled=json_response,
+                    event_store=event_store,  # Enable resumability
                 )
                 server_instances[http_transport.mcp_session_id] = http_transport
+                logger.info(f"Created new transport with session ID: {new_session_id}")
             async with http_transport.connect() as streams:
                 read_stream, write_stream = streams