Implement SSE polling support (SEP-1699)

Implements the SSE polling behavior defined in SEP-1699:
- Server sends priming event (empty data with event ID) on SSE stream
- Server can call close_sse_stream() to trigger client reconnection
- Client auto-reconnects using server-provided retryInterval or exponential backoff

Github-Issue:#1654

Author: felixweinberger

src/mcp/client/streamable_http.py

@@ -175,8 +175,14 @@ def _get_next_reconnection_delay(self, attempt: int) -> float:
         Returns:
             Time to wait in seconds before next reconnection attempt
         """
-        # TODO: Implement proper delay calculation with server retry for SEP-1699
-        return 1.0
+        # Use server-provided retry value if available
+        if self._server_retry_seconds is not None:
+            return self._server_retry_seconds
+
+        # Fall back to exponential backoff
+        opts = self.reconnection_options
+        delay = opts.initial_reconnection_delay * (opts.reconnection_delay_grow_factor**attempt)
+        return min(delay, opts.max_reconnection_delay)
 
     async def _handle_sse_event(
         self,
@@ -376,25 +382,39 @@ async def _handle_sse_response(
         attempt: int = 0,
     ) -> None:
         """Handle SSE response from the server with automatic reconnection."""
+        has_priming_event = False
+        last_event_id: str | None = None
+        is_complete = False
+
         try:
             event_source = EventSource(response)
             async for sse in event_source.aiter_sse():  # pragma: no branch
-                is_complete, _has_event_id = await self._handle_sse_event(
+                is_complete, has_event_id = await self._handle_sse_event(
                     sse,
                     ctx.read_stream_writer,
                     resumption_callback=(ctx.metadata.on_resumption_token_update if ctx.metadata else None),
                     is_initialization=is_initialization,
                 )
 
+                # Track priming events
+                if has_event_id:
+                    has_priming_event = True
+                    last_event_id = sse.id
+
                 # If the SSE event indicates completion, like returning response/error
                 # break the loop
                 if is_complete:
                     await response.aclose()
                     break
         except Exception as e:  # pragma: no cover
             logger.exception("Error reading SSE stream:")
-            await ctx.read_stream_writer.send(e)
-        # TODO: Implement auto-reconnection for SEP-1699
+            # Don't send exception if we can reconnect
+            if not (has_priming_event and last_event_id):
+                await ctx.read_stream_writer.send(e)
+
+        # Auto-reconnect if stream ended without completion and we have priming event
+        if not is_complete and has_priming_event and last_event_id:  # pragma: no cover
+            await self._attempt_sse_reconnection(ctx, last_event_id, attempt)
 
     async def _attempt_sse_reconnection(  # pragma: no cover
         self,
@@ -407,8 +427,42 @@ async def _attempt_sse_reconnection(  # pragma: no cover
         Called when SSE stream ends without receiving a response/error,
         but we have a priming event indicating resumability.
         """
-        # TODO: Implement SSE reconnection for SEP-1699
-        pass
+        max_retries = self.reconnection_options.max_retries
+
+        if attempt >= max_retries:
+            error_msg = f"Max reconnection attempts ({max_retries}) exceeded"
+            logger.error(error_msg)
+            await ctx.read_stream_writer.send(StreamableHTTPError(error_msg))
+            return
+
+        # Calculate delay (uses server retry if available, else exponential backoff)
+        delay = self._get_next_reconnection_delay(attempt)
+        logger.info(f"SSE stream closed, reconnecting in {delay:.1f}s (attempt {attempt + 1}/{max_retries})")
+
+        await anyio.sleep(delay)
+
+        # Build resumption context with last_event_id
+        resumption_metadata = ClientMessageMetadata(
+            resumption_token=last_event_id,
+            on_resumption_token_update=(ctx.metadata.on_resumption_token_update if ctx.metadata else None),
+        )
+
+        resumption_ctx = RequestContext(
+            client=ctx.client,
+            headers=ctx.headers,
+            session_id=ctx.session_id,
+            session_message=ctx.session_message,
+            metadata=resumption_metadata,
+            read_stream_writer=ctx.read_stream_writer,
+            sse_read_timeout=ctx.sse_read_timeout,
+        )
+
+        try:
+            await self._handle_resumption_request(resumption_ctx)
+        except Exception as e:
+            logger.warning(f"Reconnection attempt {attempt + 1} failed: {e}")
+            # Recursive retry with incremented attempt counter
+            await self._attempt_sse_reconnection(ctx, last_event_id, attempt + 1)
 
     async def _handle_unexpected_content_type(
         self,

src/mcp/server/streamable_http.py

@@ -263,8 +263,25 @@ async def _create_priming_event(self, stream_id: str) -> dict[str, str | int] |
         Returns:
             Event data dictionary for the priming event, or None if no event store
         """
-        # TODO: Implement priming event creation for SEP-1699
-        return None
+        if self._event_store is None:
+            return None
+
+        # Store an empty message to get an event ID
+        # Using an empty dict as a placeholder - it won't be sent as actual data
+        priming_event_id = await self._event_store.store_event(
+            stream_id, JSONRPCMessage.model_validate({"jsonrpc": "2.0", "method": "_priming"})
+        )
+
+        event_data: dict[str, str | int] = {
+            "id": priming_event_id,
+            "data": "",  # Empty data for priming event
+        }
+
+        # Add retry interval if configured (sse_starlette expects int, not str)
+        if self._retry_interval is not None:
+            event_data["retry"] = self._retry_interval
+
+        return event_data
 
     async def _clean_up_memory_streams(self, request_id: RequestId) -> None:  # pragma: no cover
         """Clean up memory streams for a given request ID."""
@@ -739,8 +756,17 @@ async def close_sse_stream(self, request_id: RequestId) -> None:
         Args:
             request_id: The request ID (or stream key) of the stream to close
         """
-        # TODO: Implement stream closing for SEP-1699
-        pass
+        request_id_str = str(request_id)
+        if request_id_str in self._request_streams:
+            try:
+                sender, receiver = self._request_streams[request_id_str]
+                await sender.aclose()
+                await receiver.aclose()
+            except Exception:  # pragma: no cover
+                # Stream might already be closed
+                logger.debug(f"Error closing SSE stream {request_id_str} - may already be closed")
+            finally:
+                self._request_streams.pop(request_id_str, None)
 
     async def _handle_unsupported_request(self, request: Request, send: Send) -> None:  # pragma: no cover
         """Handle unsupported HTTP methods."""

tests/shared/test_streamable_http.py

@@ -1663,11 +1663,12 @@ async def test_handle_sse_event_skips_empty_data():
     write_stream, read_stream = anyio.create_memory_object_stream[SessionMessage | Exception](1)
 
     try:
-        # Call _handle_sse_event with empty data - should return False and not raise
-        result = await transport._handle_sse_event(mock_sse, write_stream)
+        # Call _handle_sse_event with empty data - should return (False, False) and not raise
+        is_complete, has_event_id = await transport._handle_sse_event(mock_sse, write_stream)
 
-        # Should return False (not complete) for empty data
-        assert result is False
+        # Should return False (not complete) for empty data, False for no event ID
+        assert is_complete is False
+        assert has_event_id is False
 
         # Nothing should have been written to the stream
         # Check buffer is empty (statistics().current_buffer_used returns buffer size)