Simplify SEP-1699 API design

- Remove StreamableHTTPReconnectionOptions dataclass (4 params -> 1 max_retries)
- Per spec, client MUST use server's retry field; exponential backoff removed
- Add get_stream_id_for_event_id() to EventStore for stream identification
- Simplify CloseSSEStreamCallback to no-arg (remove unused retry_interval_ms)
- Add close_standalone_sse_stream() for GET notification streams
- Update tests and examples for simplified API

Net: -84 lines, cleaner API that follows spec more closely

Author: felixweinberger

examples/snippets/clients/sse_polling_client.py

@@ -24,7 +24,7 @@
 import logging
 
 from mcp import ClientSession
-from mcp.client.streamable_http import StreamableHTTPReconnectionOptions, streamablehttp_client
+from mcp.client.streamable_http import streamablehttp_client
 from mcp.types import LoggingMessageNotificationParams, TextContent
 
 logging.basicConfig(
@@ -50,19 +50,13 @@ async def logging_callback(params: LoggingMessageNotificationParams) -> None:
             notifications_received.append(data_str)
             print(f"[Progress] {data_str}")
 
-    # Configure reconnection behavior
-    reconnection_options = StreamableHTTPReconnectionOptions(
-        initial_reconnection_delay=1.0,  # Start with 1 second
-        max_reconnection_delay=30.0,  # Cap at 30 seconds
-        reconnection_delay_grow_factor=1.5,  # Exponential backoff
-        max_retries=5,  # Try up to 5 times
-    )
-
     print("[Client] Connecting to server...")
 
+    # The client automatically uses the server's retry interval (per SEP-1699 spec)
+    # max_retries controls how many reconnection attempts before giving up
     async with streamablehttp_client(
         "http://localhost:3001/mcp",
-        reconnection_options=reconnection_options,
+        max_retries=5,
     ) as (read_stream, write_stream, get_session_id):
         # Create session with logging callback to receive progress notifications
         async with ClientSession(

src/mcp/client/streamable_http.py

@@ -58,25 +58,10 @@ class ResumptionError(StreamableHTTPError):
     """Raised when resumption request is invalid."""
 
 
-@dataclass
-class StreamableHTTPReconnectionOptions:
-    """Configuration options for reconnection behavior of StreamableHTTPTransport.
-
-    Attributes:
-        initial_reconnection_delay: Initial backoff time in seconds. Default is 1.0.
-        max_reconnection_delay: Maximum backoff time in seconds. Default is 30.0.
-        reconnection_delay_grow_factor: Factor by which delay increases. Default is 1.5.
-        max_retries: Maximum reconnection attempts. Default is 2.
-    """
-
-    initial_reconnection_delay: float = 1.0
-    max_reconnection_delay: float = 30.0
-    reconnection_delay_grow_factor: float = 1.5
-    max_retries: int = 2
-
-    def __post_init__(self) -> None:
-        if self.initial_reconnection_delay > self.max_reconnection_delay:
-            raise ValueError("initial_reconnection_delay cannot exceed max_reconnection_delay")
+# Default reconnection settings per SEP-1699
+# The server controls timing via SSE retry field; this is just a fallback
+DEFAULT_RECONNECTION_DELAY = 1.0  # seconds, used when server doesn't provide retry
+DEFAULT_MAX_RETRIES = 2
 
 
 @dataclass
@@ -102,7 +87,7 @@ def __init__(
         timeout: float | timedelta = 30,
         sse_read_timeout: float | timedelta = 60 * 5,
         auth: httpx.Auth | None = None,
-        reconnection_options: StreamableHTTPReconnectionOptions | None = None,
+        max_retries: int = DEFAULT_MAX_RETRIES,
     ) -> None:
         """Initialize the StreamableHTTP transport."""
         self.url = url
@@ -114,7 +99,7 @@ def __init__(
         self.auth = auth
         self.session_id = None
         self.protocol_version = None
-        self.reconnection_options = reconnection_options or StreamableHTTPReconnectionOptions()
+        self.max_retries = max_retries
         self._server_retry_seconds: float | None = None  # Server-provided retry delay
         self.request_headers = {
             ACCEPT: f"{JSON}, {SSE}",
@@ -166,23 +151,15 @@ def _maybe_extract_protocol_version_from_message(
                 )  # pragma: no cover
                 logger.warning(f"Raw result: {message.root.result}")
 
-    def _get_next_reconnection_delay(self, attempt: int) -> float:
-        """Calculate the next reconnection delay using exponential backoff.
+    def _get_reconnection_delay(self) -> float:
+        """Get the reconnection delay per SEP-1699.
 
-        Args:
-            attempt: Current reconnection attempt count
-
-        Returns:
-            Time to wait in seconds before next reconnection attempt
+        Returns the server-provided retry value if available (MUST per spec),
+        otherwise falls back to a simple default.
         """
-        # 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)
+        return DEFAULT_RECONNECTION_DELAY
 
     async def _handle_sse_event(
         self,
@@ -427,17 +404,15 @@ 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.
         """
-        max_retries = self.reconnection_options.max_retries
-
-        if attempt >= max_retries:
-            error_msg = f"Max reconnection attempts ({max_retries}) exceeded"
+        if attempt >= self.max_retries:
+            error_msg = f"Max reconnection attempts ({self.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})")
+        # Use server-provided retry interval (MUST per spec) or fallback
+        delay = self._get_reconnection_delay()
+        logger.info(f"SSE stream closed, reconnecting in {delay:.1f}s (attempt {attempt + 1}/{self.max_retries})")
 
         await anyio.sleep(delay)
 
@@ -635,7 +610,7 @@ async def streamablehttp_client(
     terminate_on_close: bool = True,
     httpx_client_factory: McpHttpClientFactory = create_mcp_http_client,
     auth: httpx.Auth | None = None,
-    reconnection_options: StreamableHTTPReconnectionOptions | None = None,
+    max_retries: int = DEFAULT_MAX_RETRIES,
 ) -> AsyncGenerator[
     tuple[
         MemoryObjectReceiveStream[SessionMessage | Exception],
@@ -649,8 +624,10 @@ async def streamablehttp_client(
 
     `sse_read_timeout` determines how long (in seconds) the client will wait for a new
     event before disconnecting. All other HTTP operations are controlled by `timeout`.
+    `max_retries` controls how many times the client will attempt to reconnect when
+    the server closes an SSE stream mid-operation (SEP-1699 polling).
     """
-    transport = StreamableHTTPTransport(url, headers, timeout, sse_read_timeout, auth, reconnection_options)
+    transport = StreamableHTTPTransport(url, headers, timeout, sse_read_timeout, auth, max_retries)
 
     read_stream_writer, read_stream = anyio.create_memory_object_stream[SessionMessage | Exception](0)
     write_stream, write_stream_reader = anyio.create_memory_object_stream[SessionMessage](0)

src/mcp/server/fastmcp/server.py

@@ -1304,23 +1304,17 @@ async def error(self, message: str, **extra: Any) -> None:
         """Send an error log message."""
         await self.log("error", message, **extra)
 
-    async def close_sse_stream(self, retry_interval_ms: int | None = None) -> None:
+    async def close_sse_stream(self) -> None:
         """Close the SSE stream for this request, triggering client reconnection.
 
         Use this to implement polling behavior during long-running operations.
-        The client will reconnect after the retry interval and receive any events
-        that were stored while disconnected.
+        The client will reconnect after the retry interval (configured at the
+        transport level) and receive any events that were stored while disconnected.
 
         This is only available when using StreamableHTTP transport with an
         event store configured for resumability. It is a no-op otherwise,
         allowing portable tool code.
-
-        Args:
-            retry_interval_ms: Optional retry interval in milliseconds to suggest
-                              to the client before closing (currently unused,
-                              reserved for future use - the retry interval is
-                              configured at the transport level)
         """
         callback = self._request_context.close_sse_stream if self._request_context else None
         if callback:
-            await callback(retry_interval_ms)
+            await callback()

src/mcp/server/streamable_http.py

@@ -118,6 +118,26 @@ async def replay_events_after(
         """
         pass  # pragma: no cover
 
+    async def get_stream_id_for_event_id(self, event_id: EventId) -> StreamId | None:
+        """
+        Get the stream ID associated with a given event ID.
+
+        This optional method allows servers to look up which stream an event
+        belongs to, supporting proper stream identification for resumption.
+
+        Args:
+            event_id: The event ID to look up
+
+        Returns:
+            The stream ID, or None if not found or not implemented
+
+        Note:
+            The default implementation returns None. Override this in your
+            EventStore implementation if you need explicit stream identification
+            (e.g., for multiple concurrent resumable streams).
+        """
+        return None
+
 
 class StreamableHTTPServerTransport:
     """
@@ -740,9 +760,7 @@ def _create_close_sse_callback(self, request_id: str) -> CloseSSEStreamCallback
         if not self._event_store:
             return None
 
-        async def close_callback(retry_interval_ms: int | None = None) -> None:
-            # Note: retry_interval_ms is accepted for API compatibility but not used
-            # The retry interval is set via the priming event
+        async def close_callback() -> None:
             await self.close_sse_stream(request_id)
 
         return close_callback
@@ -754,7 +772,7 @@ async def close_sse_stream(self, request_id: RequestId) -> None:
         client will reconnect after the retry interval specified in the priming event.
 
         Args:
-            request_id: The request ID (or stream key) of the stream to close
+            request_id: The request ID (or GET_STREAM_KEY for standalone GET stream)
         """
         request_id_str = str(request_id)
         if request_id_str in self._request_streams:
@@ -768,6 +786,14 @@ async def close_sse_stream(self, request_id: RequestId) -> None:
             finally:
                 self._request_streams.pop(request_id_str, None)
 
+    async def close_standalone_sse_stream(self) -> None:
+        """Close the standalone GET notification stream, triggering client reconnection.
+
+        This is a convenience method equivalent to close_sse_stream(GET_STREAM_KEY).
+        Use this to implement polling behavior for server-initiated notifications.
+        """
+        await self.close_sse_stream(GET_STREAM_KEY)
+
     async def _handle_unsupported_request(self, request: Request, send: Send) -> None:  # pragma: no cover
         """Handle unsupported HTTP methods."""
         headers = {

src/mcp/shared/message.py

@@ -14,8 +14,8 @@
 
 ResumptionTokenUpdateCallback = Callable[[ResumptionToken], Awaitable[None]]
 
-# Callback type for closing SSE streams (takes optional retry_interval_ms)
-CloseSSEStreamCallback = Callable[[int | None], Awaitable[None]]
+# Callback type for closing SSE streams to trigger client reconnection
+CloseSSEStreamCallback = Callable[[], Awaitable[None]]
 
 
 @dataclass