Emit a priming SSE event on POST tool-call streams so stock MCP clients reconnect

The MCP TypeScript SDK only auto-reconnects a dropped POST SSE stream when it saw an event id before the drop. Our POST tool-call stream emitted its first and only id together with the final result, so a network blip or edge close mid-call left the SDK unable to reconnect and callTool hung forever while the server held the completed result. Write a priming SSE event as the first frame on each POST tool-call stream: a real event-store entry whose id sorts before the eventual response, framed under a non-message event type so the SDK records the id (and will reconnect) without dispatching it as a JSON-RPC message. A reconnect with last-event-id=<primingId> then replays the result.

Rhys Sullivan committed 4 months ago

Commit 89bd2ad1 parent c46730b4 files · +519 −9Browse files

Unified · Split

4 files

.changeset/mcp-priming-reconnect.md +5 −0
e2e/cloud/mcp-priming-reconnect.test.ts +218 −0
packages/hosts/cloudflare/src/mcp/agents-priming-event.test.ts +212 −0
patches/agents@0.17.3.patch +84 −9

.changeset/mcp-priming-reconnect.md+5−0

@@ -0,0 +1,5 @@
	1	+---
	2	+"executor": patch
	3	+---
	4	+
	5	+Let stock MCP clients auto-reconnect and recover a tool result when a POST stream drops mid-call.

e2e/cloud/mcp-priming-reconnect.test.ts+218−0

@@ -0,0 +1,218 @@
	1	+import {
	2	+ request as httpRequest,
	3	+ createServer,
	4	+ type ClientRequest,
	5	+ type ServerResponse,
	6	+} from "node:http";
	7	+import { randomUUID } from "node:crypto";
	8	+import type { AddressInfo } from "node:net";
	9	+
	10	+import { expect } from "@effect/vitest";
	11	+import { Effect } from "effect";
	12	+
	13	+import { Client } from "@modelcontextprotocol/sdk/client/index.js";
	14	+import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
	15	+
	16	+import { scenario } from "../src/scenario";
	17	+import { Mcp, Target } from "../src/services";
	18	+import type { Identity } from "../src/target";
	19	+
	20	+const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label;
	21	+
	22	+const delay = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));
	23	+
	24	+/**
	25	+ * A minimal HTTP relay that forwards to `origin` and, on `sever()`, destroys
	26	+ * the in-flight POST tools/call socket. This models a real network cut of the
	27	+ * POST tools/call SSE stream that the SDK observes as a stream read error, NOT
	28	+ * a client-initiated fetch abort. The relay itself stays up so the SDK's own
	29	+ * reconnect GET has something to connect back to.
	30	+ *
	31	+ * The cut is armed on a real event — the tools/call POST response headers — so
	32	+ * the trigger is identical whether or not the server emits a priming event.
	33	+ * `severPostMidCall()` waits for that stream to open, then cuts it a bounded
	34	+ * grace later: after any priming event has flushed and been read, but well
	35	+ * before the multi-second tool result.
	36	+ */
	37	+const startSeverableRelay = (
	38	+ originUrl: string,
	39	+): Promise<{
	40	+ readonly url: (path: string) => string;
	41	+ readonly severPostMidCall: () => Promise<void>;
	42	+ readonly close: () => void;
	43	+}> => {
	44	+ const origin = new URL(originUrl);
	45	+ // The tools/call POST hop, captured the moment its response headers arrive
	46	+ // (before any body: this is independent of whether the server emits a priming
	47	+ // event, so the trigger is identical on fixed and unfixed servers). The sever
	48	+ // must land AFTER the stream opened but BEFORE the tool result, so it fires a
	49	+ // bounded grace after the headers — long enough for a priming event to have
	50	+ // flushed and been read by the SDK, far short of the multi-second tool call.
	51	+ const SEVER_GRACE_MS = 2_000;
	52	+ let toolCallHop: { readonly upstream: ClientRequest; readonly res: ServerResponse } \| null = null;
	53	+ let onToolCallOpen: (() => void) \| null = null;
	54	+ const toolCallOpened = new Promise<void>((resolve) => {
	55	+ onToolCallOpen = resolve;
	56	+ });
	57	+
	58	+ const server = createServer((req, res) => {
	59	+ // Detect the tools/call POST by its request body (initialize also returns
	60	+ // SSE, so a plain content-type check would arm on the wrong stream).
	61	+ const bodyChunks: Buffer[] = [];
	62	+ let isToolCall = false;
	63	+ req.on("data", (chunk: Buffer) => {
	64	+ bodyChunks.push(chunk);
	65	+ if (!isToolCall && Buffer.concat(bodyChunks).toString("utf8").includes('"tools/call"')) {
	66	+ isToolCall = true;
	67	+ }
	68	+ });
	69	+ const upstream = httpRequest(
	70	+ {
	71	+ host: origin.hostname,
	72	+ port: Number(origin.port),
	73	+ path: req.url,
	74	+ method: req.method,
	75	+ headers: { ...req.headers, host: origin.host },
	76	+ },
	77	+ (upstreamRes) => {
	78	+ res.writeHead(upstreamRes.statusCode ?? 502, upstreamRes.headers);
	79	+ if (
	80	+ isToolCall &&
	81	+ (upstreamRes.headers["content-type"] ?? "").includes("text/event-stream")
	82	+ ) {
	83	+ toolCallHop = { upstream, res };
	84	+ onToolCallOpen?.();
	85	+ }
	86	+ upstreamRes.pipe(res);
	87	+ },
	88	+ );
	89	+ upstream.on("error", () => {
	90	+ try {
	91	+ res.destroy();
	92	+ } catch {
	93	+ /* already torn down */
	94	+ }
	95	+ });
	96	+ req.pipe(upstream);
	97	+ req.on("aborted", () => {
	98	+ try {
	99	+ upstream.destroy();
	100	+ } catch {
	101	+ /* already torn down */
	102	+ }
	103	+ });
	104	+ });
	105	+
	106	+ const severPostMidCall = async (): Promise<void> => {
	107	+ await toolCallOpened;
	108	+ await delay(SEVER_GRACE_MS);
	109	+ const hop = toolCallHop;
	110	+ if (!hop) return;
	111	+ // Destroy BOTH ends so every fetch client (undici and bun) observes a hard
	112	+ // stream error: the client-facing response socket (what the SDK's reader is
	113	+ // attached to) AND the upstream to the origin.
	114	+ for (const destroy of [
	115	+ () => hop.res.socket?.destroy(),
	116	+ () => hop.res.destroy(),
	117	+ () => hop.upstream.socket?.destroy(),
	118	+ () => hop.upstream.destroy(),
	119	+ ]) {
	120	+ try {
	121	+ destroy();
	122	+ } catch {
	123	+ /* already gone */
	124	+ }
	125	+ }
	126	+ };
	127	+
	128	+ return new Promise((resolve) => {
	129	+ server.listen(0, "127.0.0.1", () => {
	130	+ const port = (server.address() as AddressInfo).port;
	131	+ resolve({
	132	+ url: (path: string) => `http://127.0.0.1:${port}${path}`,
	133	+ severPostMidCall,
	134	+ close: () => server.close(),
	135	+ });
	136	+ });
	137	+ });
	138	+};
	139	+
	140	+scenario(
	141	+ "MCP streamable HTTP · a stock SDK client recovers a tool result after its POST stream is severed mid-call",
	142	+ { timeout: 160_000 },
	143	+ Effect.gen(function* () {
	144	+ const target = yield* Target;
	145	+ const mcp = yield* Mcp;
	146	+ const identity = yield* target.newIdentity();
	147	+ const bearer = yield* mcp.mintBearer(emailOf(identity));
	148	+
	149	+ const mcpPath = new URL(target.mcpUrl).pathname;
	150	+ const origin = new URL(target.mcpUrl).origin;
	151	+ const relay = yield* Effect.promise(() => startSeverableRelay(origin));
	152	+
	153	+ // A fetch wrapper for the SDK transport that (1) records the SDK's own
	154	+ // reconnect GET (with its last-event-id) so the test can assert the SDK
	155	+ // actually reconnected, and (2) STRIPS the SDK's abort signal so a severed
	156	+ // relay socket surfaces as a stream read error (the real churn scenario),
	157	+ // not a client-side abort. Auth rides on the transport's requestInit.
	158	+ const reconnectGetEventIds: string[] = [];
	159	+ const drivenFetch: typeof fetch = (input, init) => {
	160	+ const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url;
	161	+ const method = init?.method ?? (input instanceof Request ? input.method : "GET");
	162	+ const headers = new Headers(
	163	+ init?.headers ?? (input instanceof Request ? input.headers : undefined),
	164	+ );
	165	+ const lastEventId = headers.get("last-event-id");
	166	+ if (method === "GET" && lastEventId) reconnectGetEventIds.push(lastEventId);
	167	+ const { signal: _stripped, ...rest } = init ?? {};
	168	+ return fetch(url, rest);
	169	+ };
	170	+
	171	+ const transport = new StreamableHTTPClientTransport(new URL(relay.url(mcpPath)), {
	172	+ requestInit: { headers: { authorization: `Bearer ${bearer}` } },
	173	+ fetch: drivenFetch,
	174	+ });
	175	+ const client = new Client({ name: "executor-e2e-priming-reconnect", version: "0.0.1" });
	176	+
	177	+ const result = yield* Effect.promise(async () => {
	178	+ try {
	179	+ await client.connect(transport);
	180	+
	181	+ const marker = `MARKER_PRIMING_RECONNECT_${randomUUID()}`;
	182	+ const code = [
	183	+ `const marker = ${JSON.stringify(marker)};`,
	184	+ "await new Promise((resolve) => setTimeout(resolve, 15000));",
	185	+ "return marker;",
	186	+ ].join("\n");
	187	+
	188	+ // Arm the sever: the relay cuts the POST tools/call socket a short
	189	+ // grace after the stream opens (see SEVER_GRACE_MS) — after any priming
	190	+ // event has flushed and been read by the SDK, but ~13s before the tool
	191	+ // result. The trigger (stream open) is identical on a fixed and an
	192	+ // unfixed server, so the ONLY variable is whether the client saw a
	193	+ // priming id to reconnect from.
	194	+ void relay.severPostMidCall();
	195	+ // Explicit request timeout > (call duration + reconnect backoff) so a
	196	+ // successful recover resolves well inside it; the scenario timeout is
	197	+ // the real ceiling.
	198	+ const call = await client.callTool({ name: "execute", arguments: { code } }, undefined, {
	199	+ timeout: 90_000,
	200	+ });
	201	+ return { marker, call, reconnected: reconnectGetEventIds.length > 0 };
	202	+ } finally {
	203	+ await client.close().catch(() => undefined);
	204	+ await delay(200);
	205	+ relay.close();
	206	+ }
	207	+ });
	208	+
	209	+ expect(
	210	+ reconnectGetEventIds.length,
	211	+ "the stock SDK issued a reconnect GET carrying a last-event-id (it had a priming id to resume from)",
	212	+ ).toBeGreaterThan(0);
	213	+ expect(
	214	+ JSON.stringify(result.call),
	215	+ "callTool resolves with the completed tool result after the mid-call sever",
	216	+ ).toContain(result.marker);
	217	+ }),
	218	+);

packages/hosts/cloudflare/src/mcp/agents-priming-event.test.ts+212−0

@@ -0,0 +1,212 @@
	1	+// Unit coverage for the POST-stream priming SSE event (see
	2	+// patches/agents@0.17.3.patch). Executor's POST tools/call stream used to emit
	3	+// its first and only event `id:` together with the final result, so the MCP TS
	4	+// SDK's StreamableHTTPClientTransport never set hasPrimingEvent and would not
	5	+// auto-reconnect a stream that dropped mid-call: callTool hung while the DO
	6	+// held the completed result. The patch writes a priming event as the first
	7	+// frame on the POST stream, carrying a real event-store id that sorts before
	8	+// the response so a `last-event-id: <primingId>` reconnect replays the result.
	9	+//
	10	+// Two properties are pinned here against real code:
	11	+// 1. Event-store ordering + replay: with the real DurableObjectEventStore, a
	12	+// priming event stored before the response sorts first, and
	13	+// replayEventsAfter(primingId) yields exactly the response.
	14	+// 2. Client contract: fed the exact priming frame the transport emits, the
	15	+// real SDK StreamableHTTPClientTransport records the priming id (so it
	16	+// would reconnect) WITHOUT dispatching it as a JSON-RPC message, then
	17	+// dispatches a following result frame normally.
	18	+import { describe, expect, it } from "@effect/vitest";
	19	+import { DurableObjectEventStore } from "agents/mcp";
	20	+import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
	21	+import type { JSONRPCMessage } from "@modelcontextprotocol/sdk/types.js";
	22	+
	23	+type ListOptions = {
	24	+ readonly prefix?: string;
	25	+ readonly start?: string;
	26	+ readonly limit?: number;
	27	+ readonly reverse?: boolean;
	28	+};
	29	+
	30	+/** Minimal in-memory stand-in for DurableObjectStorage's sorted KV surface. */
	31	+const makeFakeStorage = () => {
	32	+ const entries = new Map<string, unknown>();
	33	+ return {
	34	+ entries,
	35	+ put: (key: string, value: unknown) => {
	36	+ entries.set(key, value);
	37	+ return Promise.resolve();
	38	+ },
	39	+ delete: (keys: string \| ReadonlyArray<string>) => {
	40	+ for (const key of Array.isArray(keys) ? keys : [keys]) entries.delete(key);
	41	+ return Promise.resolve();
	42	+ },
	43	+ list: (options: ListOptions = {}) => {
	44	+ const keys = [...entries.keys()]
	45	+ .filter((key) => (options.prefix === undefined ? true : key.startsWith(options.prefix)))
	46	+ .filter((key) => (options.start === undefined ? true : key >= options.start))
	47	+ .sort();
	48	+ if (options.reverse === true) keys.reverse();
	49	+ const limited = options.limit === undefined ? keys : keys.slice(0, options.limit);
	50	+ return Promise.resolve(new Map(limited.map((key) => [key, entries.get(key)])));
	51	+ },
	52	+ };
	53	+};
	54	+
	55	+// The exact priming notification the patched transport persists, and the exact
	56	+// SSE frame it writes to the client. Mirrors emitPrimingEvent /
	57	+// writePrimingSSEEvent in patches/agents@0.17.3.patch: a benign JSON-RPC
	58	+// notification stored (so a plain-GET replay via writeSSEEvent is ignorable),
	59	+// framed live under a non-`message` event type so the SDK primes but does not
	60	+// dispatch it.
	61	+const PRIMING_MESSAGE = {
	62	+ jsonrpc: "2.0" as const,
	63	+ method: "notifications/message",
	64	+ params: { level: "debug", data: "mcp-stream-priming" },
	65	+};
	66	+const primingFrame = (eventId: string): string =>
	67	+ `event: mcp-priming\nid: ${eventId}\ndata: ${JSON.stringify(PRIMING_MESSAGE)}\n\n`;
	68	+const messageFrame = (eventId: string, message: unknown): string =>
	69	+ `event: message\nid: ${eventId}\ndata: ${JSON.stringify(message)}\n\n`;
	70	+
	71	+describe("POST-stream priming event: store ordering and replay", () => {
	72	+ it("persists the priming event before the response so replayEventsAfter(primingId) yields the response", async () => {
	73	+ const storage = makeFakeStorage();
	74	+ const store = new DurableObjectEventStore(storage as never);
	75	+ const streamId = "post-stream";
	76	+
	77	+ // Transport order: priming event first, then the tool response.
	78	+ const primingId = await store.storeEvent(streamId, PRIMING_MESSAGE);
	79	+ const response = {
	80	+ jsonrpc: "2.0" as const,
	81	+ id: 1,
	82	+ result: { content: [{ type: "text", text: "MARKER" }] },
	83	+ };
	84	+ const responseId = await store.storeEvent(streamId, response);
	85	+
	86	+ expect(primingId, "priming event is seq 1 for the stream").toBe(`${streamId}:0000000000000001`);
	87	+ expect(responseId, "response is seq 2, after the priming id").toBe(
	88	+ `${streamId}:0000000000000002`,
	89	+ );
	90	+ expect(primingId < responseId, "priming id sorts strictly before the response id").toBe(true);
	91	+
	92	+ const replayed: Array<{ readonly eventId: string; readonly message: unknown }> = [];
	93	+ await store.replayEventsAfter(primingId, {
	94	+ send: async (eventId: string, message: unknown) => {
	95	+ replayed.push({ eventId, message });
	96	+ },
	97	+ });
	98	+
	99	+ expect(
	100	+ replayed.map((entry) => entry.eventId),
	101	+ "a reconnect with last-event-id=<primingId> replays exactly the response",
	102	+ ).toEqual([responseId]);
	103	+ expect(replayed[0]?.message).toEqual(response);
	104	+ });
	105	+
	106	+ it("does not replay the priming event itself on a last-event-id reconnect", async () => {
	107	+ const storage = makeFakeStorage();
	108	+ const store = new DurableObjectEventStore(storage as never);
	109	+ const streamId = "post-stream";
	110	+ const primingId = await store.storeEvent(streamId, PRIMING_MESSAGE);
	111	+
	112	+ const replayed: string[] = [];
	113	+ await store.replayEventsAfter(primingId, {
	114	+ send: async (eventId: string) => {
	115	+ replayed.push(eventId);
	116	+ },
	117	+ });
	118	+
	119	+ expect(replayed, "nothing after the priming event yet, so replay is empty").toEqual([]);
	120	+ });
	121	+});
	122	+
	123	+describe("POST-stream priming event: SDK client contract", () => {
	124	+ // Drive the real SDK StreamableHTTPClientTransport with a controlled fetch
	125	+ // that returns a POST tools/call SSE stream: priming frame first, then the
	126	+ // result. Assert the SDK records the priming id as a resumption token (so it
	127	+ // would reconnect) but only dispatches the result as a JSON-RPC message.
	128	+ const drivePostStream = async (frames: ReadonlyArray<string>) => {
	129	+ const encoder = new TextEncoder();
	130	+ const body = new ReadableStream<Uint8Array>({
	131	+ start(controller) {
	132	+ for (const frame of frames) controller.enqueue(encoder.encode(frame));
	133	+ controller.close();
	134	+ },
	135	+ });
	136	+ // oxlint-disable-next-line executor/no-double-cast -- boundary: a minimal fetch stub for a unit test; only the Response shape the SDK reads matters.
	137	+ const fetchStub = (async () =>
	138	+ new Response(body, {
	139	+ status: 200,
	140	+ headers: { "content-type": "text/event-stream" },
	141	+ })) as unknown as typeof fetch;
	142	+
	143	+ const transport = new StreamableHTTPClientTransport(new URL("https://executor.sh/mcp"), {
	144	+ fetch: fetchStub,
	145	+ });
	146	+
	147	+ const messages: JSONRPCMessage[] = [];
	148	+ const resumptionTokens: string[] = [];
	149	+ transport.onmessage = (message) => {
	150	+ messages.push(message);
	151	+ };
	152	+
	153	+ await transport.start();
	154	+ // POST a tools/call request; the stub returns the SSE stream above.
	155	+ await transport.send(
	156	+ { jsonrpc: "2.0", id: 1, method: "tools/call", params: { name: "execute", arguments: {} } },
	157	+ {
	158	+ onresumptiontoken: (token: string) => {
	159	+ resumptionTokens.push(token);
	160	+ },
	161	+ },
	162	+ );
	163	+ // Let the SSE stream drain.
	164	+ for (let i = 0; i < 20; i += 1) await Promise.resolve();
	165	+ await transport.close();
	166	+ return { messages, resumptionTokens };
	167	+ };
	168	+
	169	+ it("records the priming id as a resumption token without dispatching it, then dispatches the result", async () => {
	170	+ const primingId = "post-stream:0000000000000001";
	171	+ const responseId = "post-stream:0000000000000002";
	172	+ const result = {
	173	+ jsonrpc: "2.0" as const,
	174	+ id: 1,
	175	+ result: { content: [{ type: "text", text: "MARKER" }] },
	176	+ };
	177	+
	178	+ const { messages, resumptionTokens } = await drivePostStream([
	179	+ primingFrame(primingId),
	180	+ messageFrame(responseId, result),
	181	+ ]);
	182	+
	183	+ expect(
	184	+ resumptionTokens,
	185	+ "the SDK records the priming id first (this is what sets hasPrimingEvent), then the response id",
	186	+ ).toEqual([primingId, responseId]);
	187	+ expect(
	188	+ messages,
	189	+ "the priming frame is NOT dispatched as a JSON-RPC message; only the result is",
	190	+ ).toEqual([result]);
	191	+ });
	192	+
	193	+ it("would not prime on a stream whose first event id arrives only with the result (the old behavior)", async () => {
	194	+ // Sanity anchor for the fix: without a priming frame, the first recorded
	195	+ // resumption token is the result's own id, which the SDK only sees at the
	196	+ // same instant it receives the result. There is no earlier id to reconnect
	197	+ // from, which is exactly the hang this patch removes.
	198	+ const responseId = "post-stream:0000000000000001";
	199	+ const result = {
	200	+ jsonrpc: "2.0" as const,
	201	+ id: 1,
	202	+ result: { content: [{ type: "text", text: "MARKER" }] },
	203	+ };
	204	+
	205	+ const { messages, resumptionTokens } = await drivePostStream([
	206	+ messageFrame(responseId, result),
	207	+ ]);
	208	+
	209	+ expect(resumptionTokens, "the only id ever seen is the result's own id").toEqual([responseId]);
	210	+ expect(messages).toEqual([result]);
	211	+ });
	212	+});

patches/agents@0.17.3.patch+84−9

@@ -19,7 +19,7 @@ index c8fad448e8797b89690a99d93490d1363851b225..77f9fe3f6f2375eadc9f7a2974f0d202 · 18 unmodified lines
19	19	McpAgent,
20	20	type McpAuthContext,
21	21	diff --git a/dist/mcp/index.js b/dist/mcp/index.js
22		−index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..b0d1c55eda79324b6d76a1177519d26a62be2e41 100644
	22	+index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..c3601bdb084bf82ebfea23cdabe1fab5f6d71fb3 100644
23	23	--- a/dist/mcp/index.js
24	24	+++ b/dist/mcp/index.js
25	25	@@ -28,13 +28,17 @@ import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/
@@ -577,7 +577,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..b0d1c55eda79324b6d76a1177519d26a · 551 unmodified lines
577	577	}
578	578	/**
579	579	* Writes an event to the SSE stream with proper formatting
580		−@@ -689,6 +977,8 @@ var StreamableHTTPServerTransport = class {
	580	+@@ -689,10 +977,65 @@ var StreamableHTTPServerTransport = class {
581	581	return connection.send(JSON.stringify({
582	582	type: "cf_mcp_agent_event",
583	583	event: eventData,
@@ -586,7 +586,82 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..b0d1c55eda79324b6d76a1177519d26a · 2 unmodified lines
586	586	close
587	587	}));
588	588	}
589		−@@ -777,9 +1067,11 @@ var StreamableHTTPServerTransport = class {
	589	+ /**
	590	++ * Persist and write the priming SSE event for a freshly opened POST
	591	++ * stream. The stored message is a benign JSON-RPC notification so that
	592	++ * if a plain recovery GET ever replays this stream via
	593	++ * {@link writeSSEEvent} (which forces `event: message`), a compliant
	594	++ * client parses and ignores it rather than erroring. The live priming
	595	++ * frame, however, uses a non-`message` SSE event type so the SDK
	596	++ * records the id (setting hasPrimingEvent / lastEventId) WITHOUT
	597	++ * dispatching it as a JSON-RPC message: its SSE loop skips any event
	598	++ * whose type is not `message`. A data line is required because the
	599	++ * SDK's SSE parser drops events with empty data before recording the
	600	++ * id, so an id-only frame would not prime.
	601	++ */
	602	++ async emitPrimingEvent(agent, connection, streamId) {
	603	++ if (!this._eventStore) return;
	604	++ const primingMessage = {
	605	++ jsonrpc: "2.0",
	606	++ method: "notifications/message",
	607	++ params: {
	608	++ level: "debug",
	609	++ data: "mcp-stream-priming"
	610	++ }
	611	++ };
	612	++ let eventId;
	613	++ try {
	614	++ eventId = await this._eventStore.storeEvent(streamId, primingMessage);
	615	++ } catch (error) {
	616	++ this.onerror?.(error);
	617	++ return;
	618	++ }
	619	++ if (!eventId) return;
	620	++ try {
	621	++ this.writePrimingSSEEvent(connection, primingMessage, eventId);
	622	++ } catch (error) {
	623	++ this.onerror?.(error);
	624	++ }
	625	++ }
	626	++ /**
	627	++ * Write a priming SSE frame: `event: mcp-priming`, an `id:` carrying a
	628	++ * real replayable event-store id, and a data line the SDK ignores
	629	++ * because the event type is not `message`. Never sets `close`.
	630	++ */
	631	++ writePrimingSSEEvent(connection, message, eventId) {
	632	++ let eventData = "event: mcp-priming\n";
	633	++ eventData += `id: ${eventId}\n`;
	634	++ eventData += `data: ${JSON.stringify(message)}\n\n`;
	635	++ return connection.send(JSON.stringify({
	636	++ type: "cf_mcp_agent_event",
	637	++ event: eventData,
	638	++ eventId,
	639	++ streamId: eventId.slice(0, eventId.lastIndexOf(":"))
	640	++ }));
	641	++ }
	642	++ /**
	643	+ * Handles POST requests containing JSON-RPC messages
	644	+ */
	645	+ async handlePostRequest(req, parsedBody) {
	646	+@@ -733,6 +1076,17 @@ var StreamableHTTPServerTransport = class {
	647	+ };
	648	+ connection.setState(postState);
	649	+ if (this._eventStore) await agent.setStreamRequestIds(streamId, requestIds);
	650	++ // Emit a priming SSE event as the very first frame on this POST
	651	++ // response stream, before dispatching the request(s). The MCP TS SDK
	652	++ // only auto-reconnects a dropped POST SSE stream when it saw an event
	653	++ // `id:` before the drop (hasPrimingEvent). Without this, a network
	654	++ // blip or edge close mid-call leaves the SDK with hasPrimingEvent
	655	++ // false, so it never issues the recovery GET that replays the
	656	++ // persisted result, and callTool hangs forever. The priming event is
	657	++ // a real event-store entry so its id sorts BEFORE the eventual
	658	++ // response; a reconnect with `last-event-id: <primingId>` replays
	659	++ // everything after it (i.e. the result). See patches/agents patch.
	660	++ await this.emitPrimingEvent(agent, connection, streamId);
	661	+ for (const message of messages) {
	662	+ if (this.messageInterceptor) {
	663	+ if (await this.messageInterceptor(message, {
	664	+@@ -777,9 +1131,11 @@ var StreamableHTTPServerTransport = class {
590	665	} catch (error) {
591	666	this.onerror?.(error);
592	667	}
@@ -600,7 +675,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..b0d1c55eda79324b6d76a1177519d26a · 7 unmodified lines
600	675	}
601	676	}
602	677	async send(message, options) {
603		−@@ -861,12 +1153,10 @@ var StreamableHTTPServerTransport = class {
	678	+@@ -861,12 +1217,10 @@ var StreamableHTTPServerTransport = class {
604	679	*
605	680	* ## Lifecycle
606	681	*
@@ -617,7 +692,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..b0d1c55eda79324b6d76a1177519d26a · 10 unmodified lines
617	692	*
618	693	* Standalone GET stream events (`_GET_stream`) are not cleared
619	694	* automatically; they accumulate for the lifetime of the DO. Bounded
620		−@@ -899,6 +1189,7 @@ var DurableObjectEventStore = class DurableObjectEventStore {
	695	+@@ -899,6 +1253,7 @@ var DurableObjectEventStore = class DurableObjectEventStore {
621	696	const eventId = `${streamId}:${seq.toString(16).padStart(DurableObjectEventStore.SEQ_PAD, "0")}`;
622	697	const eventKey = `${DurableObjectEventStore.EVENT_KEY_PREFIX}${eventId}`;
623	698	await this.storage.put(eventKey, message);
@@ -625,7 +700,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..b0d1c55eda79324b6d76a1177519d26a · 1 unmodified line
625	700	return eventId;
626	701	}
627	702	async getStreamIdForEventId(eventId) {
628		−@@ -915,9 +1206,59 @@ var DurableObjectEventStore = class DurableObjectEventStore {
	703	+@@ -915,9 +1270,59 @@ var DurableObjectEventStore = class DurableObjectEventStore {
629	704	start: startKey,
630	705	limit: DurableObjectEventStore.REPLAY_LIMIT
631	706	});
@@ -686,7 +761,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..b0d1c55eda79324b6d76a1177519d26a · 54 unmodified lines
686	761	/**
687	762	* Drop the event log for a single stream. Called by the transport
688	763	* immediately after a POST's final response has been written to the
689		−@@ -973,6 +1314,9 @@ DurableObjectEventStore.EVENT_KEY_PREFIX = "__mcp_event__:";
	764	+@@ -973,6 +1378,9 @@ DurableObjectEventStore.EVENT_KEY_PREFIX = "__mcp_event__:";
690	765	DurableObjectEventStore.SEQ_PAD = 16;
691	766	DurableObjectEventStore.DELETE_CHUNK = 128;
692	767	DurableObjectEventStore.REPLAY_LIMIT = 1e3;
@@ -696,7 +771,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..b0d1c55eda79324b6d76a1177519d26a · 3 unmodified lines
696	771	//#endregion
697	772	//#region src/mcp/client-transports.ts
698	773	/**
699		−@@ -1381,6 +1725,47 @@ var McpAgent = class McpAgent extends Agent {
	774	+@@ -1381,6 +1789,47 @@ var McpAgent = class McpAgent extends Agent {
700	775	async deleteStreamRequestIds(streamId) {
701	776	await this.ctx.storage.delete(`${McpAgent.STREAM_REQS_KEY_PREFIX}${streamId}`);
702	777	}
@@ -744,7 +819,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..b0d1c55eda79324b6d76a1177519d26a · 41 unmodified lines
744	819	/**
745	820	* Reverse lookup: find which POST stream a given `requestId` belongs
746	821	* to, and return the stream's full `requestIds` list in the same
747		−@@ -1697,7 +2082,8 @@ var McpAgent = class McpAgent extends Agent {
	822	+@@ -1697,7 +2146,8 @@ var McpAgent = class McpAgent extends Agent {
748	823	}
749	824	};
750	825	McpAgent.STREAM_REQS_KEY_PREFIX = "__mcp_stream_reqs__:";