From 6cbfd827722a793a1429c613c9846597dcc1cc0a Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Fri, 14 Aug 2026 00:01:13 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=94=8C=20fix:=20Recover=20Quietly=20From?= =?UTF-8?q?=20Stale=20MCP=20SSE=20Stream=20Conflicts=20(#14816)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Streamable HTTP server allows one standalone `GET` SSE stream per session and releases its mapping from the response stream's cancel callback. That callback never runs when the connection dies at a proxy rather than at the client, so the server keeps holding a stream nobody is reading while the client knows its stream is gone. Every reconnect carrying that session id then gets a 409: SSE stream disconnected: TypeError: terminated Transport error (may require manual intervention): Streamable HTTP error: Failed to open SSE stream: Conflict Transport error (may require manual intervention): Maximum reconnection attempts (2) exceeded. Nothing there requires manual intervention. The connection recovers on its own in a few seconds, because the rebuild the first 409 escalates to sends the spec-mandated `DELETE`, which drops the server's session along with the stream it leaked. Two things made a self-healing event read as a fatal one. `extractSSEErrorMessage` classified status by scanning the message text for digits, but `StreamableHTTPError` and `SseError` carry the status on `code` and their messages do not always repeat it. "Failed to open SSE stream: Conflict" has no digits at all, so a 409 never reached the status branch and fell through to the terminal `isTransient: false` — the same verdict as a DNS typo. A 5xx arriving on `code` alone had the same blind spot. The status is now read from `code` when it is in HTTP range, with the message scan kept as a fallback, and 409 joins 5xx as transient: the stale session it reports is cleared by the rebuild, with nothing for an operator to do. The second is volume. Each SDK retry fires `onerror` twice — once with the raw throw out of `_startOrAuthSse`, once with the `Failed to reconnect SSE stream` wrapper. Only the wrapper matched the existing suppression, so every doomed retry logged at error level, and the retries are doomed by construction: nothing about the same session id can stop conflicting. The first conflict now escalates for rebuild and the rest are logged as the echo they are, along with the SDK's out-of-retries announcement when a rebuild is already underway. The non-conflict path for that announcement is untouched, so an exhausted budget still falls through to our reconnection everywhere else. `extractSSEErrorMessage` moves to `errors.ts` alongside `isOAuthAuthenticationError`. It had no test: `MCPConnection.test.ts` held a hand-copied clone marked "keep in sync with the actual implementation", so 66 assertions were exercising the copy. The clone is deleted and the suite now imports the real function, which it turns out had not drifted. `MCPConnectionSseConflict.test.ts` drives a real client transport against a real in-process `StreamableHTTPServerTransport` reproducing the sequence above: the stream opens, its socket is destroyed underneath the client, and every later `GET` on that session id conflicts while a rebuilt session gets a healthy stream. --- .../src/mcp/__tests__/MCPConnection.test.ts | 193 +++++-------- .../MCPConnectionSseConflict.test.ts | 272 ++++++++++++++++++ packages/api/src/mcp/connection.ts | 184 +++--------- packages/api/src/mcp/errors.ts | 175 +++++++++++ 4 files changed, 568 insertions(+), 256 deletions(-) create mode 100644 packages/api/src/mcp/__tests__/MCPConnectionSseConflict.test.ts diff --git a/packages/api/src/mcp/__tests__/MCPConnection.test.ts b/packages/api/src/mcp/__tests__/MCPConnection.test.ts index 2ca6a966dc..cfbba3a00c 100644 --- a/packages/api/src/mcp/__tests__/MCPConnection.test.ts +++ b/packages/api/src/mcp/__tests__/MCPConnection.test.ts @@ -1,16 +1,21 @@ /** * Tests for MCPConnection error detection methods. * - * Rate-limit and SSE tests use standalone implementations that mirror private methods in - * MCPConnection. OAuth classification exercises the production helper shared by the connection - * and factory. + * Rate-limit tests use a standalone implementation that mirrors a private method in + * MCPConnection. OAuth and SSE classification exercise the production helpers shared by the + * connection and factory. * * Alternative approaches considered: * 1. Reflection/type casting - fragile and breaks with refactoring * 2. Protected methods with test subclass - changes public API for testing * 3. Integration tests - tested separately in the full MCP test suite */ -import { isOAuthAuthenticationError } from '~/mcp/errors'; +import { StreamableHTTPError } from '@modelcontextprotocol/sdk/client/streamableHttp.js'; +import { + extractSSEErrorMessage, + isOAuthAuthenticationError, + isStandaloneSseConflict, +} from '~/mcp/errors'; describe('MCPConnection Error Detection', () => { /** @@ -173,119 +178,6 @@ describe('MCPConnection Error Detection', () => { * particularly handling the "SSE error: undefined" case from the MCP SDK. */ describe('extractSSEErrorMessage', () => { - /** - * Standalone implementation of extractSSEErrorMessage for testing. - * This mirrors the function in connection.ts. - * Keep in sync with the actual implementation. - */ - function extractSSEErrorMessage(error: unknown): { - message: string; - code?: number; - isProxyHint: boolean; - isTransient: boolean; - } { - if (!error || typeof error !== 'object') { - return { - message: 'Unknown SSE transport error', - isProxyHint: true, - isTransient: true, - }; - } - - const errorObj = error as { message?: string; code?: number; event?: unknown }; - const rawMessage = errorObj.message ?? ''; - const code = errorObj.code; - - // Handle the common "SSE error: undefined" case - if (rawMessage === 'SSE error: undefined' || rawMessage === 'undefined' || !rawMessage) { - return { - message: - 'SSE connection closed. This can occur due to: (1) idle connection timeout (normal), ' + - '(2) reverse proxy buffering (check proxy_buffering config), or (3) network interruption.', - code, - isProxyHint: true, - isTransient: true, - }; - } - - // Check for timeout patterns with case-insensitive matching - const lowerMessage = rawMessage.toLowerCase(); - if ( - rawMessage.includes('ETIMEDOUT') || - rawMessage.includes('ESOCKETTIMEDOUT') || - lowerMessage.includes('timed out') || - lowerMessage.includes('timeout after') || - lowerMessage.includes('request timeout') - ) { - return { - message: `SSE connection timed out: ${rawMessage}. If behind a reverse proxy, increase proxy_read_timeout.`, - code, - isProxyHint: true, - isTransient: true, - }; - } - - // Connection reset is often transient - if (rawMessage.includes('ECONNRESET')) { - return { - message: `SSE connection reset: ${rawMessage}. The server or proxy may have restarted.`, - code, - isProxyHint: false, - isTransient: true, - }; - } - - // Connection refused is more serious - if (rawMessage.includes('ECONNREFUSED')) { - return { - message: `SSE connection refused: ${rawMessage}. Verify the MCP server is running and accessible.`, - code, - isProxyHint: false, - isTransient: false, - }; - } - - // DNS failure - if (rawMessage.includes('ENOTFOUND') || rawMessage.includes('getaddrinfo')) { - return { - message: `SSE DNS resolution failed: ${rawMessage}. Check the server URL is correct.`, - code, - isProxyHint: false, - isTransient: false, - }; - } - - // Check for HTTP status codes - const statusMatch = rawMessage.match(/\b(4\d{2}|5\d{2})\b/); - if (statusMatch) { - const statusCode = parseInt(statusMatch[1], 10); - const isServerError = statusCode >= 500 && statusCode < 600; - return { - message: rawMessage, - code: statusCode, - isProxyHint: statusCode === 502 || statusCode === 503 || statusCode === 504, - isTransient: isServerError, - }; - } - - if (rawMessage === 'fetch failed') { - return { - message: - 'fetch failed (request aborted, likely after a timeout — connection may still be usable)', - code, - isProxyHint: false, - isTransient: true, - }; - } - - return { - message: rawMessage, - code, - isProxyHint: false, - isTransient: false, - }; - } - describe('undefined/empty error handling', () => { it('should handle "SSE error: undefined" from MCP SDK', () => { const error = { message: 'SSE error: undefined', code: undefined }; @@ -536,6 +428,73 @@ describe('extractSSEErrorMessage', () => { expect(result.isTransient).toBe(false); }); }); + + describe('status carried on code rather than in the message', () => { + it('should classify a 409 standalone SSE stream conflict as transient', () => { + const error = new StreamableHTTPError(409, 'Failed to open SSE stream: Conflict'); + const result = extractSSEErrorMessage(error); + + expect(result.code).toBe(409); + expect(result.isTransient).toBe(true); + expect(result.isProxyHint).toBe(false); + }); + + it('should classify a 5xx as transient when the message carries no digits', () => { + const error = new StreamableHTTPError(503, 'Failed to open SSE stream: Service Unavailable'); + const result = extractSSEErrorMessage(error); + + expect(result.code).toBe(503); + expect(result.isTransient).toBe(true); + expect(result.isProxyHint).toBe(true); + }); + + it('should keep other 4xx non-transient when the message carries no digits', () => { + const error = new StreamableHTTPError(403, 'Failed to open SSE stream: Forbidden'); + const result = extractSSEErrorMessage(error); + + expect(result.code).toBe(403); + expect(result.isTransient).toBe(false); + }); + + it('should ignore non-HTTP codes so they do not reach the status branch', () => { + const error = { message: 'Something went wrong', code: 42 }; + const result = extractSSEErrorMessage(error); + + expect(result.code).toBe(42); + expect(result.isTransient).toBe(false); + }); + }); +}); + +describe('isStandaloneSseConflict', () => { + it('should detect the SDK 409 raised when opening the standalone SSE stream', () => { + const error = new StreamableHTTPError(409, 'Failed to open SSE stream: Conflict'); + + expect(isStandaloneSseConflict(error)).toBe(true); + }); + + it('should not match a 409 raised for anything other than the SSE stream', () => { + const error = new StreamableHTTPError(409, 'Failed to send message'); + + expect(isStandaloneSseConflict(error)).toBe(false); + }); + + it('should not match other statuses on the SSE stream', () => { + const error = new StreamableHTTPError(500, 'Failed to open SSE stream: Internal Server Error'); + + expect(isStandaloneSseConflict(error)).toBe(false); + }); + + it('should not match a look-alike error from another transport', () => { + const error = Object.assign(new Error('Failed to open SSE stream: Conflict'), { code: 409 }); + + expect(isStandaloneSseConflict(error)).toBe(false); + }); + + it('should not match non-error values', () => { + expect(isStandaloneSseConflict(undefined)).toBe(false); + expect(isStandaloneSseConflict('Conflict')).toBe(false); + }); }); /** diff --git a/packages/api/src/mcp/__tests__/MCPConnectionSseConflict.test.ts b/packages/api/src/mcp/__tests__/MCPConnectionSseConflict.test.ts new file mode 100644 index 0000000000..f016fcfee2 --- /dev/null +++ b/packages/api/src/mcp/__tests__/MCPConnectionSseConflict.test.ts @@ -0,0 +1,272 @@ +/** + * Integration tests for the `409` a Streamable HTTP server returns when the standalone `GET` + * SSE stream is already mapped to the session. + * + * These drive a real client transport against a real in-process server that reproduces the + * production sequence: the stream opens, dies at the socket (undici surfaces this as + * `TypeError: terminated`), and every reconnect then conflicts because the server never + * observed the disconnect and still holds the previous stream. + */ + +import * as net from 'net'; +import * as http from 'http'; +import { randomUUID } from 'crypto'; +import { logger } from '@librechat/data-schemas'; +import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'; +import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js'; +import type { Socket } from 'net'; +import { MCPConnection } from '~/mcp/connection'; + +jest.mock('@librechat/data-schemas', () => ({ + logger: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + }, +})); + +jest.mock('~/auth', () => ({ + createSSRFSafeUndiciConnect: jest.fn(() => undefined), + isOAuthUrlAllowed: jest.fn(() => false), + isSSRFTarget: jest.fn(() => false), + resolveHostnameSSRF: jest.fn(async () => false), +})); + +jest.mock('~/mcp/mcpConfig', () => ({ + mcpConfig: { + CONNECTION_CHECK_TTL: 0, + TOOLS_LIST_MAX_PAGES: 50, + TOOLS_LIST_MAX_TOOLS: 1000, + TOOLS_LIST_MAX_BYTES: 5 * 1024 * 1024, + TOOLS_LIST_TIMEOUT_MS: 30000, + }, +})); + +interface ConflictTestServer { + url: string; + getRequests: () => number; + close: () => Promise; +} + +function listenOnEphemeralPort(httpServer: http.Server): Promise { + return new Promise((resolve) => { + httpServer.listen(0, '127.0.0.1', () => { + const addr = httpServer.address() as net.AddressInfo; + resolve(addr.port); + }); + }); +} + +function trackSockets(httpServer: http.Server): () => Promise { + const sockets = new Set(); + httpServer.on('connection', (socket: Socket) => { + sockets.add(socket); + socket.once('close', () => sockets.delete(socket)); + }); + return () => + new Promise((resolve) => { + for (const socket of sockets) { + socket.destroy(); + } + sockets.clear(); + httpServer.close(() => resolve()); + }); +} + +/** + * Serves `initialize` from a real `StreamableHTTPServerTransport` so the client negotiates a + * genuine session, then models a server holding a leaked stream. The leak is per-session, as it + * is in the SDK: the first session's stream has its socket destroyed underneath the client and + * every later `GET` carrying that session id conflicts, while a rebuilt session gets a healthy + * stream — so recovery is only possible by replacing the session. + */ +async function createConflictingServer(): Promise { + const sessions = new Map(); + const streamedSessions = new Set(); + let getRequests = 0; + + const httpServer = http.createServer(async (req, res) => { + if (req.method === 'GET') { + getRequests += 1; + const streamSid = (req.headers['mcp-session-id'] as string | undefined) ?? ''; + + if (streamedSessions.has(streamSid)) { + res.writeHead(409, { 'Content-Type': 'application/json' }); + res.end( + JSON.stringify({ + jsonrpc: '2.0', + error: { + code: -32000, + message: 'Conflict: Only one SSE stream is allowed per session', + }, + id: null, + }), + ); + return; + } + + const isFirstStream = streamedSessions.size === 0; + streamedSessions.add(streamSid); + res.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache, no-transform', + }); + res.write(': open\n\n'); + + if (isFirstStream) { + setTimeout(() => req.socket.destroy(), 50); + } + return; + } + + const sid = req.headers['mcp-session-id'] as string | undefined; + let transport = sid ? sessions.get(sid) : undefined; + const isNewTransport = !transport; + + if (!transport) { + transport = new StreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID() }); + const mcp = new McpServer({ name: 'test-conflict', version: '0.0.1' }); + await mcp.connect(transport); + } + + await transport.handleRequest(req, res); + + if (isNewTransport && transport.sessionId) { + const registeredId = transport.sessionId; + sessions.set(registeredId, transport); + transport.onclose = () => sessions.delete(registeredId); + } + }); + + const destroySockets = trackSockets(httpServer); + const port = await listenOnEphemeralPort(httpServer); + + return { + url: `http://127.0.0.1:${port}/`, + getRequests: () => getRequests, + close: async () => { + const closing = [...sessions.values()].map((t) => t.close().catch(() => undefined)); + sessions.clear(); + await Promise.all(closing); + await destroySockets(); + }, + }; +} + +async function waitForCondition( + predicate: () => boolean, + timeoutMs = 10000, + intervalMs = 25, +): Promise { + const deadline = Date.now() + timeoutMs; + while (!predicate()) { + if (Date.now() > deadline) { + throw new Error('Timed out waiting for condition'); + } + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + } +} + +function loggedMessages(mock: jest.Mock): string[] { + return mock.mock.calls.map((call) => String(call[0])); +} + +function countMatching(mock: jest.Mock, needle: string): number { + return loggedMessages(mock).filter((message) => message.includes(needle)).length; +} + +async function safeDisconnect(conn: MCPConnection | null): Promise { + if (!conn) { + return; + } + (conn as unknown as { shouldStopReconnecting: boolean }).shouldStopReconnecting = true; + conn.removeAllListeners(); + await conn.disconnect(); +} + +describe('MCPConnection standalone SSE stream conflict', () => { + let server: ConflictTestServer | null; + let conn: MCPConnection | null; + + beforeEach(() => { + server = null; + conn = null; + jest.clearAllMocks(); + }); + + afterEach(async () => { + MCPConnection.clearCooldown('test'); + await safeDisconnect(conn); + conn = null; + await server?.close(); + }); + + it('escalates the first conflict for rebuild and reports the rest as follow-on noise', async () => { + const srv = await createConflictingServer(); + server = srv; + conn = new MCPConnection({ + serverName: 'test', + serverConfig: { type: 'streamable-http', url: srv.url }, + useSSRFProtection: false, + }); + + const escalations: string[] = []; + conn.on('connectionChange', (state: string) => { + if (state === 'error') { + escalations.push(state); + } + }); + + await conn.connect(); + + /** + * Let the SDK exhaust its own retry budget without the rebuild closing the transport + * partway through, so the full sequence of repeat conflicts is observable. + */ + (conn as unknown as { shouldStopReconnecting: boolean }).shouldStopReconnecting = true; + + const warnMock = logger.warn as jest.Mock; + const errorMock = logger.error as jest.Mock; + const debugMock = logger.debug as jest.Mock; + + await waitForCondition( + () => countMatching(warnMock, 'conflicted with a stale server-side') > 0, + ); + await waitForCondition(() => srv.getRequests() >= 3); + await waitForCondition( + () => countMatching(debugMock, 'SDK reconnection budget exhausted') > 0, + 10000, + ); + + expect(countMatching(warnMock, 'conflicted with a stale server-side')).toBe(1); + expect(escalations).toEqual(['error']); + + expect(countMatching(debugMock, 'still conflicting; session rebuild already underway')).toBe(1); + expect(countMatching(errorMock, 'may require manual intervention')).toBe(0); + expect(countMatching(errorMock, 'Maximum reconnection attempts')).toBe(0); + }, 20000); + + it('rebuilds the session so the connection recovers from the conflict', async () => { + const srv = await createConflictingServer(); + server = srv; + conn = new MCPConnection({ + serverName: 'test', + serverConfig: { type: 'streamable-http', url: srv.url }, + useSSRFProtection: false, + }); + + await conn.connect(); + const firstSessionId = (conn as unknown as { transport?: { sessionId?: string } }).transport + ?.sessionId; + expect(firstSessionId).toBeTruthy(); + + await waitForCondition(() => { + const current = (conn as unknown as { transport?: { sessionId?: string } }).transport + ?.sessionId; + return Boolean(current) && current !== firstSessionId; + }, 15000); + + expect(await conn.isConnected()).toBe(true); + }, 25000); +}); diff --git a/packages/api/src/mcp/connection.ts b/packages/api/src/mcp/connection.ts index e186e481f6..54805d95b7 100644 --- a/packages/api/src/mcp/connection.ts +++ b/packages/api/src/mcp/connection.ts @@ -23,10 +23,14 @@ import type { import type { Transport } from '@modelcontextprotocol/sdk/shared/transport.js'; import type { MCPOAuthTokens } from './oauth/types'; import type * as t from './types'; +import { + extractSSEErrorMessage, + isOAuthAuthenticationError, + isStandaloneSseConflict, +} from './errors'; import { createSSRFSafeUndiciConnect, isSSRFTarget, resolveHostnameSSRF } from '~/auth'; import { reserveMCPToolsChangedRevision } from './toolsChanged'; import { isOAuthServer, sanitizeUrlForLogging } from './utils'; -import { isOAuthAuthenticationError } from './errors'; import { runOutsideTracing } from '~/utils/tracing'; import { isAddressAllowed } from '~/auth/domain'; import { withTimeout } from '~/utils/promise'; @@ -965,6 +969,7 @@ const DEFAULT_SSE_READ_TIMEOUT = FIVE_MINUTES; */ const SDK_SSE_STREAM_DISCONNECTED = 'SSE stream disconnected'; const SDK_SSE_RECONNECT_FAILED = 'Failed to reconnect SSE stream'; +const SDK_SSE_RETRIES_EXHAUSTED = 'Maximum reconnection attempts'; /** * Headers for SSE connections. @@ -980,144 +985,6 @@ const SSE_REQUEST_HEADERS = { 'Cache-Control': 'no-cache', }; -/** - * Extracts a meaningful error message from SSE transport errors. - * The MCP SDK's SSEClientTransport can produce "SSE error: undefined" when the - * underlying eventsource library encounters connection issues without a specific message. - * - * @returns Object containing: - * - message: Human-readable error description - * - code: HTTP status code if available - * - isProxyHint: Whether this error suggests proxy misconfiguration - * - isTransient: Whether this is likely a transient error that will auto-reconnect - */ -function extractSSEErrorMessage(error: unknown): { - message: string; - code?: number; - isProxyHint: boolean; - isTransient: boolean; -} { - if (!error || typeof error !== 'object') { - return { - message: 'Unknown SSE transport error', - isProxyHint: true, - isTransient: true, - }; - } - - const errorObj = error as { message?: string; code?: number; event?: unknown }; - const rawMessage = errorObj.message ?? ''; - const code = errorObj.code; - - /** - * Handle the common "SSE error: undefined" case. - * This typically occurs when: - * 1. A reverse proxy buffers the SSE stream (proxy issue) - * 2. The server closes an idle connection (normal SSE behavior) - * 3. Network interruption without specific error details - * - * In all cases, the eventsource library will attempt to reconnect automatically. - */ - if (rawMessage === 'SSE error: undefined' || rawMessage === 'undefined' || !rawMessage) { - return { - message: - 'SSE connection closed. This can occur due to: (1) idle connection timeout (normal), ' + - '(2) reverse proxy buffering (check proxy_buffering config), or (3) network interruption.', - code, - isProxyHint: true, - isTransient: true, - }; - } - - /** - * Check for timeout patterns. Use case-insensitive matching for common timeout error codes: - * - ETIMEDOUT: TCP connection timeout - * - ESOCKETTIMEDOUT: Socket timeout - * - "timed out" / "timeout": Generic timeout messages - */ - const lowerMessage = rawMessage.toLowerCase(); - if ( - rawMessage.includes('ETIMEDOUT') || - rawMessage.includes('ESOCKETTIMEDOUT') || - lowerMessage.includes('timed out') || - lowerMessage.includes('timeout after') || - lowerMessage.includes('request timeout') - ) { - return { - message: `SSE connection timed out: ${rawMessage}. If behind a reverse proxy, increase proxy_read_timeout.`, - code, - isProxyHint: true, - isTransient: true, - }; - } - - // Connection reset is often transient (server restart, proxy reload) - if (rawMessage.includes('ECONNRESET')) { - return { - message: `SSE connection reset: ${rawMessage}. The server or proxy may have restarted.`, - code, - isProxyHint: false, - isTransient: true, - }; - } - - // Connection refused is more serious - server may be down - if (rawMessage.includes('ECONNREFUSED')) { - return { - message: `SSE connection refused: ${rawMessage}. Verify the MCP server is running and accessible.`, - code, - isProxyHint: false, - isTransient: false, - }; - } - - // DNS failure is usually a configuration issue, not transient - if (rawMessage.includes('ENOTFOUND') || rawMessage.includes('getaddrinfo')) { - return { - message: `SSE DNS resolution failed: ${rawMessage}. Check the server URL is correct.`, - code, - isProxyHint: false, - isTransient: false, - }; - } - - // Check for HTTP status codes in the message - const statusMatch = rawMessage.match(/\b(4\d{2}|5\d{2})\b/); - if (statusMatch) { - const statusCode = parseInt(statusMatch[1], 10); - // 5xx errors are often transient, 4xx are usually not - const isServerError = statusCode >= 500 && statusCode < 600; - return { - message: rawMessage, - code: statusCode, - isProxyHint: statusCode === 502 || statusCode === 503 || statusCode === 504, - isTransient: isServerError, - }; - } - - /** - * "fetch failed" is a generic undici TypeError that occurs when an in-flight HTTP request - * is aborted (e.g. after an MCP protocol-level timeout fires). The transport itself is still - * functional — only the individual request was lost — so treat this as transient. - */ - if (rawMessage === 'fetch failed') { - return { - message: - 'fetch failed (request aborted, likely after a timeout — connection may still be usable)', - code, - isProxyHint: false, - isTransient: true, - }; - } - - return { - message: rawMessage, - code, - isProxyHint: false, - isTransient: false, - }; -} - interface MCPConnectionParams { serverName: string; serverConfig: t.MCPOptions; @@ -1148,6 +1015,8 @@ export class MCPConnection extends EventEmitter { private isReconnecting = false; private isInitializing = false; private reconnectAttempts = 0; + /** Set once per transport, so only the first of a conflict's repeat reports escalates. */ + private reportedStandaloneSseConflict = false; private agents: Dispatcher[] = []; private readonly userId?: string; private lastPingTime: number; @@ -2264,6 +2133,8 @@ export class MCPConnection extends EventEmitter { } private setupTransportErrorHandlers(transport: Transport): void { + this.reportedStandaloneSseConflict = false; + transport.onerror = (error) => { const rawMessage = error && typeof error === 'object' ? ((error as { message?: string }).message ?? '') : ''; @@ -2285,6 +2156,41 @@ export class MCPConnection extends EventEmitter { return; } + /** + * A stale server-side stream can only be cleared by rebuilding the session, so escalate + * on the first report and treat the rest as the echo they are: the SDK keeps retrying the + * `GET` on its own schedule, and each of those retries conflicts for the same reason while + * the rebuild triggered here is already running. + */ + if (isStandaloneSseConflict(error)) { + if (this.reportedStandaloneSseConflict) { + logger.debug( + `${this.getLogPrefix()} SSE stream still conflicting; session rebuild already underway`, + ); + return; + } + + this.reportedStandaloneSseConflict = true; + logger.warn( + `${this.getLogPrefix()} SSE stream conflicted with a stale server-side stream for this ` + + `session (409). Rebuilding the session; no action needed.`, + ); + this.emit('connectionChange', 'error'); + return; + } + + /** + * The SDK announcing it is out of retries normally has to fall through so our reconnection + * takes over, but adds nothing once a conflict rebuild is already running — that rebuild is + * what exhausted the retries, and it is the recovery. + */ + if (this.reportedStandaloneSseConflict && rawMessage.startsWith(SDK_SSE_RETRIES_EXHAUSTED)) { + logger.debug( + `${this.getLogPrefix()} SDK reconnection budget exhausted; session rebuild already underway`, + ); + return; + } + const { message: errorMessage, code: errorCode, diff --git a/packages/api/src/mcp/errors.ts b/packages/api/src/mcp/errors.ts index f503e8dbb5..3c356b3bad 100644 --- a/packages/api/src/mcp/errors.ts +++ b/packages/api/src/mcp/errors.ts @@ -1,6 +1,7 @@ /** * MCP-specific error classes */ +import { StreamableHTTPError } from '@modelcontextprotocol/sdk/client/streamableHttp.js'; export const MCPErrorCodes = { DOMAIN_NOT_ALLOWED: 'MCP_DOMAIN_NOT_ALLOWED', @@ -51,6 +52,180 @@ export function isOAuthAuthenticationError(error: unknown): boolean { ); } +/** The SDK's wording when the standalone `GET` SSE stream could not be opened. */ +const SDK_SSE_OPEN_FAILED = 'Failed to open SSE stream'; + +/** + * A `409` opening the standalone `GET` SSE stream means the server still has the previous + * stream mapped to our session. The spec allows only one such stream per session, and the + * server drops its mapping from the response stream's cancel callback — which never runs when + * the connection dies at a proxy instead of at the client, leaving the server holding a stream + * no one is reading. + * + * Retrying cannot clear it: every `GET` carrying that session id conflicts for the same reason. + * Only a transport rebuild does, because the teardown `DELETE` drops the server's session along + * with the stream it leaked. + */ +export function isStandaloneSseConflict(error: unknown): boolean { + return ( + error instanceof StreamableHTTPError && + error.code === 409 && + error.message.includes(SDK_SSE_OPEN_FAILED) + ); +} + +export interface SSEErrorDetails { + message: string; + code?: number; + isProxyHint: boolean; + isTransient: boolean; +} + +/** + * Extracts a meaningful error message from SSE transport errors. + * The MCP SDK's SSEClientTransport can produce "SSE error: undefined" when the + * underlying eventsource library encounters connection issues without a specific message. + * + * @returns Object containing: + * - message: Human-readable error description + * - code: HTTP status code if available + * - isProxyHint: Whether this error suggests proxy misconfiguration + * - isTransient: Whether this is likely a transient error that will auto-reconnect + */ +export function extractSSEErrorMessage(error: unknown): SSEErrorDetails { + if (!error || typeof error !== 'object') { + return { + message: 'Unknown SSE transport error', + isProxyHint: true, + isTransient: true, + }; + } + + const errorObj = error as { message?: string; code?: number; event?: unknown }; + const rawMessage = errorObj.message ?? ''; + const code = errorObj.code; + + /** + * Handle the common "SSE error: undefined" case. + * This typically occurs when: + * 1. A reverse proxy buffers the SSE stream (proxy issue) + * 2. The server closes an idle connection (normal SSE behavior) + * 3. Network interruption without specific error details + * + * In all cases, the eventsource library will attempt to reconnect automatically. + */ + if (rawMessage === 'SSE error: undefined' || rawMessage === 'undefined' || !rawMessage) { + return { + message: + 'SSE connection closed. This can occur due to: (1) idle connection timeout (normal), ' + + '(2) reverse proxy buffering (check proxy_buffering config), or (3) network interruption.', + code, + isProxyHint: true, + isTransient: true, + }; + } + + /** + * Check for timeout patterns. Use case-insensitive matching for common timeout error codes: + * - ETIMEDOUT: TCP connection timeout + * - ESOCKETTIMEDOUT: Socket timeout + * - "timed out" / "timeout": Generic timeout messages + */ + const lowerMessage = rawMessage.toLowerCase(); + if ( + rawMessage.includes('ETIMEDOUT') || + rawMessage.includes('ESOCKETTIMEDOUT') || + lowerMessage.includes('timed out') || + lowerMessage.includes('timeout after') || + lowerMessage.includes('request timeout') + ) { + return { + message: `SSE connection timed out: ${rawMessage}. If behind a reverse proxy, increase proxy_read_timeout.`, + code, + isProxyHint: true, + isTransient: true, + }; + } + + // Connection reset is often transient (server restart, proxy reload) + if (rawMessage.includes('ECONNRESET')) { + return { + message: `SSE connection reset: ${rawMessage}. The server or proxy may have restarted.`, + code, + isProxyHint: false, + isTransient: true, + }; + } + + // Connection refused is more serious - server may be down + if (rawMessage.includes('ECONNREFUSED')) { + return { + message: `SSE connection refused: ${rawMessage}. Verify the MCP server is running and accessible.`, + code, + isProxyHint: false, + isTransient: false, + }; + } + + // DNS failure is usually a configuration issue, not transient + if (rawMessage.includes('ENOTFOUND') || rawMessage.includes('getaddrinfo')) { + return { + message: `SSE DNS resolution failed: ${rawMessage}. Check the server URL is correct.`, + code, + isProxyHint: false, + isTransient: false, + }; + } + + /** + * `StreamableHTTPError` and `SseError` carry the status on `code`, and their messages do not + * always repeat it — "Failed to open SSE stream: Conflict" arrives here with no digits to + * scan, so reading the message alone classified a 409 as an unrecognized fatal error. Prefer + * the structured status; keep scanning the message for transports that only stringify it. + */ + const statusMatch = rawMessage.match(/\b(4\d{2}|5\d{2})\b/); + const scannedStatus = statusMatch ? parseInt(statusMatch[1], 10) : undefined; + const carriesHttpStatus = code != null && code >= 400 && code < 600; + const httpStatus = carriesHttpStatus ? code : scannedStatus; + + if (httpStatus != null) { + /** + * 5xx is transient by nature. 409 is transient by recovery: it reports a stale server-side + * session that the rebuild clears on its own, with nothing for an operator to do. Every + * other 4xx stays non-transient. + */ + const isServerError = httpStatus >= 500 && httpStatus < 600; + return { + message: rawMessage, + code: httpStatus, + isProxyHint: httpStatus === 502 || httpStatus === 503 || httpStatus === 504, + isTransient: isServerError || httpStatus === 409, + }; + } + + /** + * "fetch failed" is a generic undici TypeError that occurs when an in-flight HTTP request + * is aborted (e.g. after an MCP protocol-level timeout fires). The transport itself is still + * functional — only the individual request was lost — so treat this as transient. + */ + if (rawMessage === 'fetch failed') { + return { + message: + 'fetch failed (request aborted, likely after a timeout — connection may still be usable)', + code, + isProxyHint: false, + isTransient: true, + }; + } + + return { + message: rawMessage, + code, + isProxyHint: false, + isTransient: false, + }; +} + /** * Custom error for MCP domain restriction violations. * Thrown when a user attempts to connect to an MCP server whose domain is not in the allowlist.