diff --git a/packages/opencode/test/mcp/transport.test.ts b/packages/opencode/test/mcp/transport.test.ts new file mode 100644 index 000000000000..f7a74d4d7299 --- /dev/null +++ b/packages/opencode/test/mcp/transport.test.ts @@ -0,0 +1,38 @@ +import { expect, test } from "bun:test" +import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js" + +test("does not reconnect an SSE stream after a JSON-RPC error response", async () => { + let requests = 0 + const transport = new StreamableHTTPClientTransport(new URL("http://mcp.invalid"), { + fetch: async () => { + requests += 1 + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("id: prime\nretry: 1\ndata:\n\n")) + controller.enqueue( + new TextEncoder().encode( + 'id: error\ndata: {"jsonrpc":"2.0","error":{"code":-32601,"message":"Method not found"},"id":1}\n\n', + ), + ) + controller.close() + }, + }), + { status: 200, headers: { "content-type": "text/event-stream" } }, + ) + }, + reconnectionOptions: { + initialReconnectionDelay: 1, + maxReconnectionDelay: 1, + reconnectionDelayGrowFactor: 1, + maxRetries: 2, + }, + }) + + await transport.start() + await transport.send({ jsonrpc: "2.0", method: "resources/list", id: 1 }) + await Bun.sleep(25) + await transport.close() + + expect(requests).toBe(1) +}) diff --git a/patches/@modelcontextprotocol%2Fsdk@1.29.0.patch b/patches/@modelcontextprotocol%2Fsdk@1.29.0.patch index 13b8000a0139..3b70bea55868 100644 --- a/patches/@modelcontextprotocol%2Fsdk@1.29.0.patch +++ b/patches/@modelcontextprotocol%2Fsdk@1.29.0.patch @@ -141,6 +141,15 @@ diff --git a/dist/cjs/client/streamableHttp.js b/dist/cjs/client/streamableHttp. index a29a7d3a0f14d9cd800ef5b296485237350c666f..c362ae5fe6c62c8c8eae7e2e61de1eedff5443c9 100644 --- a/dist/cjs/client/streamableHttp.js +++ b/dist/cjs/client/streamableHttp.js +@@ -204,7 +204,7 @@ class StreamableHTTPClientTransport { + if (!event.event || event.event === 'message') { + try { + const message = types_js_1.JSONRPCMessageSchema.parse(JSON.parse(event.data)); +- if ((0, types_js_1.isJSONRPCResultResponse)(message)) { ++ if ((0, types_js_1.isJSONRPCResultResponse)(message) || (0, types_js_1.isJSONRPCErrorResponse)(message)) { + // Mark that we received a response - no need to reconnect for this request + receivedResponse = true; + if (replayMessageId !== undefined) { @@ -290,7 +290,38 @@ class StreamableHTTPClientTransport { this.onclose?.(); } @@ -518,10 +527,19 @@ index 624172aa24ae255a67c083f9c19053343e4a0581..ac75b14545fda44aff7ff4d97cc5da88 @@ -1,5 +1,5 @@ import { createFetchWithInit, normalizeHeaders } from '../shared/transport.js'; -import { isInitializedNotification, isJSONRPCRequest, isJSONRPCResultResponse, JSONRPCMessageSchema } from '../types.js'; -+import { isInitializedNotification, isInitializeRequest, isJSONRPCRequest, isJSONRPCResultResponse, JSONRPCMessageSchema } from '../types.js'; ++import { isInitializedNotification, isInitializeRequest, isJSONRPCErrorResponse, isJSONRPCRequest, isJSONRPCResultResponse, JSONRPCMessageSchema } from '../types.js'; import { auth, extractWWWAuthenticateParams, UnauthorizedError } from './auth.js'; import { EventSourceParserStream } from 'eventsource-parser/stream'; // Default reconnection options for StreamableHTTP connections +@@ -200,7 +200,7 @@ export class StreamableHTTPClientTransport { + if (!event.event || event.event === 'message') { + try { + const message = JSONRPCMessageSchema.parse(JSON.parse(event.data)); +- if (isJSONRPCResultResponse(message)) { ++ if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) { + // Mark that we received a response - no need to reconnect for this request + receivedResponse = true; + if (replayMessageId !== undefined) { @@ -286,7 +286,38 @@ export class StreamableHTTPClientTransport { this.onclose?.(); }