Add TDD stubs and failing tests for SSE polling (close_sse_stream)

This commit adds the API stubs and failing tests for the server-side
disconnect feature that enables SSE polling. When implemented, this
will allow servers to disconnect SSE streams without terminating them,
triggering client reconnection for polling patterns.

API stubs added:
- CloseSSEStreamCallback type in message.py
- close_sse_stream field in ServerMessageMetadata and RequestContext
- close_sse_stream() stub in StreamableHTTPServerTransport
- close_sse_stream() stub in FastMCP Context
- retry_interval parameter in session manager and transport

Tests added (all expected to fail until implementation):
- test_streamablehttp_client_receives_priming_event
- test_server_close_sse_stream_via_context
- test_streamablehttp_client_auto_reconnects
- test_streamablehttp_client_respects_retry_interval
- test_streamablehttp_sse_polling_full_cycle
- test_streamablehttp_events_replayed_after_disconnect

Github-Issue:#1699

Author: felixweinberger

src/mcp/server/fastmcp/server.py

@@ -1282,6 +1282,25 @@ def session(self):
         """Access to the underlying session for advanced usage."""
         return self.request_context.session
 
+    async def close_sse_stream(self) -> None:
+        """Close the SSE stream to trigger client reconnection.
+
+        This method closes the HTTP connection for the current request, triggering
+        client reconnection. Events continue to be stored in the event store and will
+        be replayed when the client reconnects with Last-Event-ID.
+
+        Use this to implement polling behavior during long-running operations -
+        client will reconnect after the retry interval specified in the priming event.
+
+        Note:
+            This is a no-op if not using StreamableHTTP transport with event_store.
+            The callback is only available when event_store is configured.
+
+        Raises:
+            NotImplementedError: Feature not yet implemented.
+        """
+        raise NotImplementedError("close_sse_stream not yet implemented")
+
     # Convenience methods for common log levels
     async def debug(self, message: str, **extra: Any) -> None:
         """Send a debug log message."""

src/mcp/server/streamable_http.py

@@ -140,6 +140,7 @@ def __init__(
         is_json_response_enabled: bool = False,
         event_store: EventStore | None = None,
         security_settings: TransportSecuritySettings | None = None,
+        retry_interval: int | None = None,
     ) -> None:
         """
         Initialize a new StreamableHTTP server transport.
@@ -153,6 +154,10 @@ def __init__(
                         resumability will be enabled, allowing clients to
                         reconnect and resume messages.
             security_settings: Optional security settings for DNS rebinding protection.
+            retry_interval: Retry interval in milliseconds to suggest to clients in SSE
+                           retry field. When set, the server will send a retry field in
+                           SSE priming events to control client reconnection timing for
+                           polling behavior. Only used when event_store is provided.
 
         Raises:
             ValueError: If the session ID contains invalid characters.
@@ -164,6 +169,7 @@ def __init__(
         self.is_json_response_enabled = is_json_response_enabled
         self._event_store = event_store
         self._security = TransportSecurityMiddleware(security_settings)
+        self._retry_interval = retry_interval
         self._request_streams: dict[
             RequestId,
             tuple[
@@ -178,6 +184,26 @@ def is_terminated(self) -> bool:
         """Check if this transport has been explicitly terminated."""
         return self._terminated
 
+    def close_sse_stream(self, request_id: RequestId) -> None:
+        """Close SSE connection for a specific request without terminating the stream.
+
+        This method closes the HTTP connection for the specified request, triggering
+        client reconnection. Events continue to be stored in the event store and will
+        be replayed when the client reconnects with Last-Event-ID.
+
+        Use this to implement polling behavior during long-running operations -
+        client will reconnect after the retry interval specified in the priming event.
+
+        Args:
+            request_id: The request ID whose SSE stream should be closed.
+
+        Note:
+            This is a no-op if there is no active stream for the request ID.
+            Requires event_store to be configured for events to be stored during
+            the disconnect.
+        """
+        raise NotImplementedError("close_sse_stream not yet implemented")
+
     def _create_error_response(
         self,
         error_message: str,

src/mcp/server/streamable_http_manager.py

@@ -51,6 +51,9 @@ class StreamableHTTPSessionManager:
         json_response: Whether to use JSON responses instead of SSE streams
         stateless: If True, creates a completely fresh transport for each request
                    with no session tracking or state persistence between requests.
+        security_settings: Optional transport security settings.
+        retry_interval: Retry interval in milliseconds to suggest to clients in SSE
+                       retry field. Used for SSE polling behavior.
     """
 
     def __init__(
@@ -60,12 +63,14 @@ def __init__(
         json_response: bool = False,
         stateless: bool = False,
         security_settings: TransportSecuritySettings | None = None,
+        retry_interval: int | None = None,
     ):
         self.app = app
         self.event_store = event_store
         self.json_response = json_response
         self.stateless = stateless
         self.security_settings = security_settings
+        self.retry_interval = retry_interval
 
         # Session tracking (only used if not stateless)
         self._session_creation_lock = anyio.Lock()
@@ -226,6 +231,7 @@ async def _handle_stateful_request(
                     is_json_response_enabled=self.json_response,
                     event_store=self.event_store,  # May be None (no resumability)
                     security_settings=self.security_settings,
+                    retry_interval=self.retry_interval,
                 )
 
                 assert http_transport.mcp_session_id is not None

src/mcp/shared/context.py

@@ -3,6 +3,7 @@
 
 from typing_extensions import TypeVar
 
+from mcp.shared.message import CloseSSEStreamCallback
 from mcp.shared.session import BaseSession
 from mcp.types import RequestId, RequestParams
 
@@ -18,3 +19,4 @@ class RequestContext(Generic[SessionT, LifespanContextT, RequestT]):
     session: SessionT
     lifespan_context: LifespanContextT
     request: RequestT | None = None
+    close_sse_stream: CloseSSEStreamCallback | None = None

src/mcp/shared/message.py

@@ -14,6 +14,9 @@
 
 ResumptionTokenUpdateCallback = Callable[[ResumptionToken], Awaitable[None]]
 
+# Callback type for closing SSE streams without terminating
+CloseSSEStreamCallback = Callable[[], Awaitable[None]]
+
 
 @dataclass
 class ClientMessageMetadata:
@@ -30,6 +33,8 @@ class ServerMessageMetadata:
     related_request_id: RequestId | None = None
     # Request-specific context (e.g., headers, auth info)
     request_context: object | None = None
+    # Callback to close SSE stream without terminating
+    close_sse_stream: CloseSSEStreamCallback | None = None
 
 
 MessageMetadata = ClientMessageMetadata | ServerMessageMetadata | None