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 <noreply@anthropic.com>
This commit is contained in:
Michal
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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<string, unknown>,
|
||||
timeoutMs = 30_000,
|
||||
timeoutMs = SSE_DEFAULT_TIMEOUT_MS,
|
||||
): Promise<McpProxyResponse> {
|
||||
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<string, unknown> | undefined,
|
||||
signal: AbortSignal,
|
||||
): Promise<McpProxyResponse> {
|
||||
// 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',
|
||||
|
||||
59
src/mcpd/tests/sse-client.test.ts
Normal file
59
src/mcpd/tests/sse-client.test.ts
Normal file
@@ -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<void>((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<string> {
|
||||
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<void>((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);
|
||||
});
|
||||
Reference in New Issue
Block a user