Skip to content

Commit 91c515f

Browse files
committed
fix: fail fast after SSE reconnect exhaustion
1 parent 5fc42e9 commit 91c515f

3 files changed

Lines changed: 82 additions & 17 deletions

File tree

.changeset/quiet-cups-retry.md

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"@modelcontextprotocol/client": patch
3+
---
4+
5+
Make StreamableHTTPClientTransport retry standalone SSE streams longer and fail fast after reconnection exhaustion instead of allowing later request responses to disappear behind a dead SSE channel. SSE open failures now include a fallback HTTP status when statusText is empty.

packages/client/src/client/streamableHttp.ts

Lines changed: 51 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -23,9 +23,34 @@ const DEFAULT_STREAMABLE_HTTP_RECONNECTION_OPTIONS: StreamableHTTPReconnectionOp
2323
initialReconnectionDelay: 1000,
2424
maxReconnectionDelay: 30_000,
2525
reconnectionDelayGrowFactor: 1.5,
26-
maxRetries: 2
26+
maxRetries: 10
2727
};
2828

29+
function errorMessage(error: unknown): string {
30+
if (error instanceof Error && error.message) {
31+
return error.message;
32+
}
33+
34+
if (typeof error === 'object' && error) {
35+
const maybeError = error as { name?: string; status?: number; statusText?: string };
36+
if (maybeError.statusText) {
37+
return maybeError.statusText;
38+
}
39+
if (maybeError.status !== undefined) {
40+
return `HTTP ${maybeError.status}`;
41+
}
42+
if (maybeError.name) {
43+
return maybeError.name;
44+
}
45+
}
46+
47+
if (typeof error === 'string' && error) {
48+
return error;
49+
}
50+
51+
return 'unknown';
52+
}
53+
2954
/**
3055
* Options for starting or authenticating an SSE connection
3156
*/
@@ -75,7 +100,7 @@ export interface StreamableHTTPReconnectionOptions {
75100

76101
/**
77102
* Maximum number of reconnection attempts before giving up.
78-
* Default is 2.
103+
* Default is 10.
79104
*/
80105
maxRetries: number;
81106
}
@@ -185,6 +210,7 @@ export class StreamableHTTPClientTransport implements Transport {
185210
private _serverRetryMs?: number; // Server-provided retry delay from SSE retry field
186211
private readonly _reconnectionScheduler?: ReconnectionScheduler;
187212
private _cancelReconnection?: () => void;
213+
private _standaloneSseReconnectError?: Error;
188214

189215
onclose?: () => void;
190216
onerror?: (error: Error) => void;
@@ -290,12 +316,14 @@ export class StreamableHTTPClientTransport implements Transport {
290316
return;
291317
}
292318

293-
throw new SdkHttpError(SdkErrorCode.ClientHttpFailedToOpenStream, `Failed to open SSE stream: ${response.statusText}`, {
319+
const statusText = response.statusText || `HTTP ${response.status}`;
320+
throw new SdkHttpError(SdkErrorCode.ClientHttpFailedToOpenStream, `Failed to open SSE stream: ${statusText}`, {
294321
status: response.status,
295322
statusText: response.statusText
296323
});
297324
}
298325

326+
this._standaloneSseReconnectError = undefined;
299327
this._handleSseStream(response.body, options, true);
300328
} catch (error) {
301329
this.onerror?.(error as Error);
@@ -330,13 +358,17 @@ export class StreamableHTTPClientTransport implements Transport {
330358
* @param lastEventId The ID of the last received event for resumability
331359
* @param attemptCount Current reconnection attempt count for this specific stream
332360
*/
333-
private _scheduleReconnection(options: StartSSEOptions, attemptCount = 0): void {
361+
private _scheduleReconnection(options: StartSSEOptions, attemptCount = 0, failFutureRequests = false): void {
334362
// Use provided options or default options
335363
const maxRetries = this._reconnectionOptions.maxRetries;
336364

337365
// Check if we've exceeded maximum retry attempts
338366
if (attemptCount >= maxRetries) {
339-
this.onerror?.(new Error(`Maximum reconnection attempts (${maxRetries}) exceeded.`));
367+
const error = new Error(`Maximum reconnection attempts (${maxRetries}) exceeded.`);
368+
if (failFutureRequests) {
369+
this._standaloneSseReconnectError = error;
370+
}
371+
this.onerror?.(error);
340372
return;
341373
}
342374

@@ -347,9 +379,9 @@ export class StreamableHTTPClientTransport implements Transport {
347379
this._cancelReconnection = undefined;
348380
if (this._abortController?.signal.aborted) return;
349381
this._startOrAuthSse(options).catch(error => {
350-
this.onerror?.(new Error(`Failed to reconnect SSE stream: ${error instanceof Error ? error.message : String(error)}`));
382+
this.onerror?.(new Error(`Failed to reconnect SSE stream: ${errorMessage(error)}`));
351383
try {
352-
this._scheduleReconnection(options, attemptCount + 1);
384+
this._scheduleReconnection(options, attemptCount + 1, failFutureRequests);
353385
} catch (scheduleError) {
354386
this.onerror?.(scheduleError instanceof Error ? scheduleError : new Error(String(scheduleError)));
355387
}
@@ -445,12 +477,13 @@ export class StreamableHTTPClientTransport implements Transport {
445477
onresumptiontoken,
446478
replayMessageId
447479
},
448-
0
480+
0,
481+
isReconnectable
449482
);
450483
}
451484
} catch (error) {
452485
// Handle stream errors - likely a network disconnect
453-
this.onerror?.(new Error(`SSE stream disconnected: ${error}`));
486+
this.onerror?.(new Error(`SSE stream disconnected: ${errorMessage(error)}`));
454487

455488
// Attempt to reconnect if the stream disconnects unexpectedly and we aren't closing
456489
// Reconnect if: already reconnectable (GET stream) OR received a priming event (POST stream with event ID)
@@ -466,10 +499,11 @@ export class StreamableHTTPClientTransport implements Transport {
466499
onresumptiontoken,
467500
replayMessageId
468501
},
469-
0
502+
0,
503+
isReconnectable
470504
);
471505
} catch (error) {
472-
this.onerror?.(new Error(`Failed to reconnect: ${error instanceof Error ? error.message : String(error)}`));
506+
this.onerror?.(new Error(`Failed to reconnect: ${errorMessage(error)}`));
473507
}
474508
}
475509
}
@@ -540,6 +574,12 @@ export class StreamableHTTPClientTransport implements Transport {
540574
return;
541575
}
542576

577+
const messages = Array.isArray(message) ? message : [message];
578+
const hasRequests = messages.some(msg => 'method' in msg && 'id' in msg && msg.id !== undefined);
579+
if (hasRequests && this._standaloneSseReconnectError) {
580+
throw new Error(`SSE stream reconnection failed: ${this._standaloneSseReconnectError.message}`);
581+
}
582+
543583
const headers = await this._commonHeaders();
544584
headers.set('content-type', 'application/json');
545585
const userAccept = headers.get('accept');
@@ -654,11 +694,6 @@ export class StreamableHTTPClientTransport implements Transport {
654694
return;
655695
}
656696

657-
// Get original message(s) for detecting request IDs
658-
const messages = Array.isArray(message) ? message : [message];
659-
660-
const hasRequests = messages.some(msg => 'method' in msg && 'id' in msg && msg.id !== undefined);
661-
662697
// Check the response type
663698
const contentType = response.headers.get('content-type');
664699

packages/client/test/client/streamableHttp.test.ts

Lines changed: 26 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -928,7 +928,7 @@ describe('StreamableHTTPClientTransport', () => {
928928
// ASSERT
929929
expect(errorSpy).toHaveBeenCalledWith(
930930
expect.objectContaining({
931-
message: expect.stringContaining('SSE stream disconnected: Error: Network failure')
931+
message: expect.stringContaining('SSE stream disconnected: Network failure')
932932
})
933933
);
934934
// THE KEY ASSERTION: A second fetch call proves reconnection was attempted.
@@ -1811,6 +1811,31 @@ describe('StreamableHTTPClientTransport', () => {
18111811
// Clean up the pending reconnection to avoid test pollution
18121812
transport['_cancelReconnection']?.();
18131813
});
1814+
1815+
it('should fail future requests after standalone SSE reconnect attempts are exhausted', async () => {
1816+
transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
1817+
reconnectionOptions: {
1818+
initialReconnectionDelay: 10,
1819+
maxRetries: 0,
1820+
maxReconnectionDelay: 1000,
1821+
reconnectionDelayGrowFactor: 1
1822+
}
1823+
});
1824+
1825+
transport['_scheduleReconnection']({}, 0, true);
1826+
1827+
const message: JSONRPCRequest = {
1828+
jsonrpc: '2.0',
1829+
method: 'tools/call',
1830+
params: {},
1831+
id: 'request-after-dead-sse'
1832+
};
1833+
1834+
await expect(transport.send(message)).rejects.toThrow(
1835+
'SSE stream reconnection failed: Maximum reconnection attempts (0) exceeded.'
1836+
);
1837+
expect(globalThis.fetch).not.toHaveBeenCalled();
1838+
});
18141839
});
18151840

18161841
describe('prevent infinite recursion when server returns 401 after successful auth', () => {

0 commit comments

Comments
 (0)