Restore SSE keepalive ping after the deferred-start rewrite

The 5820a22 raw-ASGI rewrite dropped the comment-line keepalive that
EventSourceResponse used to provide, so a long-silent handler behind a
proxy idle-read timeout would have its stream closed (which on this path
cancels the handler). The receive loop now emits ': ping' on a 15s
move_on_after between events, matching the legacy paths' interval.

Author: maxisbey

src/mcp/server/_streamable_http_modern.py

@@ -147,6 +147,9 @@ async def _to_jsonrpc_response(
     return JSONRPCResponse(jsonrpc="2.0", id=request_id, result=result)
 
 
+_SSE_PING_INTERVAL: float = 15.0
+"""Seconds between SSE comment-line keepalives once `text/event-stream` has committed."""
+
 _SSE_HEADERS: Final[list[tuple[bytes, bytes]]] = [
     (b"content-type", b"text/event-stream"),
     (b"cache-control", b"no-cache, no-transform"),
@@ -304,9 +307,15 @@ async def watch_disconnect(cancel_scope: anyio.CancelScope) -> None:
             await _write(result[0], scope, receive, send)
         else:
             await send({"type": "http.response.start", "status": _OK_STATUS, "headers": _SSE_HEADERS})
-            await send({"type": "http.response.body", "body": first, "more_body": True})
-            async for event in recv_ch:
-                await send({"type": "http.response.body", "body": event, "more_body": True})
+            event: bytes | None = first
+            while True:
+                await send({"type": "http.response.body", "body": event or b": ping\r\n\r\n", "more_body": True})
+                event = None
+                with anyio.move_on_after(_SSE_PING_INTERVAL):
+                    try:
+                        event = await recv_ch.receive()
+                    except anyio.EndOfStream:
+                        break
             await send({"type": "http.response.body", "body": _sse_event(result[0]), "more_body": False})
 
         tg.cancel_scope.cancel()

tests/server/test_streamable_http_modern.py

@@ -371,6 +371,31 @@ async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams |
     assert events[2]["result"]["tools"] == []
 
 
+async def test_sse_mode_emits_keepalive_comment_between_events(monkeypatch: pytest.MonkeyPatch) -> None:
+    """SSE mode: while the stream is idle between events the server emits an SSE comment line so a
+    proxy idle-read timeout does not close the stream (which would cancel the handler).
+    SDK-defined: spec encourages keepalive comments for long-lived streams."""
+    monkeypatch.setattr("mcp.server._streamable_http_modern._SSE_PING_INTERVAL", 0.01)
+
+    async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
+        await ctx.session.report_progress(1.0)
+        await anyio.sleep(0.05)
+        return ListToolsResult(tools=[], ttl_ms=0, cache_scope="public")
+
+    async with _asgi_client(Server("test", on_list_tools=list_tools), json_response=False) as http:
+        with anyio.fail_after(5):
+            response = await http.post(
+                "/mcp", json=_list_tools_body_with_token("tok"), headers={MCP_METHOD_HEADER: "tools/list"}
+            )
+
+    assert response.headers["content-type"].split(";", 1)[0] == "text/event-stream"
+    assert b": ping\r\n\r\n" in response.content
+    events = _sse_payloads(response.text)
+    assert len(events) == 2
+    assert events[0]["method"] == "notifications/progress"
+    assert events[1]["result"]["tools"] == []
+
+
 async def test_sse_mode_streams_log_notification() -> None:
     """SSE mode: a request-scoped `notifications/message` emitted by the handler precedes the
     terminal response on the same stream. SDK-defined: notifications sent on the request's outbound