diff --git a/packages/mcp/src/transports/streamable-http.ts b/packages/mcp/src/transports/streamable-http.ts index 9ecdb2973..85e3f116e 100644 --- a/packages/mcp/src/transports/streamable-http.ts +++ b/packages/mcp/src/transports/streamable-http.ts @@ -64,8 +64,7 @@ export async function consumeSseStream(stream: ReadableStream, optio dataBytes = 0; }; - const processLine = (rawLine: string) => { - const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; + const processLine = (line: string) => { if (line === "") { dispatch(); return; @@ -86,21 +85,45 @@ export async function consumeSseStream(stream: ReadableStream, optio } else if (field === "retry" && /^\d+$/.test(value)) options.onRetry?.(Number(value)); }; + // SSE lines end with CRLF, LF, or a lone CR. + let skipLeadingLf = false; + const consumeLines = () => { + let start = 0; + for (let i = 0; i < buffered.length; i++) { + const char = buffered[i]; + if (char !== "\r" && char !== "\n") continue; + processLine(buffered.slice(start, i)); + if (char === "\r") { + if (i + 1 < buffered.length) { + if (buffered[i + 1] === "\n") i++; + } else { + skipLeadingLf = true; + } + } + start = i + 1; + } + buffered = buffered.slice(start); + }; + try { while (true) { const { done, value } = await reader.read(); if (done) break; - buffered += decoder.decode(value, { stream: true }); - let newline = buffered.indexOf("\n"); - while (newline >= 0) { - processLine(buffered.slice(0, newline)); - buffered = buffered.slice(newline + 1); - newline = buffered.indexOf("\n"); + let text = decoder.decode(value, { stream: true }); + // A CRLF split across chunks: the CR already ended the line, so drop the LF. + if (skipLeadingLf && text.length > 0) { + if (text.startsWith("\n")) text = text.slice(1); + skipLeadingLf = false; } + buffered += text; + consumeLines(); if (Buffer.byteLength(buffered) > maxEventBytes) throw new Error(`MCP SSE event exceeds ${maxEventBytes} bytes`); } - buffered += decoder.decode(); + let tail = decoder.decode(); + if (skipLeadingLf && tail.startsWith("\n")) tail = tail.slice(1); + buffered += tail; + consumeLines(); if (buffered) processLine(buffered); dispatch(); } catch (error) { diff --git a/packages/mcp/test/streamable-http.test.ts b/packages/mcp/test/streamable-http.test.ts index bfef0210e..0b3656199 100644 --- a/packages/mcp/test/streamable-http.test.ts +++ b/packages/mcp/test/streamable-http.test.ts @@ -104,6 +104,45 @@ describe("consumeSseStream", () => { expect(events).toEqual([{ id: "7", data: '{"one":\n1}' }]); }); + it("recognizes CR, LF, and CRLF line endings, including CRLF split across chunks", async () => { + const encoder = new TextEncoder(); + const parse = async (chunks: string[]) => { + const stream = new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(encoder.encode(chunk)); + controller.close(); + }, + }); + const events: SseEvent[] = []; + await consumeSseStream(stream, { onEvent: (event) => events.push(event) }); + return events; + }; + const expected = [{ id: "1", data: "a\nb" }, { data: "c" }]; + for (const separator of ["\n", "\r\n", "\r"]) { + const text = ["id: 1", "data: a", "data: b", "", "data: c", "", ""].join(separator); + expect(await parse([text])).toEqual(expected); + // One character per chunk splits every CRLF between chunks. + expect(await parse([...text])).toEqual(expected); + } + expect(await parse(["id: 1\rdata: a\r\ndata: b\n\r", "\ndata: c\r\r"])).toEqual(expected); + }); + + it("dispatches CR-terminated events while the stream stays open", async () => { + const encoder = new TextEncoder(); + let controller!: ReadableStreamDefaultController; + const stream = new ReadableStream({ + start(c) { + controller = c; + }, + }); + const received = new Promise((resolve) => { + void consumeSseStream(stream, { onEvent: resolve }).catch(() => {}); + }); + controller.enqueue(encoder.encode("data: first\r\r")); + expect(await received).toEqual({ data: "first" }); + controller.close(); + }); + it("rejects events whose data lines exceed the limit without a blank line", async () => { const encoder = new TextEncoder(); let sent = 0;