Store priming event before request dispatch so storage order == wire order

Splits the old _maybe_send_priming_event into _mint_priming_event (store +
return wire dict) and _run_sse_writer (forward request_stream onto the wire).
The POST handler now awaits _mint_priming_event before writer.send(), so the
priming row is in the event store before the server can produce any message
for that request id — ordering by data dependency, not scheduler timing.

The replay path keeps its priming event (test_streamable_http_multiple_reconnections
relies on it as a stream-re-registered signal); its replay→live-tail ordering
window is pre-existing and orthogonal.

Also extracts the inline sse_writer closure to a method (drops _handle_post_request
below the C901 threshold) and widens the SSE-dict stream type to SSEEvent
(dict[str, Any]) — the previous dict[str, str] was a lie masked by the old
helper's Any parameter, since priming events carry retry: int.

Author: maxisbey

src/mcp/server/streamable_http.py

@@ -12,6 +12,7 @@
 from collections.abc import AsyncGenerator, Awaitable, Callable
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
+from functools import partial
 from http import HTTPStatus
 from typing import Any, Final
 
@@ -71,6 +72,8 @@
 # Type aliases
 StreamId = str
 EventId = str
+# An SSE event-dict as accepted by sse-starlette (`event`, `data`, `id`, `retry`).
+SSEEvent = dict[str, Any]
 
 
 @dataclass
@@ -174,7 +177,7 @@ def __init__(
                 MemoryObjectReceiveStream[EventMessage],
             ],
         ] = {}
-        self._sse_stream_writers: dict[RequestId, MemoryObjectSendStream[dict[str, str]]] = {}
+        self._sse_stream_writers: dict[RequestId, MemoryObjectSendStream[SSEEvent]] = {}
         self._terminated = False
         # Idle timeout cancel scope; managed by the session manager.
         self.idle_scope: anyio.CancelScope | None = None
@@ -261,31 +264,48 @@ async def close_standalone_stream_callback() -> None:
 
         return SessionMessage(message, metadata=metadata)
 
-    async def _maybe_send_priming_event(
-        self,
-        request_id: RequestId,
-        sse_stream_writer: MemoryObjectSendStream[dict[str, Any]],
-        protocol_version: str,
-    ) -> None:
-        """Send priming event for SSE resumability if event_store is configured.
+    async def _mint_priming_event(self, stream_id: StreamId, protocol_version: str) -> SSEEvent | None:
+        """Store the priming cursor for `stream_id` and return its SSE wire form.
 
-        Only sends priming events to clients with protocol version >= 2025-11-25,
-        which includes the fix for handling empty SSE data. Older clients would
-        crash trying to parse empty data as JSON.
+        Called before the request is dispatched so the priming row precedes
+        anything `message_router` can store for this stream. Returns `None`
+        when no event store is configured or the client predates 2025-11-25
+        (older clients cannot parse the empty-data event).
         """
         if not self._event_store:
-            return
-        # Priming events have empty data which older clients cannot handle.
+            return None
         if not is_version_at_least(protocol_version, "2025-11-25"):
-            return
-        priming_event_id = await self._event_store.store_event(
-            str(request_id),  # Convert RequestId to StreamId (str)
-            None,  # Priming event has no payload
-        )
-        priming_event: dict[str, str | int] = {"id": priming_event_id, "data": ""}
+            return None
+        priming_event_id = await self._event_store.store_event(stream_id, None)
+        priming_event: SSEEvent = {"id": priming_event_id, "data": ""}
         if self._retry_interval is not None:
             priming_event["retry"] = self._retry_interval
-        await sse_stream_writer.send(priming_event)
+        return priming_event
+
+    async def _run_sse_writer(
+        self,
+        request_id: RequestId,
+        sse_stream_writer: MemoryObjectSendStream[SSEEvent],
+        request_stream_reader: MemoryObjectReceiveStream[EventMessage],
+        priming_event: SSEEvent | None,
+    ) -> None:
+        """Forward `_request_streams[request_id]` onto the SSE wire for one POST."""
+        try:
+            async with sse_stream_writer, request_stream_reader:
+                if priming_event is not None:
+                    await sse_stream_writer.send(priming_event)
+                async for event_message in request_stream_reader:
+                    await sse_stream_writer.send(self._create_event_data(event_message))
+                    if isinstance(event_message.message, JSONRPCResponse | JSONRPCError):
+                        break
+        except anyio.ClosedResourceError:  # pragma: lax no cover
+            logger.debug("SSE stream closed by close_sse_stream()")
+        except Exception:  # pragma: lax no cover
+            logger.exception("Error in SSE writer")
+        finally:
+            logger.debug("Closing SSE writer")
+            self._sse_stream_writers.pop(request_id, None)
+            await self._clean_up_memory_streams(request_id)
 
     def _create_error_response(
         self,
@@ -339,7 +359,7 @@ def _get_session_id(self, request: Request) -> str | None:
         """Extract the session ID from request headers."""
         return request.headers.get(MCP_SESSION_ID_HEADER)
 
-    def _create_event_data(self, event_message: EventMessage) -> dict[str, str]:
+    def _create_event_data(self, event_message: EventMessage) -> SSEEvent:
         """Create event data dictionary from an EventMessage."""
         event_data = {
             "event": "message",
@@ -579,40 +599,16 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                     await self._clean_up_memory_streams(request_id)
             else:
                 # Create SSE stream
-                sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[dict[str, str]](0)
+                sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[SSEEvent](0)
 
                 # Store writer reference so close_sse_stream() can close it
                 self._sse_stream_writers[request_id] = sse_stream_writer
 
-                async def sse_writer():
-                    # Get the request ID from the incoming request message
-                    try:
-                        async with sse_stream_writer, request_stream_reader:
-                            # Send priming event for SSE resumability
-                            await self._maybe_send_priming_event(request_id, sse_stream_writer, protocol_version)
-
-                            # Process messages from the request-specific stream
-                            async for event_message in request_stream_reader:
-                                # Build the event data
-                                event_data = self._create_event_data(event_message)
-                                await sse_stream_writer.send(event_data)
-
-                                # If response, remove from pending streams and close
-                                if isinstance(event_message.message, JSONRPCResponse | JSONRPCError):
-                                    break
-                    except anyio.ClosedResourceError:  # pragma: lax no cover
-                        # Expected when close_sse_stream() is called
-                        logger.debug("SSE stream closed by close_sse_stream()")
-                    except Exception:  # pragma: lax no cover
-                        logger.exception("Error in SSE writer")
-                    finally:
-                        logger.debug("Closing SSE writer")
-                        self._sse_stream_writers.pop(request_id, None)
-                        await self._clean_up_memory_streams(request_id)
-
-                # Create and start EventSourceResponse
-                # SSE stream mode (original behavior)
-                # Set up headers
+                # Store the priming event before the request is dispatched so its
+                # event-store position precedes anything message_router can store
+                # for this id (storage order == wire order by construction).
+                priming_event = await self._mint_priming_event(request_id, protocol_version)
+
                 headers = {
                     "Cache-Control": "no-cache, no-transform",
                     "Connection": "keep-alive",
@@ -621,7 +617,9 @@ async def sse_writer():
                 }
                 response = EventSourceResponse(
                     content=sse_stream_reader,
-                    data_sender_callable=sse_writer,
+                    data_sender_callable=partial(
+                        self._run_sse_writer, request_id, sse_stream_writer, request_stream_reader, priming_event
+                    ),
                     headers=headers,
                 )
 
@@ -704,7 +702,7 @@ async def _handle_get_request(self, request: Request, send: Send) -> None:
             return
 
         # Create SSE stream
-        sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[dict[str, str]](0)
+        sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[SSEEvent](0)
 
         async def standalone_sse_writer():
             try:
@@ -880,7 +878,7 @@ async def _replay_events(self, last_event_id: str, request: Request, send: Send)
             replay_protocol_version = request.headers.get(MCP_PROTOCOL_VERSION_HEADER, DEFAULT_NEGOTIATED_VERSION)
 
             # Create SSE stream for replay
-            sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[dict[str, str]](0)
+            sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[SSEEvent](0)
 
             async def replay_sender():
                 try:
@@ -898,8 +896,12 @@ async def send_event(event_message: EventMessage) -> None:
                             # Register SSE writer so close_sse_stream() can close it
                             self._sse_stream_writers[stream_id] = sse_stream_writer
 
-                            # Send priming event for this new connection
-                            await self._maybe_send_priming_event(stream_id, sse_stream_writer, replay_protocol_version)
+                            # Prime the resumed connection so the client sees the stream
+                            # is re-registered. The replay→live-tail ordering window here
+                            # is pre-existing and tracked separately.
+                            priming_event = await self._mint_priming_event(stream_id, replay_protocol_version)
+                            if priming_event is not None:  # pragma: no branch
+                                await sse_stream_writer.send(priming_event)
 
                             # Create new request streams for this connection
                             self._request_streams[stream_id] = anyio.create_memory_object_stream[EventMessage](

tests/server/test_streamable_http_router.py

@@ -5,11 +5,27 @@
 
 from mcp.server.streamable_http import (
     REQUEST_STREAM_BUFFER_SIZE,
+    EventCallback,
+    EventId,
     EventMessage,
+    EventStore,
     StreamableHTTPServerTransport,
+    StreamId,
 )
 from mcp.shared.message import SessionMessage
-from mcp.types import JSONRPCResponse
+from mcp.types import JSONRPCMessage, JSONRPCResponse
+
+
+class _OrderTrackingStore(EventStore):
+    def __init__(self) -> None:
+        self.stored: list[tuple[StreamId, JSONRPCMessage | None]] = []
+
+    async def store_event(self, stream_id: StreamId, message: JSONRPCMessage | None) -> EventId:
+        self.stored.append((stream_id, message))
+        return str(len(self.stored))
+
+    async def replay_events_after(self, last_event_id: EventId, send_callback: EventCallback) -> StreamId | None:
+        raise NotImplementedError
 
 
 @pytest.mark.anyio
@@ -49,3 +65,37 @@ async def server_writes() -> None:
             assert a_send.statistics().current_buffer_used == 1
             await a_recv.aclose()
             await a_send.aclose()
+
+
+@pytest.mark.anyio
+async def test_priming_event_is_stored_before_any_routed_message() -> None:
+    """`_mint_priming_event` is awaited before the request is dispatched, so the
+    priming row precedes every `message_router` store for that stream regardless
+    of when `sse_writer` is scheduled.
+    """
+    store = _OrderTrackingStore()
+    transport = StreamableHTTPServerTransport(mcp_session_id="sid", is_json_response_enabled=False, event_store=store)
+    streams = transport._request_streams
+
+    async with transport.connect() as (_read_stream, write_stream):
+        # POST handler step: mint priming for "A" before dispatch.
+        priming = await transport._mint_priming_event("A", "2025-11-25")
+        assert priming is not None
+        streams["A"] = anyio.create_memory_object_stream[EventMessage](REQUEST_STREAM_BUFFER_SIZE)
+        a_send, a_recv = streams["A"]
+
+        # Server emits 5 messages for "A" with no sse_writer scheduled. Each
+        # write_stream.send() rendezvous-hands to message_router, which stores
+        # then deposits into A's buffer; reading them back proves the router
+        # has finished storing.
+        for i in range(5):
+            await write_stream.send(SessionMessage(JSONRPCResponse(jsonrpc="2.0", id="A", result={"n": i})))
+        with anyio.fail_after(5):
+            for _ in range(5):
+                await a_recv.receive()
+        await a_recv.aclose()
+        await a_send.aclose()
+
+    assert store.stored[0] == ("A", None)
+    assert [sid for sid, _ in store.stored] == ["A"] * 6
+    assert all(msg is not None for _, msg in store.stored[1:])

tests/shared/test_streamable_http.py

@@ -1636,80 +1636,30 @@ async def test_handle_sse_event_skips_empty_data() -> None:
 
 @pytest.mark.anyio
 async def test_priming_event_not_sent_for_old_protocol_version() -> None:
-    """_maybe_send_priming_event skips for old protocol versions (backwards compat)."""
-    # Create a transport with an event store
-    transport = StreamableHTTPServerTransport(
-        "/mcp",
-        event_store=SimpleEventStore(),
-    )
-
-    # Create a mock stream writer
-    write_stream, read_stream = anyio.create_memory_object_stream[dict[str, Any]](1)
-
-    try:
-        # Call _maybe_send_priming_event with OLD protocol version - should NOT send
-        await transport._maybe_send_priming_event("test-request-id", write_stream, "2025-06-18")
-
-        # Nothing should have been written to the stream
-        assert write_stream.statistics().current_buffer_used == 0
-
-        # Now test with NEW protocol version - should send
-        await transport._maybe_send_priming_event("test-request-id-2", write_stream, "2025-11-25")
-
-        # Should have written a priming event
-        assert write_stream.statistics().current_buffer_used == 1
-    finally:
-        await write_stream.aclose()
-        await read_stream.aclose()
+    """`_mint_priming_event` skips for old protocol versions (backwards compat)."""
+    transport = StreamableHTTPServerTransport("/mcp", event_store=SimpleEventStore())
+    assert await transport._mint_priming_event("test-request-id", "2025-06-18") is None
+    assert await transport._mint_priming_event("test-request-id-2", "2025-11-25") is not None
 
 
 @pytest.mark.anyio
 async def test_priming_event_not_sent_without_event_store() -> None:
-    """_maybe_send_priming_event returns early when no event_store is configured."""
-    # Create a transport WITHOUT an event store
+    """`_mint_priming_event` returns `None` when no event_store is configured."""
     transport = StreamableHTTPServerTransport("/mcp")
-
-    # Create a mock stream writer
-    write_stream, read_stream = anyio.create_memory_object_stream[dict[str, Any]](1)
-
-    try:
-        # Call _maybe_send_priming_event - should return early without sending
-        await transport._maybe_send_priming_event("test-request-id", write_stream, "2025-11-25")
-
-        # Nothing should have been written to the stream
-        assert write_stream.statistics().current_buffer_used == 0
-    finally:
-        await write_stream.aclose()
-        await read_stream.aclose()
+    assert await transport._mint_priming_event("test-request-id", "2025-11-25") is None
 
 
 @pytest.mark.anyio
 async def test_priming_event_includes_retry_interval() -> None:
-    """_maybe_send_priming_event includes the retry field when retry_interval is set."""
-    # Create a transport with an event store AND retry_interval
+    """`_mint_priming_event` includes the retry field when `retry_interval` is set."""
     transport = StreamableHTTPServerTransport(
         "/mcp",
         event_store=SimpleEventStore(),
         retry_interval=5000,
     )
-
-    # Create a mock stream writer
-    write_stream, read_stream = anyio.create_memory_object_stream[dict[str, Any]](1)
-
-    try:
-        # Call _maybe_send_priming_event with new protocol version
-        await transport._maybe_send_priming_event("test-request-id", write_stream, "2025-11-25")
-
-        # Should have written a priming event with retry field
-        assert write_stream.statistics().current_buffer_used == 1
-
-        # Read the event and verify it has retry field
-        event = await read_stream.receive()
-        assert "retry" in event
-        assert event["retry"] == 5000
-    finally:
-        await write_stream.aclose()
-        await read_stream.aclose()
+    event = await transport._mint_priming_event("test-request-id", "2025-11-25")
+    assert event is not None
+    assert event["retry"] == 5000
 
 
 @pytest.mark.anyio
@@ -1746,26 +1696,13 @@ async def test_close_sse_stream_callback_not_provided_for_old_protocol_version()
 
 @pytest.mark.anyio
 async def test_priming_event_not_sent_for_unknown_protocol_version() -> None:
-    """_maybe_send_priming_event treats unrecognized version strings conservatively.
+    """`_mint_priming_event` treats unrecognized version strings conservatively.
 
     A garbage version must not be mistaken for a future one (lexicographically
     "zzz" sorts after every date-shaped revision).
     """
-    transport = StreamableHTTPServerTransport(
-        "/mcp",
-        event_store=SimpleEventStore(),
-    )
-
-    write_stream, read_stream = anyio.create_memory_object_stream[dict[str, Any]](1)
-
-    try:
-        await transport._maybe_send_priming_event("test-request-id", write_stream, "zzz")
-
-        # Nothing should have been written to the stream
-        assert write_stream.statistics().current_buffer_used == 0
-    finally:
-        await write_stream.aclose()
-        await read_stream.aclose()
+    transport = StreamableHTTPServerTransport("/mcp", event_store=SimpleEventStore())
+    assert await transport._mint_priming_event("test-request-id", "zzz") is None
 
 
 @pytest.mark.anyio