Give each SSE response its own header list (#3726)

Author: Kludex

src/mcp/server/_streamable_http_modern.py

@@ -77,9 +77,6 @@
 logger = logging.getLogger(__name__)
 
 
-_OK_STATUS = 200
-
-
 @dataclass
 class _SingleExchangeDispatchContext:
     """`DispatchContext` for one inbound HTTP request.
@@ -146,13 +143,6 @@ async def _to_jsonrpc_response(
 _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"),
-    (b"connection", b"keep-alive"),
-    (b"x-accel-buffering", b"no"),
-]
-
 
 def _sse_event(msg: JSONRPCResponse | JSONRPCError | JSONRPCNotification) -> bytes:
     """Serialise a JSON-RPC message as one SSE `event: message` frame.
@@ -188,7 +178,7 @@ async def _write(
     send: Send,
 ) -> None:
     """Serialise a JSON-RPC reply with the table-mapped HTTP status."""
-    status = ERROR_CODE_HTTP_STATUS.get(msg.error.code, _OK_STATUS) if isinstance(msg, JSONRPCError) else _OK_STATUS
+    status = ERROR_CODE_HTTP_STATUS.get(msg.error.code, 200) if isinstance(msg, JSONRPCError) else 200
     body = msg.model_dump(mode="json", by_alias=True, exclude_none=True)
     if isinstance(msg, JSONRPCError) and msg.id is None:
         # JSON-RPC requires `id: null` to appear on the wire when the request
@@ -512,7 +502,18 @@ async def watch_disconnect(cancel_scope: anyio.CancelScope) -> None:
             # 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})
+            await send(
+                {
+                    "type": "http.response.start",
+                    "status": 200,
+                    "headers": [
+                        (b"content-type", b"text/event-stream"),
+                        (b"cache-control", b"no-cache, no-transform"),
+                        (b"connection", b"keep-alive"),
+                        (b"x-accel-buffering", b"no"),
+                    ],
+                }
+            )
             while not done:
                 await send({"type": "http.response.body", "body": event or b": ping\r\n\r\n", "more_body": True})
                 event = None

tests/server/test_streamable_http_modern.py

@@ -500,6 +500,38 @@ async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams |
     assert events[2]["result"]["tools"] == []
 
 
+async def test_sse_mode_headers_are_not_shared_across_responses() -> None:
+    """SSE mode: each response gets its own header list, so middleware appending in place
+    cannot leak headers into later responses."""
+
+    async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
+        await ctx.session.report_progress(1.0)
+        return ListToolsResult(tools=[], ttl_ms=0, cache_scope="public")
+
+    server = Server("test", on_list_tools=list_tools)
+
+    async def app(scope: Scope, receive: Receive, send: Send) -> None:
+        async def append_header(message: Message) -> None:
+            if message["type"] == "http.response.start":
+                message["headers"].append((b"x-id", b"abc"))
+            await send(message)
+
+        async with server.lifespan(server) as lifespan_state:
+            await handle_modern_request(server, None, False, lifespan_state, scope, receive, append_header)
+
+    async with httpx2.AsyncClient(
+        transport=StreamingASGITransport(app),
+        base_url="http://testserver",
+        headers={MCP_PROTOCOL_VERSION_HEADER: LATEST_MODERN_VERSION, "accept": "application/json, text/event-stream"},
+    ) as http:
+        for _ in range(2):
+            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 response.headers.get_list("x-id") == ["abc"]
+
+
 @pytest.mark.parametrize(
     "anyio_backend",
     [pytest.param(("trio", {"clock": MockClock(autojump_threshold=0)}), id="trio-mockclock")],