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/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:

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,