From fbd289d2ba4593fd385983dc47d7c376a60f2602 Mon Sep 17 00:00:00 2001 From: Michal Date: Sat, 7 Nov 2026 13:49:46 +0000 Subject: [PATCH] fix(mcpd): SSE transport no longer cuts replies off after 5s sendViaSse capped the reply read at a hard-coded 5s, so any SSE tool call slower than that (techdocs fetch_url falling back after a Playwright timeout) failed with 'No response received from SSE stream'. The read is now bounded only by the exchange timeout, which defaults to 120s to match mcplocal's TOOLCALL_TIMEOUT_MS and honours the server's toolCallTimeoutSeconds. Hitting it returns an explicit timeout error. Co-Authored-By: Claude Opus 5.5 --- src/mcpd/src/services/mcp-proxy-service.ts | 10 +- src/mcpd/src/services/transport/sse-client.ts | 254 ++++++++++-------- src/mcpd/tests/sse-client.test.ts | 59 ++++ 3 files changed, 204 insertions(+), 119 deletions(-) create mode 100644 src/mcpd/tests/sse-client.test.ts diff --git a/src/mcpd/src/services/mcp-proxy-service.ts b/src/mcpd/src/services/mcp-proxy-service.ts index df85367..5b7ac8a 100644 --- a/src/mcpd/src/services/mcp-proxy-service.ts +++ b/src/mcpd/src/services/mcp-proxy-service.ts @@ -177,7 +177,7 @@ export class McpProxyService { const url = server.externalUrl as string; if (server.transport === 'SSE') { - return sendViaSse(url, method, params); + return sendViaSse(url, method, params, sseTimeoutMs(server)); } // STREAMABLE_HTTP (default for external) @@ -257,7 +257,7 @@ export class McpProxyService { const baseUrl = await this.resolveBaseUrl(instance, server); if (transport === 'SSE') { - return sendViaSse(baseUrl, method, params); + return sendViaSse(baseUrl, method, params, sseTimeoutMs(server)); } // STREAMABLE_HTTP (default) @@ -417,3 +417,9 @@ export class McpProxyService { }); } } + +/** Per-server tool-call budget (toolCallTimeoutSeconds), else the SSE default. */ +function sseTimeoutMs(server: McpServer): number | undefined { + const seconds = server.toolCallTimeoutSeconds as number | null; + return seconds !== null && seconds !== undefined ? seconds * 1000 : undefined; +} diff --git a/src/mcpd/src/services/transport/sse-client.ts b/src/mcpd/src/services/transport/sse-client.ts index f1377c6..ab703a8 100644 --- a/src/mcpd/src/services/transport/sse-client.ts +++ b/src/mcpd/src/services/transport/sse-client.ts @@ -8,139 +8,159 @@ import type { McpProxyResponse } from '../mcp-proxy-service.js'; * * Each call opens a fresh SSE connection, initializes, sends the request, * reads the response, and closes. Session caching may be added later. + * + * `timeoutMs` bounds the WHOLE exchange, including waiting for the reply. + * The default matches mcplocal's TOOLCALL_TIMEOUT_MS so a slow tool (a + * Playwright render that times out and falls back) answers instead of being + * cut off — a separate 5s cap on the reply read used to fail every call + * longer than that with "No response received from SSE stream". */ +export const SSE_DEFAULT_TIMEOUT_MS = 120_000; + export async function sendViaSse( baseUrl: string, method: string, params?: Record, - timeoutMs = 30_000, + timeoutMs = SSE_DEFAULT_TIMEOUT_MS, ): Promise { const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), timeoutMs); try { - // 1. GET /sse → SSE stream - const sseResp = await fetch(`${baseUrl}/sse`, { - method: 'GET', - headers: { 'Accept': 'text/event-stream' }, - signal: controller.signal, - }); - - if (!sseResp.ok) { - return errorResponse(`SSE connect failed: HTTP ${sseResp.status}`); + return await exchange(baseUrl, method, params, controller.signal); + } catch (err) { + if (controller.signal.aborted) { + return errorResponse(`SSE request timed out after ${String(timeoutMs)}ms`); } - - const reader = sseResp.body?.getReader(); - if (!reader) { - return errorResponse('No SSE stream body'); - } - - // 2. Read until we get the endpoint event with messages URL - const decoder = new TextDecoder(); - let buffer = ''; - let messagesUrl = ''; - - while (!messagesUrl) { - const { done, value } = await reader.read(); - if (done) break; - buffer += decoder.decode(value, { stream: true }); - - for (const line of buffer.split('\n')) { - if (line.startsWith('data: ') && buffer.includes('event: endpoint')) { - const endpoint = line.slice(6).trim(); - messagesUrl = endpoint.startsWith('http') ? endpoint : `${baseUrl}${endpoint}`; - } - } - const lines = buffer.split('\n'); - buffer = lines[lines.length - 1] ?? ''; - } - - if (!messagesUrl) { - reader.cancel(); - return errorResponse('No endpoint event from SSE stream'); - } - - const postHeaders = { 'Content-Type': 'application/json' }; - - // 3. Initialize - const initResp = await fetch(messagesUrl, { - method: 'POST', - headers: postHeaders, - body: JSON.stringify({ - jsonrpc: '2.0', - id: 1, - method: 'initialize', - params: { - protocolVersion: '2024-11-05', - capabilities: {}, - clientInfo: { name: 'mcpctl-proxy', version: '0.0.1' }, - }, - }), - signal: controller.signal, - }); - - if (!initResp.ok) { - reader.cancel(); - return errorResponse(`SSE initialize failed: HTTP ${initResp.status}`); - } - - // 4. Send notifications/initialized - await fetch(messagesUrl, { - method: 'POST', - headers: postHeaders, - body: JSON.stringify({ jsonrpc: '2.0', method: 'notifications/initialized' }), - signal: controller.signal, - }); - - // 5. Send the actual request - const requestId = 2; - await fetch(messagesUrl, { - method: 'POST', - headers: postHeaders, - body: JSON.stringify({ - jsonrpc: '2.0', - id: requestId, - method, - ...(params !== undefined ? { params } : {}), - }), - signal: controller.signal, - }); - - // 6. Read response from SSE stream (matched by request ID) - let responseBuffer = ''; - const readTimeout = setTimeout(() => reader.cancel(), 5000); - - while (true) { - const { done, value } = await reader.read(); - if (done) break; - responseBuffer += decoder.decode(value, { stream: true }); - - for (const line of responseBuffer.split('\n')) { - if (line.startsWith('data: ')) { - try { - const parsed = JSON.parse(line.slice(6)) as McpProxyResponse; - if (parsed.id === requestId) { - clearTimeout(readTimeout); - reader.cancel(); - return parsed; - } - } catch { - // Not valid JSON, skip - } - } - } - const respLines = responseBuffer.split('\n'); - responseBuffer = respLines[respLines.length - 1] ?? ''; - } - - clearTimeout(readTimeout); - reader.cancel(); - return errorResponse('No response received from SSE stream'); + throw err; } finally { clearTimeout(timer); } } +async function exchange( + baseUrl: string, + method: string, + params: Record | undefined, + signal: AbortSignal, +): Promise { + // 1. GET /sse → SSE stream + const sseResp = await fetch(`${baseUrl}/sse`, { + method: 'GET', + headers: { 'Accept': 'text/event-stream' }, + signal, + }); + + if (!sseResp.ok) { + return errorResponse(`SSE connect failed: HTTP ${sseResp.status}`); + } + + const reader = sseResp.body?.getReader(); + if (!reader) { + return errorResponse('No SSE stream body'); + } + + // 2. Read until we get the endpoint event with messages URL + const decoder = new TextDecoder(); + let buffer = ''; + let messagesUrl = ''; + + while (!messagesUrl) { + const { done, value } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + + for (const line of buffer.split('\n')) { + if (line.startsWith('data: ') && buffer.includes('event: endpoint')) { + const endpoint = line.slice(6).trim(); + messagesUrl = endpoint.startsWith('http') ? endpoint : `${baseUrl}${endpoint}`; + } + } + const lines = buffer.split('\n'); + buffer = lines[lines.length - 1] ?? ''; + } + + if (!messagesUrl) { + reader.cancel(); + return errorResponse('No endpoint event from SSE stream'); + } + + const postHeaders = { 'Content-Type': 'application/json' }; + + // 3. Initialize + const initResp = await fetch(messagesUrl, { + method: 'POST', + headers: postHeaders, + body: JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { + protocolVersion: '2024-11-05', + capabilities: {}, + clientInfo: { name: 'mcpctl-proxy', version: '0.0.1' }, + }, + }), + signal, + }); + + if (!initResp.ok) { + reader.cancel(); + return errorResponse(`SSE initialize failed: HTTP ${initResp.status}`); + } + + // 4. Send notifications/initialized + await fetch(messagesUrl, { + method: 'POST', + headers: postHeaders, + body: JSON.stringify({ jsonrpc: '2.0', method: 'notifications/initialized' }), + signal, + }); + + // 5. Send the actual request + const requestId = 2; + await fetch(messagesUrl, { + method: 'POST', + headers: postHeaders, + body: JSON.stringify({ + jsonrpc: '2.0', + id: requestId, + method, + ...(params !== undefined ? { params } : {}), + }), + signal, + }); + + // 6. Read response from SSE stream (matched by request ID). Bounded by + // the caller's signal only: aborting the GET fetch also ends this read. + let responseBuffer = ''; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + responseBuffer += decoder.decode(value, { stream: true }); + + for (const line of responseBuffer.split('\n')) { + if (line.startsWith('data: ')) { + try { + const parsed = JSON.parse(line.slice(6)) as McpProxyResponse; + if (parsed.id === requestId) { + reader.cancel(); + return parsed; + } + } catch { + // Not valid JSON, skip + } + } + } + const respLines = responseBuffer.split('\n'); + responseBuffer = respLines[respLines.length - 1] ?? ''; + } + + reader.cancel(); + return errorResponse('No response received from SSE stream'); +} + function errorResponse(message: string): McpProxyResponse { return { jsonrpc: '2.0', diff --git a/src/mcpd/tests/sse-client.test.ts b/src/mcpd/tests/sse-client.test.ts new file mode 100644 index 0000000..b1fb8a9 --- /dev/null +++ b/src/mcpd/tests/sse-client.test.ts @@ -0,0 +1,59 @@ +import { describe, it, expect, afterEach } from 'vitest'; +import http from 'node:http'; +import type { AddressInfo } from 'node:net'; +import { sendViaSse } from '../src/services/transport/sse-client.js'; + +let server: http.Server | undefined; + +afterEach(async () => { + if (server) { + server.closeAllConnections(); + await new Promise((resolve) => server!.close(() => resolve())); + server = undefined; + } +}); + +/** Legacy-SSE MCP server whose reply to the tools/call (id 2) arrives after `replyDelayMs`. */ +async function startSlowSseServer(replyDelayMs: number): Promise { + let stream: http.ServerResponse | undefined; + server = http.createServer((req, res) => { + if (req.method === 'GET' && req.url === '/sse') { + res.writeHead(200, { 'Content-Type': 'text/event-stream' }); + res.write('event: endpoint\ndata: /messages\n\n'); + stream = res; + return; + } + let body = ''; + req.on('data', (c: Buffer) => { body += c.toString(); }); + req.on('end', () => { + res.writeHead(202).end(); + const msg = JSON.parse(body) as { id?: number }; + if (msg.id === 1) { + stream?.write(`data: ${JSON.stringify({ jsonrpc: '2.0', id: 1, result: {} })}\n\n`); + } else if (msg.id === 2) { + setTimeout(() => { + stream?.write(`data: ${JSON.stringify({ jsonrpc: '2.0', id: 2, result: { ok: true } })}\n\n`); + }, replyDelayMs); + } + }); + }); + await new Promise((resolve) => server!.listen(0, '127.0.0.1', () => resolve())); + return `http://127.0.0.1:${String((server.address() as AddressInfo).port)}`; +} + +describe('sendViaSse', () => { + // Regression: a fixed 5s cap on the reply read failed every slower tool + // (e.g. techdocs fetch_url) with "No response received from SSE stream". + it('waits for a reply slower than 5s within the timeout', async () => { + const url = await startSlowSseServer(6_000); + const res = await sendViaSse(url, 'tools/call', { name: 'slow' }, 15_000); + expect(res.error).toBeUndefined(); + expect(res.result).toEqual({ ok: true }); + }, 20_000); + + it('returns a timeout error when the reply exceeds timeoutMs', async () => { + const url = await startSlowSseServer(5_000); + const res = await sendViaSse(url, 'tools/call', { name: 'slow' }, 300); + expect(res.error?.message).toBe('SSE request timed out after 300ms'); + }, 10_000); +});