diff --git a/src/client/streamableHttp.ts b/src/client/streamableHttp.ts index 5ec6fdde8d..9dc21ee5e4 100644 --- a/src/client/streamableHttp.ts +++ b/src/client/streamableHttp.ts @@ -1,6 +1,13 @@ import { mediaTypeEssence } from '../shared/mediaType.js'; import { Transport, FetchLike, createFetchWithInit, normalizeHeaders } from '../shared/transport.js'; -import { isInitializedNotification, isJSONRPCRequest, isJSONRPCResultResponse, JSONRPCMessage, JSONRPCMessageSchema } from '../types.js'; +import { + isInitializedNotification, + isJSONRPCRequest, + isJSONRPCErrorResponse, + isJSONRPCResultResponse, + JSONRPCMessage, + JSONRPCMessageSchema +} from '../types.js'; import { auth, AuthResult, extractWWWAuthenticateParams, OAuthClientProvider, UnauthorizedError } from './auth.js'; import { EventSourceParserStream } from 'eventsource-parser/stream'; @@ -351,7 +358,7 @@ export class StreamableHTTPClientTransport implements Transport { 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) { diff --git a/test/client/streamableHttp.test.ts b/test/client/streamableHttp.test.ts index 52c8f10748..1ee1492dd8 100644 --- a/test/client/streamableHttp.test.ts +++ b/test/client/streamableHttp.test.ts @@ -941,6 +941,61 @@ describe('StreamableHTTPClientTransport', () => { expect(fetchMock.mock.calls[0][1]?.method).toBe('POST'); }); + it('should NOT reconnect a POST stream when an error response was received', async () => { + // ARRANGE + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 1, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + + // A JSON-RPC error response is a terminal response to a request, just + // like a result response, so it must also prevent reconnection. + const streamWithError = new ReadableStream({ + start(controller) { + // Priming event with ID (enables potential reconnection) + controller.enqueue(new TextEncoder().encode('id: priming-123\ndata: \n\n')); + // The terminal error response to the request + controller.enqueue( + new TextEncoder().encode( + 'id: response-456\ndata: {"jsonrpc":"2.0","error":{"code":-32000,"message":"boom"},"id":"request-1"}\n\n' + ) + ); + // Stream closes normally + controller.close(); + } + }); + + const fetchMock = global.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: streamWithError + }); + + const requestMessage: JSONRPCRequest = { + jsonrpc: '2.0', + method: 'tools/list', + id: 'request-1', + params: {} + }; + + // ACT + await transport.start(); + await transport.send(requestMessage); + await vi.advanceTimersByTimeAsync(50); + + // ASSERT + // Fetch was called ONCE only - the error response completed the request, + // so there is no need to reconnect. + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(fetchMock.mock.calls[0][1]?.method).toBe('POST'); + }); + it('should not attempt reconnection after close() is called', async () => { // ARRANGE transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {