Implement SSE priming events for resumability (SEP-1699)

Server now sends a priming event (SSE event with ID but empty data) at the
start of POST SSE streams when an EventStore is configured. This enables
clients to reconnect with Last-Event-ID even if the server closes the
connection before sending any actual data.

Changes:
- EventStore.store_event now accepts JSONRPCMessage | None (None for priming)
- Server sends priming event before processing messages in sse_writer
- Client calls resumption callback for empty-data events that have an ID

Author: felixweinberger

src/mcp/client/streamable_http.py

@@ -160,8 +160,11 @@ async def _handle_sse_event(
     ) -> bool:
         """Handle an SSE event, returning True if the response is complete."""
         if sse.event == "message":
-            # Skip empty data (keep-alive pings)
+            # Handle priming events (empty data with ID) for resumability
             if not sse.data:
+                # Call resumption callback for priming events that have an ID
+                if sse.id and resumption_callback:
+                    await resumption_callback(sse.id)
                 return False
             try:
                 message = JSONRPCMessage.model_validate_json(sse.data)

src/mcp/server/streamable_http.py

@@ -87,13 +87,13 @@ class EventStore(ABC):
     """
 
     @abstractmethod
-    async def store_event(self, stream_id: StreamId, message: JSONRPCMessage) -> EventId:
+    async def store_event(self, stream_id: StreamId, message: JSONRPCMessage | None) -> EventId:
         """
         Stores an event for later retrieval.
 
         Args:
             stream_id: ID of the stream the event belongs to
-            message: The JSON-RPC message to store
+            message: The JSON-RPC message to store, or None for priming events
 
         Returns:
             The generated event ID for the stored event
@@ -489,6 +489,17 @@ async def sse_writer():
                     # Get the request ID from the incoming request message
                     try:
                         async with sse_stream_writer, request_stream_reader:
+                            # Send priming event if event_store is configured
+                            # This sends an event with ID but empty data, enabling
+                            # the client to reconnect with Last-Event-ID if needed
+                            if self._event_store:
+                                priming_event_id = await self._event_store.store_event(
+                                    request_id,
+                                    None,  # Priming event has no payload
+                                )
+                                priming_event = {"id": priming_event_id, "data": ""}
+                                await sse_stream_writer.send(priming_event)
+
                             # Process messages from the request-specific stream
                             async for event_message in request_stream_reader:
                                 # Build the event data

tests/shared/test_streamable_http.py

@@ -77,10 +77,10 @@ class SimpleEventStore(EventStore):
     """Simple in-memory event store for testing."""
 
     def __init__(self):
-        self._events: list[tuple[StreamId, EventId, types.JSONRPCMessage]] = []
+        self._events: list[tuple[StreamId, EventId, types.JSONRPCMessage | None]] = []
         self._event_id_counter = 0
 
-    async def store_event(self, stream_id: StreamId, message: types.JSONRPCMessage) -> EventId:  # pragma: no cover
+    async def store_event(self, stream_id: StreamId, message: types.JSONRPCMessage | None) -> EventId:  # pragma: no cover
         """Store an event and return its ID."""
         self._event_id_counter += 1
         event_id = str(self._event_id_counter)