Add retry_interval to FastMCP and test_reconnection to everything-server (SEP-1699)

- Add retry_interval parameter to FastMCP for SSE polling control
- Add InMemoryEventStore and test_reconnection tool to everything-server
- Enables SSE polling conformance test to pass (server-sse-polling scenario)

Author: felixweinberger

examples/clients/sse-polling-client/README.md

@@ -24,7 +24,7 @@ uv run mcp-sse-polling-client --url http://localhost:3000/mcp --items 20 --check
 
 ## Options
 
-- `--url`: Server URL (default: http://localhost:3000/mcp)
+- `--url`: Server URL (default: <http://localhost:3000/mcp>)
 - `--items`: Number of items to process (default: 10)
 - `--checkpoint-every`: Checkpoint interval (default: 3)
 - `--log-level`: Logging level (default: DEBUG)

examples/servers/everything-server/mcp_everything_server/server.py

@@ -14,13 +14,15 @@
 from mcp.server.fastmcp import Context, FastMCP
 from mcp.server.fastmcp.prompts.base import UserMessage
 from mcp.server.session import ServerSession
+from mcp.server.streamable_http import EventCallback, EventMessage, EventStore
 from mcp.types import (
     AudioContent,
     Completion,
     CompletionArgument,
     CompletionContext,
     EmbeddedResource,
     ImageContent,
+    JSONRPCMessage,
     PromptReference,
     ResourceTemplateReference,
     SamplingMessage,
@@ -31,6 +33,43 @@
 
 logger = logging.getLogger(__name__)
 
+# Type aliases for event store
+StreamId = str
+EventId = str
+
+
+class InMemoryEventStore(EventStore):
+    """Simple in-memory event store for SSE resumability testing."""
+
+    def __init__(self) -> None:
+        self._events: list[tuple[StreamId, EventId, JSONRPCMessage | None]] = []
+        self._event_id_counter = 0
+
+    async def store_event(self, stream_id: StreamId, message: JSONRPCMessage | None) -> EventId:
+        """Store an event and return its ID."""
+        self._event_id_counter += 1
+        event_id = str(self._event_id_counter)
+        self._events.append((stream_id, event_id, message))
+        return event_id
+
+    async def replay_events_after(self, last_event_id: EventId, send_callback: EventCallback) -> StreamId | None:
+        """Replay events after the specified ID."""
+        target_stream_id = None
+        for stream_id, event_id, _ in self._events:
+            if event_id == last_event_id:
+                target_stream_id = stream_id
+                break
+        if target_stream_id is None:
+            return None
+        last_event_id_int = int(last_event_id)
+        for stream_id, event_id, message in self._events:
+            if stream_id == target_stream_id and int(event_id) > last_event_id_int:
+                # Skip priming events (None message)
+                if message is not None:
+                    await send_callback(EventMessage(message, event_id))
+        return target_stream_id
+
+
 # Test data
 TEST_IMAGE_BASE64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg=="
 TEST_AUDIO_BASE64 = "UklGRiYAAABXQVZFZm10IBAAAAABAAEAQB8AAAB9AAACABAAZGF0YQIAAAA="
@@ -39,8 +78,13 @@
 resource_subscriptions: set[str] = set()
 watched_resource_content = "Watched resource content"
 
+# Create event store for SSE resumability (SEP-1699)
+event_store = InMemoryEventStore()
+
 mcp = FastMCP(
     name="mcp-conformance-test-server",
+    event_store=event_store,
+    retry_interval=100,  # 100ms retry interval for SSE polling
 )
 
 
@@ -263,6 +307,19 @@ def test_error_handling() -> str:
     raise RuntimeError("This tool intentionally returns an error for testing")
 
 
+@mcp.tool()
+async def test_reconnection(ctx: Context[ServerSession, None]) -> str:
+    """Tests SSE polling by closing stream mid-call (SEP-1699)"""
+    await ctx.info("Before disconnect")
+
+    await ctx.close_sse_stream()
+
+    await asyncio.sleep(0.2)  # Wait for client to reconnect
+
+    await ctx.info("After reconnect")
+    return "Reconnection test completed"
+
+
 # Resources
 @mcp.resource("test://static-text")
 def static_text_resource() -> str:

examples/servers/simple-streamablehttp/mcp_simple_streamablehttp/event_store.py

@@ -24,7 +24,7 @@ class EventEntry:
 
     event_id: EventId
     stream_id: StreamId
-    message: JSONRPCMessage
+    message: JSONRPCMessage | None
 
 
 class InMemoryEventStore(EventStore):
@@ -48,7 +48,7 @@ def __init__(self, max_events_per_stream: int = 100):
         # event_id -> EventEntry for quick lookup
         self.event_index: dict[EventId, EventEntry] = {}
 
-    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 with a generated event ID."""
         event_id = str(uuid4())
         event_entry = EventEntry(event_id=event_id, stream_id=stream_id, message=message)
@@ -88,7 +88,9 @@ async def replay_events_after(
         found_last = False
         for event in stream_events:
             if found_last:
-                await send_callback(EventMessage(event.message, event.event_id))
+                # Skip priming events (None message)
+                if event.message is not None:
+                    await send_callback(EventMessage(event.message, event.event_id))
             elif event.event_id == last_event_id:
                 found_last = True
 

src/mcp/client/streamable_http.py

@@ -403,7 +403,7 @@ async def _handle_reconnection(
                 logger.info("Reconnected to SSE stream")
 
                 # Track for potential further reconnection
-                reconnect_last_event_id: str | None = last_event_id
+                reconnect_last_event_id: str = last_event_id
                 reconnect_retry_ms = retry_interval_ms
 
                 async for sse in event_source.aiter_sse():
@@ -423,11 +423,8 @@ async def _handle_reconnection(
                         return
 
                 # Stream ended again without response - reconnect again (reset attempt counter)
-                if reconnect_last_event_id is not None:
-                    logger.info("SSE stream disconnected, reconnecting...")
-                    await self._handle_reconnection(
-                        ctx, reconnect_last_event_id, reconnect_retry_ms, 0
-                    )
+                logger.info("SSE stream disconnected, reconnecting...")
+                await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, 0)
         except Exception as e:
             logger.debug(f"Reconnection failed: {e}")
             # Try to reconnect again if we still have an event ID

src/mcp/server/fastmcp/server.py

@@ -153,6 +153,7 @@ def __init__(  # noqa: PLR0913
         auth_server_provider: (OAuthAuthorizationServerProvider[Any, Any, Any] | None) = None,
         token_verifier: TokenVerifier | None = None,
         event_store: EventStore | None = None,
+        retry_interval: int | None = None,
         *,
         tools: list[Tool] | None = None,
         debug: bool = False,
@@ -221,6 +222,7 @@ def __init__(  # noqa: PLR0913
         if auth_server_provider and not token_verifier:  # pragma: no cover
             self._token_verifier = ProviderTokenVerifier(auth_server_provider)
         self._event_store = event_store
+        self._retry_interval = retry_interval
         self._custom_starlette_routes: list[Route] = []
         self.dependencies = self.settings.dependencies
         self._session_manager: StreamableHTTPSessionManager | None = None
@@ -940,6 +942,7 @@ def streamable_http_app(self) -> Starlette:
             self._session_manager = StreamableHTTPSessionManager(
                 app=self._mcp_server,
                 event_store=self._event_store,
+                retry_interval=self._retry_interval,
                 json_response=self.settings.json_response,
                 stateless=self.settings.stateless_http,  # Use the stateless setting
                 security_settings=self.settings.transport_security,

tests/shared/test_streamable_http.py

@@ -110,7 +110,9 @@ async def replay_events_after(  # pragma: no cover
         # Replay only events from the same stream with ID > last_event_id
         for stream_id, event_id, message in self._events:
             if stream_id == target_stream_id and int(event_id) > last_event_id_int:
-                await send_callback(EventMessage(message, event_id))
+                # Skip priming events (None message)
+                if message is not None:
+                    await send_callback(EventMessage(message, event_id))
 
         return target_stream_id