From bf242008e9ac4401de0c079b489315f1ea5b98e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=83=A7=E6=B4=97=E5=8F=91=E7=94=A8=E9=A3=98?= =?UTF-8?q?=E6=9F=94?= <2906266135@qq.com> Date: Wed, 9 Sep 2026 09:08:29 +0800 Subject: [PATCH] fix(client): clear resumption state on empty SSE event IDs --- .changeset/clear-empty-sse-event-id.md | 5 ++ packages/client/src/client/streamableHttp.ts | 6 +- .../client/test/client/streamableHttp.test.ts | 81 +++++++++++++++++++ 3 files changed, 89 insertions(+), 3 deletions(-) create mode 100644 .changeset/clear-empty-sse-event-id.md diff --git a/.changeset/clear-empty-sse-event-id.md b/.changeset/clear-empty-sse-event-id.md new file mode 100644 index 0000000000..7a061043e4 --- /dev/null +++ b/.changeset/clear-empty-sse-event-id.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +Clear the Streamable HTTP client's resumption token when an SSE event explicitly supplies an empty `id` field. Notify `onresumptiontoken` with the empty string and omit `Last-Event-ID` on subsequent GET reconnects, while preserving the previous token when the ID field is absent. An empty ID also clears POST-stream resumability so a request is not resumed without a usable token. diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index c6eaef46d8..8a427eac87 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -759,10 +759,10 @@ export class StreamableHTTPClientTransport implements Transport { } // Update last event ID if provided - if (event.id) { + if (event.id !== undefined) { lastEventId = event.id; - // Mark that we've received a priming event - stream is now resumable - hasPrimingEvent = true; + // An empty ID clears the token and cannot resume a POST stream. + hasPrimingEvent = event.id !== ''; onresumptiontoken?.(event.id); } diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index a36bbc0ad3..9a0bcc5ba7 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -1356,6 +1356,43 @@ describe('StreamableHTTPClientTransport', () => { expect(postCall).toBeDefined(); }); + it.each(['', 'id: stale-event-id\ndata: \n\n'])('does not resume a POST stream after an empty id (prefix: %j)', async prefix => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 1, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(`${prefix}id:\ndata: \n\n`)); + controller.close(); + } + }) + }); + const onresumptiontoken = vi.fn(); + const onRequestStreamEnd = vi.fn(); + + await transport.start(); + await transport.send( + { jsonrpc: '2.0', method: 'tools/call', id: 'request-1', params: { name: 'slow-tool' } }, + { onresumptiontoken, onRequestStreamEnd } + ); + await vi.advanceTimersByTimeAsync(50); + + expect(onresumptiontoken).toHaveBeenLastCalledWith(''); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(fetchMock.mock.calls[0]![1]?.method).toBe('POST'); + expect(onRequestStreamEnd).toHaveBeenCalledTimes(1); + }); + it('should NOT reconnect a POST stream when response was received', async () => { // ARRANGE transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { @@ -2462,6 +2499,50 @@ describe('StreamableHTTPClientTransport', () => { const secondCallHeaders = fetchMock.mock.calls[1]![1]?.headers; expect(secondCallHeaders?.get('last-event-id')).toBe('evt-1'); }); + + it.each([ + { idField: 'id:\n', expectedToken: '', expectedTokens: [['stale-event-id'], ['']] }, + { idField: '', expectedToken: 'stale-event-id', expectedTokens: [['stale-event-id']] } + ])('handles SSE id field "$idField" on GET reconnect', async ({ idField, expectedToken, expectedTokens }) => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 1 + } + }); + + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(`id: stale-event-id\ndata: \n\n${idField}data: \n\n`)); + controller.close(); + } + }); + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: stream + }); + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream() + }); + const onresumptiontoken = vi.fn(); + + await transport.start(); + await transport.resumeStream('initial-event-id', { onresumptiontoken }); + await vi.advanceTimersByTimeAsync(50); + + expect(onresumptiontoken.mock.calls).toEqual(expectedTokens); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(fetchMock.mock.calls[1]![1]?.headers.get('last-event-id')).toBe(expectedToken || null); + }); }); describe('Reconnection Logic with maxRetries 0', () => {