Defer SSE start until first notification; restore 404/400 status mapping

The previous commit committed text/event-stream headers immediately, so a
METHOD_NOT_FOUND raised inside serve_one reached the wire as HTTP 200 + an
SSE event — breaking the spec MUST that an unimplemented method respond
404. server-stateless conformance failed 7/25 on that.

The handler now runs as a sibling task while the parent waits for the first
event from the rendezvous stream. If the handler completes (or raises)
without emitting, the result is written as application/json with the
ERROR_CODE_HTTP_STATUS-mapped status — same as JSON mode. text/event-stream
headers go out only once a notification has actually arrived; from there a
disconnect-watcher in the same task group cancels serve_one on
http.disconnect. EventSourceResponse is dropped for this path (it sends
headers on __call__ and its task group can't adopt an already-running
handler); SSE frames are written directly via ASGI send.

Also: drop the _check_accept_headers wrapper method (review feedback).

Author: maxisbey

src/mcp/server/_streamable_http_modern.py

@@ -6,9 +6,11 @@
 
 A 2026-07-28 request is a self-contained POST: no `initialize` handshake, no
 `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 inside an `EventSourceResponse` so request-scoped
-notifications stream before the terminal response.
+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.
 """
 
 from __future__ import annotations
@@ -17,7 +19,7 @@
 import logging
 from collections.abc import Awaitable, Mapping
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any, TypeVar
+from typing import TYPE_CHECKING, Any, Final, TypeVar
 
 import anyio
 from anyio.streams.memory import MemoryObjectSendStream
@@ -36,7 +38,6 @@
     RequestId,
 )
 from pydantic import BaseModel, ValidationError
-from sse_starlette import EventSourceResponse
 from starlette.requests import Request
 from starlette.responses import Response
 from starlette.types import Receive, Scope, Send
@@ -80,7 +81,7 @@ class _SingleExchangeDispatchContext:
     request_id: RequestId
     message_metadata: MessageMetadata
     progress_token: ProgressToken | None = None
-    sink: MemoryObjectSendStream[dict[str, str]] | None = None
+    sink: MemoryObjectSendStream[bytes] | None = None
     cancel_requested: anyio.Event = field(default_factory=anyio.Event)
     can_send_request: bool = field(default=False, init=False)
 
@@ -146,14 +147,23 @@ async def _to_jsonrpc_response(
     return JSONRPCResponse(jsonrpc="2.0", id=request_id, result=result)
 
 
-def _sse_event(msg: JSONRPCResponse | JSONRPCError | JSONRPCNotification) -> dict[str, str]:
-    """Serialise a JSON-RPC message as an sse-starlette event dict.
+_SSE_HEADERS: Final[list[tuple[bytes, bytes]]] = [
+    (b"content-type", b"text/event-stream"),
+    (b"cache-control", b"no-store"),
+    (b"connection", b"keep-alive"),
+    (b"x-accel-buffering", b"no"),
+]
 
-    SSE mode begins after the request body has parsed, so a `JSONRPCError` here
+
+def _sse_event(msg: JSONRPCResponse | JSONRPCError | JSONRPCNotification) -> bytes:
+    """Serialise a JSON-RPC message as one SSE `event: message` frame.
+
+    SSE mode begins after the handler has emitted, so a `JSONRPCError` here
     always carries the request's id; the `id: null` case lives in `_write`.
     """
     body = msg.model_dump(mode="json", by_alias=True, exclude_none=True)
-    return {"event": "message", "data": json.dumps(body, separators=(",", ":"))}
+    data = json.dumps(body, separators=(",", ":"))
+    return f"event: message\r\ndata: {data}\r\n\r\n".encode()
 
 
 async def _write(
@@ -263,18 +273,40 @@ async def handle_modern_request(
         await _write(msg, scope, receive, send)
         return
 
-    send_ch, recv_ch = anyio.create_memory_object_stream[dict[str, str]](0)
+    send_ch, recv_ch = anyio.create_memory_object_stream[bytes](0)
     dctx.sink = send_ch
+    result: list[JSONRPCResponse | JSONRPCError] = []
 
     async def run_handler() -> None:
         async with send_ch:
-            msg = await _to_jsonrpc_response(
-                req.id,
-                serve_one(app, dctx, req.method, req.params, connection=connection, lifespan_state=lifespan_state),
+            result.append(
+                await _to_jsonrpc_response(
+                    req.id,
+                    serve_one(app, dctx, req.method, req.params, connection=connection, lifespan_state=lifespan_state),
+                )
             )
-            await send_ch.send(_sse_event(msg))
 
-    try:
-        await EventSourceResponse(content=recv_ch, data_sender_callable=run_handler)(scope, receive, send)
-    finally:
-        await recv_ch.aclose()
+    async def watch_disconnect(cancel_scope: anyio.CancelScope) -> None:
+        while (await receive()).get("type") != "http.disconnect":
+            pass  # pragma: no cover
+        cancel_scope.cancel()
+
+    async with recv_ch, anyio.create_task_group() as tg:
+        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:
+            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})
+            await send({"type": "http.response.body", "body": _sse_event(result[0]), "more_body": False})
+
+        tg.cancel_scope.cancel()

src/mcp/server/streamable_http.py

@@ -433,9 +433,6 @@ async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> No
         else:
             await self._handle_unsupported_request(request, send)
 
-    def _check_accept_headers(self, request: Request) -> tuple[bool, bool]:
-        return check_accept_headers(request)
-
     def _check_content_type(self, request: Request) -> bool:
         """Check if the request has the correct Content-Type."""
         content_type = request.headers.get("content-type", "")
@@ -445,7 +442,7 @@ def _check_content_type(self, request: Request) -> bool:
 
     async def _validate_accept_header(self, request: Request, scope: Scope, send: Send) -> bool:
         """Validate Accept header based on response mode. Returns True if valid."""
-        has_json, has_sse = self._check_accept_headers(request)
+        has_json, has_sse = check_accept_headers(request)
         if self.is_json_response_enabled:
             # For JSON-only responses, only require application/json
             if not has_json:
@@ -665,7 +662,7 @@ async def _handle_get_request(self, request: Request, send: Send) -> None:
             raise ValueError("No read stream writer available. Ensure connect() is called first.")
 
         # Validate Accept header - must include text/event-stream
-        _, has_sse = self._check_accept_headers(request)
+        _, has_sse = check_accept_headers(request)
 
         if not has_sse:
             response = self._create_error_response(

tests/interaction/transports/test_hosting_http_modern.py

@@ -106,7 +106,7 @@ async def test_modern_tools_call_returns_result_type_complete_without_initialize
         "method": "tools/call",
         "params": {"name": "add", "arguments": {"a": 2, "b": 3}, "_meta": _meta_envelope()},
     }
-    async with mounted_app(_server(), json_response=True) as (http, _):
+    async with mounted_app(_server()) as (http, _):
         response = await http.post("/mcp", json=body, headers=_modern_headers(method="tools/call", name="add"))
 
     assert response.status_code == 200
@@ -151,7 +151,7 @@ async def test_modern_initialize_is_method_not_found() -> None:
     negative.
     """
     body = {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"_meta": _meta_envelope()}}
-    async with mounted_app(_server(), json_response=True) as (http, _):
+    async with mounted_app(_server()) as (http, _):
         response = await http.post("/mcp", json=body, headers=_modern_headers(method="initialize"))
 
     assert response.status_code == 404
@@ -199,7 +199,7 @@ async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) ->
         "method": "tools/call",
         "params": {"name": "boom", "arguments": {}, "_meta": _meta_envelope()},
     }
-    async with mounted_app(Server("modern", on_call_tool=call_tool), json_response=True) as (http, _):
+    async with mounted_app(Server("modern", on_call_tool=call_tool)) as (http, _):
         response = await http.post("/mcp", json=body, headers=_modern_headers(method="tools/call", name="boom"))
 
     assert response.status_code == 200
@@ -218,7 +218,7 @@ async def test_modern_server_discover_returns_capabilities_and_supported_version
     the raw result body.
     """
     body = {"jsonrpc": "2.0", "id": 1, "method": "server/discover", "params": {"_meta": _meta_envelope()}}
-    async with mounted_app(_server(), json_response=True) as (http, _):
+    async with mounted_app(_server()) as (http, _):
         response = await http.post("/mcp", json=body, headers=_modern_headers(method="server/discover"))
 
     assert response.status_code == 200
@@ -238,7 +238,7 @@ async def test_modern_removed_method_is_method_not_found_at_http_404() -> None:
     wire because the HTTP status is the assertion.
     """
     body = {"jsonrpc": "2.0", "id": 1, "method": "ping", "params": {"_meta": _meta_envelope()}}
-    async with mounted_app(_server(), json_response=True) as (http, _):
+    async with mounted_app(_server()) as (http, _):
         response = await http.post("/mcp", json=body, headers=_modern_headers(method="ping"))
 
     assert response.status_code == 404
@@ -285,7 +285,7 @@ async def cap_check(ctx: ServerRequestContext, params: RequestParams) -> EmptyRe
     server = _server()
     server.add_request_handler("test/cap-check", RequestParams, cap_check)
     body = {"jsonrpc": "2.0", "id": 1, "method": "test/cap-check", "params": {"_meta": _meta_envelope()}}
-    async with mounted_app(server, json_response=True) as (http, _):
+    async with mounted_app(server) as (http, _):
         response = await http.post("/mcp", json=body, headers=_modern_headers(method="test/cap-check"))
 
     assert response.status_code == 400

tests/server/test_streamable_http_modern.py

@@ -415,10 +415,10 @@ async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams |
     assert "notifications/progress" not in response.text
 
 
-async def test_sse_mode_handler_error_is_event() -> None:
-    """SSE mode: a handler-raised `MCPError` is delivered as the terminal SSE event (HTTP 200).
-    SDK-defined: SSE responses commit headers before the handler runs, so the error rides the
-    event stream rather than the HTTP status line."""
+async def test_sse_mode_error_before_any_notify_is_json_with_mapped_status() -> None:
+    """SSE mode: an error raised before the handler emits any notification is written as
+    `application/json` with the table-mapped HTTP status — SSE has not committed yet.
+    Spec-mandated: METHOD_NOT_FOUND MUST be `404 Not Found`."""
 
     async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
         raise MCPError(code=METHOD_NOT_FOUND, message="nope")
@@ -427,17 +427,37 @@ async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams |
         with anyio.fail_after(5):
             response = await http.post("/mcp", json=_list_tools_body(), headers={MCP_METHOD_HEADER: "tools/list"})
 
+    assert response.status_code == 404
+    assert response.headers["content-type"].split(";", 1)[0] == "application/json"
+    assert response.json() == {"jsonrpc": "2.0", "id": 1, "error": {"code": METHOD_NOT_FOUND, "message": "nope"}}
+
+
+async def test_sse_mode_error_after_notify_is_sse_event() -> None:
+    """SSE mode: an error raised after the handler has emitted is delivered as the terminal SSE
+    event (HTTP 200) — `text/event-stream` headers were committed on the first notification."""
+
+    async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
+        await ctx.session.report_progress(1.0)
+        raise MCPError(code=INTERNAL_ERROR, message="boom")
+
+    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.status_code == 200
     assert response.headers["content-type"].split(";", 1)[0] == "text/event-stream"
     events = _sse_payloads(response.text)
-    assert len(events) == 1
-    assert events[0] == {"jsonrpc": "2.0", "id": 1, "error": {"code": METHOD_NOT_FOUND, "message": "nope"}}
+    assert len(events) == 2
+    assert events[0]["method"] == "notifications/progress"
+    assert events[1] == {"jsonrpc": "2.0", "id": 1, "error": {"code": INTERNAL_ERROR, "message": "boom"}}
 
 
-async def test_sse_mode_no_token_progress_noop() -> None:
-    """SSE mode: with no `progressToken` in `_meta`, `report_progress` is a no-op and the stream
-    carries only the terminal event. Spec-mandated: progress notifications are sent only against a
-    caller-supplied token."""
+async def test_sse_mode_no_notify_response_is_json() -> None:
+    """SSE mode: a handler that emits nothing (here `report_progress` is a no-op because no
+    `progressToken` was supplied) gets a plain `application/json` response. SDK-defined: SSE only
+    commits once there is something to stream."""
 
     async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
         await ctx.session.report_progress(1, total=2)
@@ -447,9 +467,9 @@ async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams |
         with anyio.fail_after(5):
             response = await http.post("/mcp", json=_list_tools_body(), headers={MCP_METHOD_HEADER: "tools/list"})
 
-    events = _sse_payloads(response.text)
-    assert len(events) == 1
-    assert "result" in events[0]
+    assert response.status_code == 200
+    assert response.headers["content-type"].split(";", 1)[0] == "application/json"
+    assert response.json()["result"]["tools"] == []
 
 
 async def test_accept_missing_sse_406_in_sse_mode() -> None:
@@ -496,7 +516,7 @@ async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams |
 async def test_late_notify_after_terminal_dropped() -> None:
     """SDK-defined: a `notify()` after the SSE sink has closed is silently dropped — the closed
     stream must not propagate as an exception out of the dispatch context."""
-    send_ch, recv_ch = anyio.create_memory_object_stream[dict[str, str]](0)
+    send_ch, recv_ch = anyio.create_memory_object_stream[bytes](0)
     dctx = _SingleExchangeDispatchContext(
         transport=TransportContext(kind="streamable-http", can_send_request=False),
         request_id=1,
@@ -556,7 +576,7 @@ async def receive() -> Message:
         await handler_started.wait()
         return {"type": "http.disconnect"}
 
-    async def send(message: Message) -> None:
+    async def send(message: Message) -> None:  # pragma: no cover
         pass
 
     with anyio.fail_after(5):