Bound the SSE deferral window at the keepalive interval

A handler that runs silent past _SSE_PING_INTERVAL now commits
text/event-stream and starts pinging instead of leaving zero bytes on the
wire — so a proxy idle-read timeout can't close the connection (which on
this path cancels the handler) before the result arrives. Kernel-dispatch
errors still get the table-mapped JSON status because they resolve well
within the 15s window.

Author: maxisbey

src/mcp/server/_streamable_http_modern.py

@@ -8,9 +8,12 @@
 `Mcp-Session-Id`, one JSON-RPC request in, one JSON-RPC response out. JSON
 mode handles the request directly in the ASGI task. SSE mode runs the handler
 as a sibling task and defers committing to `text/event-stream` until the
-handler actually emits a notification: a handler that completes (or raises)
-without emitting still gets a JSON response with the table-mapped HTTP
-status, so the spec's `404`/`400` MUSTs hold for kernel-dispatch errors.
+handler emits a notification or `_SSE_PING_INTERVAL` elapses, whichever
+comes first: a handler that completes (or raises) within that window without
+emitting still gets a JSON response with the table-mapped HTTP status, so
+the spec's `404`/`400` MUSTs hold for kernel-dispatch errors; a handler that
+runs silent past the window commits SSE so the keepalive ping can keep the
+connection open behind a proxy idle-read timeout.
 """
 
 from __future__ import annotations
@@ -298,24 +301,33 @@ async def watch_disconnect(cancel_scope: anyio.CancelScope) -> None:
         tg.start_soon(run_handler)
         tg.start_soon(watch_disconnect, tg.cancel_scope)
 
-        try:
-            first = await recv_ch.receive()
-        except anyio.EndOfStream:
-            first = None
-
-        if first is None:
+        event: bytes | None = None
+        done = False
+        with anyio.move_on_after(_SSE_PING_INTERVAL):
+            try:
+                event = await recv_ch.receive()
+            except anyio.EndOfStream:
+                done = True
+
+        if done:
+            # Handler completed within the deferral window without emitting:
+            # `application/json` with the table-mapped status. Kernel-dispatch
+            # errors (METHOD_NOT_FOUND, missing-capability, INVALID_PARAMS)
+            # resolve here in practice.
             await _write(result[0], scope, receive, send)
         else:
+            # First notification arrived, or the deferral window elapsed: commit
+            # `text/event-stream` and start pinging so a proxy idle-read timeout
+            # cannot close the stream (which on this path cancels the handler).
             await send({"type": "http.response.start", "status": _OK_STATUS, "headers": _SSE_HEADERS})
-            event: bytes | None = first
-            while True:
+            while not done:
                 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
+                        done = True
             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

@@ -396,6 +396,29 @@ async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams |
     assert events[1]["result"]["tools"] == []
 
 
+async def test_sse_mode_silent_handler_commits_sse_after_ping_interval(monkeypatch: pytest.MonkeyPatch) -> None:
+    """SSE mode: a handler that runs silent past the deferral window commits `text/event-stream`
+    and starts pinging — even though it never emits a notification — so a proxy idle-read timeout
+    does not close the connection and cancel it. SDK-defined: the deferral window is bounded by
+    `_SSE_PING_INTERVAL`."""
+    monkeypatch.setattr("mcp.server._streamable_http_modern._SSE_PING_INTERVAL", 0.01)
+
+    async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
+        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(), headers={MCP_METHOD_HEADER: "tools/list"})
+
+    assert response.status_code == 200
+    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) == 1
+    assert events[0]["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