Skip to content

Buffer replay before network delivery and preserve the live handoff #3541

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Kludex wants to merge 1 commit into
base: preserve-request-stream-ownership
Choose a base branch
from
+496 −18
Open

Buffer replay before network delivery and preserve the live handoff #3541

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions docs/run/legacy-clients.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,14 @@ On one worker that is invisible. On two, it is the whole problem: a request that
events to a client reconnecting to the *same* session), not a session store. It never makes a
session reachable from another process.

!!! note "Replay is buffered before network delivery"
With `event_store=`, the SDK collects replayed events before sending them, so a slow replay
reader does not hold the event-store lock. The buffer spills to a temporary file above
1 MiB instead of retaining the whole history in memory. Historical events, any new resumption
cursor, and live events are sent in that order. The lock does not serialize incoming POSTs
or reserve JSON-RPC request IDs. Live-stream backpressure can still delay other messages in
the same session.

!!! note "Request cleanup preserves newer streams"
Closing an HTTP request releases its own streams, including during cancellation.
If you reuse a JSON-RPC request ID after the previous request completes, cleanup from an
Expand Down
82 changes: 65 additions & 17 deletions src/mcp/server/streamable_http.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
responses, with streaming support for long-running operations.
"""

import json
import logging
import math
import re
Expand Down Expand Up @@ -210,11 +211,13 @@ def __init__(
self.mcp_session_id = mcp_session_id
self.is_json_response_enabled = is_json_response_enabled
self._event_store = event_store
self._event_store_lock = anyio.Lock()
self._security = TransportSecurityMiddleware(security_settings)
self._retry_interval = retry_interval
self._request_streams: dict[RequestId, _RequestStreams] = {}
self._sse_stream_writers: dict[RequestId, MemoryObjectSendStream[SSEEvent]] = {}
self._terminated = False
self._connected = False
self._idle_timeout = idle_timeout
self._requests_in_flight = 0
self.idle_scope: anyio.CancelScope | None = None
Expand Down Expand Up @@ -915,25 +918,43 @@ async def _replay_events(self, last_event_id: str, request: Request, send: Send)
async def replay_sender():
stream_id: StreamId | None = None
request_streams: _RequestStreams | None = None
priming_event: SSEEvent | None = None
replay_buffer = anyio.SpooledTemporaryFile(max_size=1024 * 1024)
try:
async with sse_stream_writer:

async def send_event(event_message: EventMessage) -> None:
await sse_stream_writer.send(self._create_event_data(event_message))

stream_id = await event_store.replay_events_after(last_event_id, send_event)
if stream_id and stream_id not in self._request_streams: # pragma: no branch
request_streams = anyio.create_memory_object_stream[EventMessage](
REQUEST_STREAM_BUFFER_SIZE
event_data = self._create_event_data(event_message)
# Whole records prevent ping frames from splitting a replayed SSE event.
await replay_buffer.write(
(json.dumps(event_data, ensure_ascii=False) + "\n").encode("utf-8")
Comment on lines +929 to +930

@claude claude Bot Sep 18, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 (optional) Servers whose event store builds EventMessage with a non-JSON event id (a uuid.UUID, Decimal, DB id object) lose replay after this merges; the base delivered those events. streamable_http.py:930 runs json.dumps on the event dict, whose id is EventMessage.event_id exactly as the store passed it. json.dumps raises TypeError for such objects; the handler at streamable_http.py:961 logs it and the client gets an empty 200 stream and retries the same cursor forever. Fix: coerce event_id to str in _create_event_data (streamable_http.py:437) or pass default=str, so the live and replay paths accept the same ids. Third-party stores cannot be enumerated from this checkout; the SDK's own stores all pass str.

Extended reasoning...

EventMessage is a plain dataclass (streamable_http.py:106-111) and its event_id annotation is not enforced; stores construct it themselves inside replay_events_after. On the base, send_event handed the dict straight to sse_starlette, which renders the id with string formatting, so a UUID object, a Decimal or a driver row id worked on the wire; the same objects still work on the live path after the merge because _create_event_data at line 428-439 only copies them. On the replay path after the merge: a store such as a Postgres-backed one whose driver returns uuid.UUID for the id column calls send_callback(EventMessage(message, row.id)). send_event at streamable_http.py:926-931 builds event_data with id set to that UUID object and calls json.dumps(event_data, ensure_ascii=False) at line 930, which raises TypeError: Object of type UUID is not JSON serializable. The exception leaves replay_events_after, exits the lock, and is caught by except Exception at line 961, which logs Error in replay sender; the finally closes the buffer. The async with sse_stream_writer block ends, so the…

Verification: nit. Trigger: a third-party EventStore whose replay_events_after builds EventMessage with a non-str event_id (uuid.UUID, Decimal, DB row id), violating the declared but runtime-unenforced type. Mechanism verified: src/mcp/server/streamable_http.py:106-111 declares event_id: str | None on a plain dataclass (no runtime check); _create_event_data (lines 428-439) copies event_message.event_id…

)
Comment on lines +929 to 931

@claude claude Bot Sep 18, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 Operators whose server has no writable temp directory lose Last-Event-ID resumption for any stream holding over 1 MiB of history; on the base branch that history streamed fine. The buffer write at src/mcp/server/streamable_http.py:929 goes into anyio.SpooledTemporaryFile, whose spill opens tempfile.TemporaryFile in gettempdir() from a worker thread. The OSError escapes send_event and aborts replay_events_after, so the client gets an empty SSE stream on every retry. Fix: keep replay deliverable without a writable temp dir, e.g. catch OSError from the spill and keep buffering in memory (or make the spill dir/threshold configurable).

Extended reasoning...

Condition: the process runs with no writable temp location (Docker --read-only without --tmpfs /tmp, Kubernetes readOnlyRootFilesystem with no emptyDir at /tmp, TMPDIR set to a missing path, or a full disk) and a resumed stream has more than 1 MiB of un-acknowledged history, e.g. a long tool call emitting many progress or log notifications while the client was disconnected. On the base branch replay_sender sent each replayed event straight to sse_stream_writer with no filesystem involvement. After the change, send_event at streamable_http.py:926-931 writes every record into replay_buffer created at line 922 with max_size=1024*1024. anyio's SpooledTemporaryFile checks tell() against max_size on write and calls rollover(), which runs tempfile.TemporaryFile(mode='w+b', dir=None, ...) via to_thread.run_sync; tempfile.gettempdir() raises FileNotFoundError('No usable temporary directory found') when TMPDIR, /tmp, /var/tmp, /usr/tmp and cwd are all unwritable, and a full disk raises OSError(ENOSPC) on the copy. That exception propagates out of send_event, through the store's…

Verification: normal — triggered when the server process has no writable temp location (read-only root filesystem without a /tmp tmpfs/emptyDir, TMPDIR pointing at a missing path, or ENOSPC) and a resumed stream carries more than 1 MiB of replayable history. Mechanism verified in /home/claude/python-sdk/src/mcp/server/streamable_http.py: line 922 creates `replay_buffer =… | normal — triggers when the process…

self._sse_stream_writers[stream_id] = sse_stream_writer
self._request_streams[stream_id] = request_streams
priming_event = await self._mint_priming_event(stream_id, replay_protocol_version)
if priming_event is not None:
await sse_stream_writer.send(priming_event)
async with request_streams[1] as msg_reader:
async for event_message in msg_reader:
await sse_stream_writer.send(self._create_event_data(event_message))

async with self._event_store_lock:

@cubic-dev-ai cubic-dev-ai Bot Sep 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: When a POST mints its priming cursor during a replay, _handle_post_request bypasses this lock and can mutate the store while replay_events_after iterates. Guard the POST priming store_event with the same lock; the deque-backed store can otherwise abort resumable replay.

Prompt for AI agents 
Check if this issue is valid — if so, understand the root cause and fix it. At src/mcp/server/streamable_http.py, line 933:

<comment>When a POST mints its priming cursor during a replay, `_handle_post_request` bypasses this lock and can mutate the store while `replay_events_after` iterates. Guard the POST priming `store_event` with the same lock; the deque-backed store can otherwise abort resumable replay.</comment>

<file context>
@@ -915,25 +918,43 @@ async def _replay_events(self, last_event_id: str, request: Request, send: Send)
-                                async for event_message in msg_reader:
-                                    await sse_stream_writer.send(self._create_event_data(event_message))
+
+                        async with self._event_store_lock:
+                            if self._terminated or not self._connected:
+                                return
</file context>

if self._terminated or not self._connected:
return
stream_id = await event_store.replay_events_after(last_event_id, send_event)
Comment on lines +933 to +936

@claude claude Bot Sep 18, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 (optional) Every message in a session now waits while one client's replay reads the event store, which the base branch never did. streamable_http.py:936 calls replay_events_after while holding _event_store_lock; the router then blocks at streamable_http.py:1074 before it can store or route anything. write_stream has no buffer, so every server handler's send in that session waits for the whole store read. Fix: hold the lock only to fix the snapshot boundary and register the live stream, and read history unlocked (e.g. record the last stored id under the lock and replay up to it outside it); this cost follows from the PR's serialize-under-lock purpose. The docs note names only live-stream backpressure. [also at: src/mcp/server/streamable_http.py:948 - Sessions with histories over 1 MiB hold _event_store_lock, and so block every message in the session, for a worker-thread hop per replayed event plus a 1 MiB copy; the base held no lock and did no thread work. streamable_http.py:929 writes each record through anyio.SpooledTemporaryFile while…]

Extended reasoning...

The PR text says the lock lets a slow replay reader not block sibling responses; that covers the network reader, not the store read itself, which now runs entirely under the lock. On the base the replay called replay_events_after with no lock and the router only serialized on its own store_event call, so a long replay never delayed other messages. After the merge: a client reconnects with Last-Event-ID to a session backed by a database or remote store holding a long history. replay_sender enters async with self._event_store_lock at streamable_http.py:933 and awaits event_store.replay_events_after at line 936; that call awaits the store once per historical row, so the lock is held for the full read. Meanwhile a tool handler for another request in the same session finishes and sends its response on write_stream (a 0-buffer stream created…

Verification: nit; conflicts with stated purpose: the PR description says the lock is held so that "a slow reader cannot block sibling responses", but that only covers network delivery — the event-store read itself now runs under the lock and stalls every outbound message in the session for its full duration. Trigger: a client reconnects with Last-Event-ID to a session whose EventStore is slow… | nit —…

if self._terminated or not self._connected:
return
if stream_id and stream_id not in self._request_streams:
request_streams = anyio.create_memory_object_stream[EventMessage](
REQUEST_STREAM_BUFFER_SIZE
)
self._sse_stream_writers[stream_id] = sse_stream_writer
self._request_streams[stream_id] = request_streams
priming_event = await self._mint_priming_event(stream_id, replay_protocol_version)

await replay_buffer.seek(0)
while event_data := await replay_buffer.readline():
await sse_stream_writer.send(json.loads(event_data))
Comment on lines +943 to +949

@claude claude Bot Sep 18, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 (optional) Servers using the polling pattern (close_sse_stream from a tool) now cut a resuming client's history mid-delivery and force it to reconnect and re-scan; on the base the replay always completed. streamable_http.py:943 publishes the writer in _sse_stream_writers before the buffered history is drained at lines 947-949, so a close during the drain closes the writer and the next send raises. Fix: make a close request during history delivery take effect only after the buffered history and priming event are sent (or keep the writer unregistered until the drain completes, as the base did).

Extended reasoning...

Tools that implement polling call ctx.close_sse_stream() periodically. A client reconnects with Last-Event-ID and has a large history (the spilled case is >1 MiB). Under the lock at line 943 the new code stores the writer in _sse_stream_writers, then releases the lock and sends the history record by record at line 949. The tool's next close_sse_stream(request_id) at line 273 pops and closes that writer and pops the request streams at line 279. The next send at line 949 raises ClosedResourceError, caught at line 958; the stream ends after a partial history with no priming event. The client waits the retry interval, reconnects, and the remaining history is scanned again under _event_store_lock, stalling the session router again; with a close interval shorter than the drain time this repeats for every chunk. On the base the writer was registered only after replay_events_after finished (old line 928), so close during replay was a no-op and the history arrived in one pass. No event is lost, but each resume becomes many round trips and many locked store scans. Remedy: defer the close…

Verification: nit. Trigger: a server using the polling pattern (a tool calling ctx.close_sse_stream() periodically, e.g. examples/servers/sse-polling-demo/mcp_sse_polling_demo/server.py:104-106) while a client resumes with Last-Event-ID and its buffered history is still being drained. Mechanism verified: src/mcp/server/streamable_http.py:943-944 registers sse_stream_writer in… | nit. Trigger: a server tool…

await replay_buffer.aclose()
if request_streams is None or self._terminated or not self._connected:
return
if priming_event is not None:
await sse_stream_writer.send(priming_event)
async with request_streams[1] as msg_reader:
async for event_message in msg_reader:
await sse_stream_writer.send(self._create_event_data(event_message))
except anyio.ClosedResourceError: # pragma: lax no cover
# Expected when close_sse_stream() is called
logger.debug("Replay SSE stream closed by close_sse_stream()")
Comment on lines +947 to 960

@claude claude Bot Sep 18, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 (optional) Clients resuming against an event store that fails partway through a long scan now receive nothing and can never advance, where the base delivered everything up to the failure. streamable_http.py:936 buffers all events under the lock and only sends at line 947 after replay_events_after returns; an exception there is caught at line 961 and the buffered records are discarded. Fix: on a mid-replay store error, still deliver the events already buffered (then end the stream) so the client's cursor advances and each retry scans less, or document that stores must complete the whole scan atomically.

Extended reasoning...

A store backed by a database or Redis scan with a query timeout trips only on long scans. On the base, send_event wrote each event straight to the SSE writer, so a timeout after N events still delivered N events; the client reconnected with the Nth id and the next scan was shorter and succeeded. Now send_event at lines 926-931 only writes to replay_buffer, and the buffer is read back at lines 947-949 only after replay_events_after returns normally. If the store raises, control jumps to line 961, logs, and the finally closes the buffer; the SSE response ends with zero events. The client reconnects with the same Last-Event-ID, the same long scan runs again under _event_store_lock (stalling the session's router each time), times out again, and the client never progresses. The dismissing finder read the code but judged only 'partial delivery differs'; it did not follow the retry loop where partial delivery is what made progress possible. Remedy: deliver the already-buffered records before ending the stream on a store error.

Verification: nit — triggers when a user-supplied EventStore.replay_events_after raises after having invoked the callback for some events (any backend error mid-scan; permanent loss only if the failure is deterministic for that scan length). Mechanism verified in /home/claude/python-sdk/src/mcp/server/streamable_http.py: send_event (lines 926-931) now only does await replay_buffer.write(...); the… | nit…

Expand All @@ -945,6 +966,8 @@ async def send_event(event_message: EventMessage) -> None:
if request_streams is not None:
assert stream_id is not None
self._clean_up_memory_streams(stream_id, request_streams)
with anyio.CancelScope(shield=True):
await replay_buffer.aclose()

# Create and start EventSourceResponse
response = EventSourceResponse(
Expand Down Expand Up @@ -999,10 +1022,12 @@ async def connect(
self._write_stream_reader = write_stream_reader
self._write_stream = write_stream

acquire_scope: anyio.CancelScope | None = None
# Start a task group for message routing
async with anyio.create_task_group() as tg:
# Create a message router that distributes messages to request streams
async def message_router():
nonlocal acquire_scope
try:
async for session_message in write_stream_reader: # pragma: no branch
# Determine which request stream(s) should receive this message
Expand Down Expand Up @@ -1038,10 +1063,28 @@ async def message_router():
# messages will be replayed on the re-connect
event_id = None
if self._event_store:
event_id = await self._event_store.store_event(request_stream_id, message)
logger.debug(f"Stored {event_id} from {request_stream_id}")
with anyio.CancelScope() as lock_scope:
acquire_scope = lock_scope
try:
try:
self._event_store_lock.acquire_nowait()
except anyio.WouldBlock:
if not self._connected:
return
await self._event_store_lock.acquire()
finally:
acquire_scope = None
if lock_scope.cancelled_caught:
return
try:
event_id = await self._event_store.store_event(request_stream_id, message)
logger.debug(f"Stored {event_id} from {request_stream_id}")
target = self._request_streams.get(request_stream_id)
finally:
self._event_store_lock.release()
else:
target = self._request_streams.get(request_stream_id)

target = self._request_streams.get(request_stream_id)
if target is not None:
try:
# Send both the message and the event ID
Comment on lines 1063 to 1090

@claude claude Bot Sep 18, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 (optional) Every other request in a session stalls while one reconnecting client drains a large replay, which the base branch never did during history delivery. streamable_http.py:944 registers the resumed stream under the lock before the buffered history is sent at :947-:949; live events for that stream now queue into the 16-slot buffer during the drain, and on the 17th the router blocks at :1091, so no response or notification in the session moves until the slow client finishes reading history. Fix: keep the live stream unregistered (or drain it) until history delivery completes, so the router only blocks on a stream that is actually being consumed; the base registered at the same point but only after the history was already on the wire. [also at: src/mcp/server/streamable_http.py:944 - Every other request in a session can stall for as long as one resuming client takes to download its history, which the base branch never did during replay. streamable_http.py:944 registers the resumed stream in _request_streams before any buffered history is sent, so live messages for that stream…]

Extended reasoning...

On the base, replay_events_after streamed history straight to the network and only afterwards registered request_streams, so live events for that stream during history delivery hit target None at :1094 and were stored without queuing. The PR moves registration to :944, inside the lock and before the drain loop at :947-:949. A client reconnects with Last-Event-ID to a GET stream (or a long-running request stream) whose history is large, up to and above the 1 MiB spill, over a slow link. During the drain the server keeps emitting notifications for that stream (progress, logging, list_changed). Each goes through the router: store_event, then target[0].send at :1091 into the 16-slot buffer created at :940. The 17th send blocks the router. The router is the only consumer of write_stream for the whole session,…

Verification: nit — acknowledged in diff: docs/run/legacy-clients.md adds "Live-stream backpressure can still delay other messages in the same session", and that bound is accurate (a delay for the remainder of the drain, not a deadlock); but the PR description's claim that delivering history outside the lock means "a slow reader cannot block sibling responses" is only true for the lock, not for the…

Expand All @@ -1066,8 +1109,10 @@ async def message_router():
tg.start_soon(message_router)

try:
self._connected = True
yield read_stream, write_stream
finally:
self._connected = False
for stream_id, streams in list(self._request_streams.items()):
self._clean_up_memory_streams(stream_id, streams)

Expand All @@ -1080,3 +1125,6 @@ async def message_router():
except Exception as e: # pragma: no cover
# During cleanup, we catch all exceptions since streams might be in various states
logger.debug(f"Error closing streams: {e}")
# Cancel only lock acquisition, not a store write that has already started.
if acquire_scope is not None:
acquire_scope.cancel()
Loading
Loading