Implement retry interval support for SSE polling (SEP-1699)

Server now sends the retry field in SSE priming events when retry_interval
is configured. Client respects this field and waits the specified interval
before reconnecting.

Changes:
- Server: Add retry field to priming event when retry_interval is set
- Server: Extract _send_priming_event() helper method
- Client: Track retry interval from SSE events
- Client: Wait for retry interval before reconnecting

Author: felixweinberger

src/mcp/client/streamable_http.py

@@ -330,6 +330,7 @@ async def _handle_sse_response(
     ) -> None:
         """Handle SSE response from the server."""
         last_event_id: str | None = None
+        retry_interval_ms: int | None = None
 
         try:
             event_source = EventSource(response)
@@ -338,6 +339,10 @@ async def _handle_sse_response(
                 if sse.id:
                     last_event_id = sse.id
 
+                # Track retry interval from server
+                if sse.retry is not None:
+                    retry_interval_ms = sse.retry
+
                 is_complete = await self._handle_sse_event(
                     sse,
                     ctx.read_stream_writer,
@@ -352,16 +357,21 @@ async def _handle_sse_response(
         except Exception as e:
             logger.debug(f"SSE stream ended: {e}")
 
-        # Stream ended without response - reconnect if we have priming event
+        # Stream ended without response - reconnect if we received an event with ID
         if last_event_id is not None:
-            await self._handle_reconnection(ctx, last_event_id)
+            await self._handle_reconnection(ctx, last_event_id, retry_interval_ms)
 
     async def _handle_reconnection(
         self,
         ctx: RequestContext,
         last_event_id: str,
+        retry_interval_ms: int | None = None,
     ) -> None:
         """Reconnect with Last-Event-ID to resume stream after server disconnect."""
+        # Wait for retry interval if specified by server
+        if retry_interval_ms is not None:
+            await anyio.sleep(retry_interval_ms / 1000.0)
+
         headers = self._prepare_request_headers(ctx.headers)
         headers[LAST_EVENT_ID] = last_event_id
 
@@ -383,10 +393,13 @@ async def _handle_reconnection(
 
                 # Track for potential further reconnection
                 reconnect_last_event_id: str | None = last_event_id
+                reconnect_retry_ms = retry_interval_ms
 
                 async for sse in event_source.aiter_sse():
                     if sse.id:
                         reconnect_last_event_id = sse.id
+                    if sse.retry is not None:
+                        reconnect_retry_ms = sse.retry
 
                     is_complete = await self._handle_sse_event(
                         sse,
@@ -400,11 +413,11 @@ async def _handle_reconnection(
 
                 # Stream ended again without response - reconnect again
                 if reconnect_last_event_id is not None:
-                    await self._handle_reconnection(ctx, reconnect_last_event_id)
+                    await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms)
         except Exception as e:
             logger.debug(f"Reconnection failed: {e}")
             # Try to reconnect again if we still have an event ID
-            await self._handle_reconnection(ctx, last_event_id)
+            await self._handle_reconnection(ctx, last_event_id, retry_interval_ms)
 
     async def _handle_unexpected_content_type(
         self,

src/mcp/server/streamable_http.py

@@ -15,6 +15,7 @@
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from http import HTTPStatus
+from typing import Any
 
 import anyio
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
@@ -224,6 +225,23 @@ async def close_stream_callback() -> None:
         )
         return SessionMessage(message, metadata=metadata)
 
+    async def _send_priming_event(
+        self,
+        request_id: RequestId,
+        sse_stream_writer: MemoryObjectSendStream[dict[str, Any]],
+    ) -> None:
+        """Send priming event for SSE resumability if event_store is configured."""
+        if not self._event_store:
+            return
+        priming_event_id = await self._event_store.store_event(
+            str(request_id),  # Convert RequestId to StreamId (str)
+            None,  # Priming event has no payload
+        )
+        priming_event: dict[str, str | int] = {"id": priming_event_id, "data": ""}
+        if self._retry_interval is not None:
+            priming_event["retry"] = self._retry_interval
+        await sse_stream_writer.send(priming_event)
+
     def _create_error_response(
         self,
         error_message: str,
@@ -512,16 +530,8 @@ 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)
+                            # Send priming event for SSE resumability
+                            await self._send_priming_event(request_id, sse_stream_writer)
 
                             # Process messages from the request-specific stream
                             async for event_message in request_stream_reader: