Merge branch 'fix/sse-read-timeout'

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;
+}

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<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
+    return await exchange(baseUrl, method, params, controller.signal);
-    const sseResp = await fetch(`${baseUrl}/sse`, {
+  } catch (err) {
-      method: 'GET',
+    if (controller.signal.aborted) {
-      headers: { 'Accept': 'text/event-stream' },
+      return errorResponse(`SSE request timed out after ${String(timeoutMs)}ms`);
-      signal: controller.signal,
-    });
-
-    if (!sseResp.ok) {
-      return errorResponse(`SSE connect failed: HTTP ${sseResp.status}`);
     }
-
+    throw err;
-    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');
   } 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',

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<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);
+});