Fix: Handle error responses in Streamable HTTP SSE streams (#1390)

Co-authored-by: Felix Weinberger <3823880+felixweinberger@users.noreply.github.com>
This commit is contained in:
Anthony 2026-03-31 12:53:52 +02:00 • committed by GitHub
commit 9bc9abc68b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 80 additions and 1 deletions

View file

@ -0,0 +1,5 @@
---
'@modelcontextprotocol/client': patch
---
Fix StreamableHTTPClientTransport to handle error responses in SSE streams

View file

@ -4,6 +4,7 @@ import type { FetchLike, JSONRPCMessage, Transport } from '@modelcontextprotocol
import {
createFetchWithInit,
isInitializedNotification,
isJSONRPCErrorResponse,
isJSONRPCRequest,
isJSONRPCResultResponse,
JSONRPCMessageSchema,
@ -412,7 +413,8 @@ export class StreamableHTTPClientTransport implements Transport {
if (!event.event || event.event === 'message') {
try {
const message = JSONRPCMessageSchema.parse(JSON.parse(event.data));
if (isJSONRPCResultResponse(message)) {
// Handle both success AND error responses for completion detection and ID remapping
if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) {
// Mark that we received a response - no need to reconnect for this request
receivedResponse = true;
if (replayMessageId !== undefined) {

View file

@ -1008,6 +1008,78 @@ describe('StreamableHTTPClientTransport', () => {
expect(fetchMock.mock.calls[0]![1]?.method).toBe('POST');
});
it('should NOT reconnect a POST stream when error response was received', async () => {
// ARRANGE
transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
reconnectionOptions: {
initialReconnectionDelay: 10,
maxRetries: 1,
maxReconnectionDelay: 1000,
reconnectionDelayGrowFactor: 1
}
});
const messageSpy = vi.fn();
transport.onmessage = messageSpy;
// Create a stream that sends:
// 1. Priming event with ID (enables potential reconnection)
// 2. An error response (should also prevent reconnection, just like success)
// 3. Then closes
const streamWithErrorResponse = new ReadableStream({
start(controller) {
// Priming event with ID
controller.enqueue(new TextEncoder().encode('id: priming-123\ndata: \n\n'));
// An error response to the request (tool not found, for example)
controller.enqueue(
new TextEncoder().encode(
'id: error-456\ndata: {"jsonrpc":"2.0","error":{"code":-32602,"message":"Tool not found"},"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: streamWithErrorResponse
});
const requestMessage: JSONRPCRequest = {
jsonrpc: '2.0',
method: 'tools/call',
id: 'request-1',
params: { name: 'nonexistent-tool' }
};
// ACT
await transport.start();
await transport.send(requestMessage);
await vi.advanceTimersByTimeAsync(50);
// ASSERT
// THE KEY ASSERTION: Fetch was called ONCE only - no reconnection!
// The error response was received, so no need to reconnect.
expect(fetchMock).toHaveBeenCalledTimes(1);
expect(fetchMock.mock.calls[0]![1]?.method).toBe('POST');
// Verify the error response was delivered to the message handler
expect(messageSpy).toHaveBeenCalledWith(
expect.objectContaining({
jsonrpc: '2.0',
error: expect.objectContaining({
code: -32602,
message: 'Tool not found'
}),
id: 'request-1'
})
);
});
it('should not attempt reconnection after close() is called', async () => {
// ARRANGE
transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {