fix: Actorized MCP servers have 30 seconds timeout to connect (#272)

MichalKalita · web-flow · commit cdffc3e80bcd · 2025-09-22T09:56:04.000+02:00
* fix: Actorized mcp server have 5 seconds timeout to connect

* feat: get mcp server tools in parallel

* fix: default stramable and backup SSE transport

* fix: improve based on code review feedback

* fix: types after master merge
diff --git a/src/errors.ts b/src/errors.ts
@@ -0,0 +1,3 @@
+export class TimeoutError extends Error {
+    override readonly name = 'TimeoutError';
+}
diff --git a/src/mcp/client.ts b/src/mcp/client.ts
@@ -4,6 +4,8 @@ import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/
 
 import log from '@apify/log';
 
+import { TimeoutError } from '../errors.js';
+import { ACTORIZED_MCP_CONNECTION_TIMEOUT_MSEC } from './const.js';
 import { getMCPServerID } from './utils.js';
 
 /**
@@ -12,16 +14,55 @@ import { getMCPServerID } from './utils.js';
  */
 export async function connectMCPClient(
     url: string, token: string,
-): Promise<Client> {
+): Promise<Client | null> {
+    let client: Client;
     try {
-        return await createMCPStreamableClient(url, token);
-    } catch {
+        client = await createMCPStreamableClient(url, token);
+        return client;
+    } catch (error) {
+        // If streamable HTTP transport fails on not timeout error, continue with SSE transport
+        if (error instanceof TimeoutError) {
+            log.warning('Connection to MCP server using streamable HTTP transport timed out', { url });
+            return null;
+        }
+
         // If streamable HTTP transport fails, fall back to SSE transport
         log.debug('Streamable HTTP transport failed, falling back to SSE transport', {
             url,
         });
-        return await createMCPSSEClient(url, token);
     }
+
+    try {
+        client = await createMCPSSEClient(url, token);
+        return client;
+    } catch (error) {
+        if (error instanceof TimeoutError) {
+            log.warning('Connection to MCP server using SSE transport timed out', { url });
+            return null;
+        }
+
+        log.error('Failed to connect to MCP server using SSE transport', { cause: error });
+        throw error;
+    }
+}
+
+async function withTimeout<T>(millis: number, promise: Promise<T>): Promise<T> {
+    let timeoutPid: NodeJS.Timeout;
+    const timeout = new Promise<never>((_resolve, reject) => {
+        timeoutPid = setTimeout(
+            () => reject(new TimeoutError(`Timed out after ${millis} ms.`)),
+            millis,
+        );
+    });
+
+    return Promise.race([
+        promise,
+        timeout,
+    ]).finally(() => {
+        if (timeoutPid) {
+            clearTimeout(timeoutPid);
+        }
+    });
 }
 
 /**
@@ -47,7 +88,7 @@ async function createMCPSSEClient(
                     headers.set('authorization', `Bearer ${token}`);
                     return fetch(input, { ...init, headers });
                 },
-            // We have to cast to "any" to use it, since it's non-standard
+                // We have to cast to "any" to use it, since it's non-standard
             } as any, // eslint-disable-line @typescript-eslint/no-explicit-any
         });
 
@@ -56,7 +97,7 @@ async function createMCPSSEClient(
         version: '1.0.0',
     });
 
-    await client.connect(transport);
+    await withTimeout(ACTORIZED_MCP_CONNECTION_TIMEOUT_MSEC, client.connect(transport));
 
     return client;
 }
@@ -82,7 +123,7 @@ async function createMCPStreamableClient(
         version: '1.0.0',
     });
 
-    await client.connect(transport);
+    await withTimeout(ACTORIZED_MCP_CONNECTION_TIMEOUT_MSEC, client.connect(transport));
 
     return client;
 }
diff --git a/src/mcp/const.ts b/src/mcp/const.ts
@@ -1,6 +1,7 @@
 export const MAX_TOOL_NAME_LENGTH = 64;
 export const SERVER_ID_LENGTH = 8;
 export const EXTERNAL_TOOL_CALL_TIMEOUT_MSEC = 120_000; // 2 minutes
+export const ACTORIZED_MCP_CONNECTION_TIMEOUT_MSEC = 30_000; // 30 seconds
 
 export const LOG_LEVEL_MAP: Record<string, number> = {
     debug: 0,
diff --git a/src/mcp/server.ts b/src/mcp/server.ts
@@ -551,9 +551,19 @@ export class ActorsMcpServer {
 
                 if (tool.type === 'actor-mcp') {
                     const serverTool = tool.tool as ActorMcpTool;
-                    let client: Client | undefined;
+                    let client: Client | null = null;
                     try {
                         client = await connectMCPClient(serverTool.serverUrl, apifyToken);
+                        if (!client) {
+                            const msg = `Failed to connect to MCP server ${serverTool.serverUrl}`;
+                            log.error(msg);
+                            await this.server.sendLoggingMessage({ level: 'error', data: msg });
+                            return {
+                                content: [
+                                    { type: 'text', text: msg },
+                                ],
+                            };
+                        }
 
                         // Only set up notification handlers if progressToken is provided by the client
                         if (progressToken) {
diff --git a/src/tools/actor.ts b/src/tools/actor.ts
@@ -180,12 +180,12 @@ export async function getNormalActorsAsTools(
 Actor description: ${actorDefinitionPruned.description}
 Instructions: ${ACTOR_ADDITIONAL_INSTRUCTIONS}`,
                         inputSchema: actorDefinitionPruned.input
-                        // So Actor without input schema works - MCP client expects JSON schema valid output
-                        || {
-                            type: 'object',
-                            properties: {},
-                            required: [],
-                        },
+                            // So Actor without input schema works - MCP client expects JSON schema valid output
+                            || {
+                                type: 'object',
+                                properties: {},
+                                required: [],
+                            },
                         // Additional props true to allow skyfire-pay-id
                         ajvValidate: fixedAjvCompile(ajv, { ...actorDefinitionPruned.input, additionalProperties: true }),
                         memoryMbytes: memoryMbytes > ACTOR_MAX_MEMORY_MBYTES ? ACTOR_MAX_MEMORY_MBYTES : memoryMbytes,
@@ -207,21 +207,22 @@ async function getMCPServersAsTools(
     /**
      * This is case for the Skyfire request without any Apify token, we do not support
      * standby Actors in this case so we can skip MCP servers since they would fail anyway (they are standby Actors).
-     */
+    */
     if (apifyToken === null || apifyToken === undefined) {
         return [];
     }
 
-    const actorsMCPServerTools: ToolEntry[] = [];
-    for (const actorInfo of actorsInfo) {
+    // Process all actors in parallel
+    const actorToolPromises = actorsInfo.map(async (actorInfo) => {
         const actorId = actorInfo.actorDefinitionPruned.id;
         if (!actorInfo.webServerMcpPath) {
             log.warning('Actor does not have a web server MCP path, skipping', {
                 actorFullName: actorInfo.actorDefinitionPruned.actorFullName,
                 actorId,
             });
-            continue;
+            return [];
         }
+
         const mcpServerUrl = await getActorMCPServerURL(
             actorInfo.actorDefinitionPruned.id, // Real ID of the Actor
             actorInfo.webServerMcpPath,
@@ -232,17 +233,25 @@ async function getMCPServersAsTools(
             mcpServerUrl,
         });
 
-        let client: Client | undefined;
+        let client: Client | null = null;
         try {
             client = await connectMCPClient(mcpServerUrl, apifyToken);
+            if (!client) {
+                // Skip this Actor, connectMCPClient will log the error
+                return [];
+            }
             const serverTools = await getMCPServerTools(actorId, client, mcpServerUrl);
-            actorsMCPServerTools.push(...serverTools);
+            return serverTools;
         } finally {
             if (client) await client.close();
         }
-    }
+    });
 
-    return actorsMCPServerTools;
+    // Wait for all actors to be processed in parallel
+    const actorToolsArrays = await Promise.all(actorToolPromises);
+
+    // Flatten the arrays of tools
+    return actorToolsArrays.flat();
 }
 
 export async function getActorsAsTools(
@@ -382,10 +391,13 @@ The step parameter enforces this workflow - you cannot call an Actor without fir
                     if (isActorMcpServer) {
                         // MCP server: list tools
                         const mcpServerUrl = mcpServerUrlOrFalse;
-                        let client: Client | undefined;
+                        let client: Client | null = null;
                         // Nested try to ensure client is closed
                         try {
                             client = await connectMCPClient(mcpServerUrl, apifyToken);
+                            if (!client) {
+                                return buildMCPResponse([`Failed to connect to MCP server ${mcpServerUrl}`]);
+                            }
                             const toolsResponse = await client.listTools();
 
                             const toolsInfo = toolsResponse.tools.map((tool) => `**${tool.name}**\n${tool.description || 'No description'}\nInput Schema: ${JSON.stringify(tool.inputSchema, null, 2)}`,
@@ -451,9 +463,12 @@ The step parameter enforces this workflow - you cannot call an Actor without fir
                     }
 
                     const mcpServerUrl = mcpServerUrlOrFalse;
-                    let client: Client | undefined;
+                    let client: Client | null = null;
                     try {
                         client = await connectMCPClient(mcpServerUrl, apifyToken);
+                        if (!client) {
+                            return buildMCPResponse([`Failed to connect to MCP server ${mcpServerUrl}`]);
+                        }
 
                         const result = await client.callTool({
                             name: mcpToolName,
@@ -495,7 +510,7 @@ The step parameter enforces this workflow - you cannot call an Actor without fir
                 if (!callResult) {
                     // Receivers of cancellation notifications SHOULD NOT send a response for the cancelled request
                     // https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/cancellation#behavior-requirements
-                    return { };
+                    return {};
                 }
 
                 const content = buildActorResponseContent(actorName, callResult);
diff --git a/tests/integration/suite.ts b/tests/integration/suite.ts
@@ -398,7 +398,7 @@ export function createIntegrationTestsSuite(
                     limit: 5,
                 },
             });
-            const content = result.content as {text: string}[];
+            const content = result.content as { text: string }[];
             expect(content.some((item) => item.text.includes(ACTOR_PYTHON_EXAMPLE))).toBe(true);
         });
 
@@ -415,7 +415,7 @@ export function createIntegrationTestsSuite(
                     limit: 100,
                 },
             });
-            const content = result.content as {text: string}[];
+            const content = result.content as { text: string }[];
             expect(content.length).toBe(1);
             const outputText = content[0].text;
 
@@ -972,5 +972,11 @@ export function createIntegrationTestsSuite(
 
             await client.close();
         });
+
+        it('should connect to MCP server and at least one tool is available', async () => {
+            client = await createClientFn({ tools: [ACTOR_MCP_SERVER_ACTOR_NAME] });
+            const tools = await client.listTools();
+            expect(tools.tools.length).toBeGreaterThan(0);
+        });
     });
 }

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+export class TimeoutError extends Error {`
	`2`	`+ override readonly name = 'TimeoutError';`
	`3`	`+}`