add actorized mcp servers integration test, internal mcp server mcp client add support for streamable - use it first then fallback to legacy sse

Author: MQ37

src/mcp/client.ts

@@ -1,12 +1,33 @@
 import { Client } from '@modelcontextprotocol/sdk/client/index.js';
 import { SSEClientTransport } from '@modelcontextprotocol/sdk/client/sse.js';
+import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js';
+
+import log from '@apify/log';
 
 import { getMCPServerID } from './utils.js';
 
+/**
+ * Creates and connects a ModelContextProtocol client.
+ * First tries streamable HTTP transport, then falls back to SSE transport.
+ */
+export async function connectMCPClient(
+    url: string, token: string,
+): Promise<Client> {
+    try {
+        return await createMCPStreamableClient(url, token);
+    } catch {
+        // If streamable HTTP transport fails, fall back to SSE transport
+        log.info('Streamable HTTP transport failed, falling back to SSE transport', {
+            url,
+        });
+        return await createMCPSSEClient(url, token);
+    }
+}
+
 /**
  * Creates and connects a ModelContextProtocol client.
  */
-export async function createMCPClient(
+async function createMCPSSEClient(
     url: string, token: string,
 ): Promise<Client> {
     const transport = new SSEClientTransport(
@@ -39,3 +60,29 @@ export async function createMCPClient(
 
     return client;
 }
+
+/**
+ * Creates and connects a ModelContextProtocol client using the streamable HTTP transport.
+ */
+async function createMCPStreamableClient(
+    url: string, token: string,
+): Promise<Client> {
+    const transport = new StreamableHTTPClientTransport(
+        new URL(url),
+        {
+            requestInit: {
+                headers: {
+                    authorization: `Bearer ${token}`,
+                },
+            },
+        });
+
+    const client = new Client({
+        name: getMCPServerID(url),
+        version: '1.0.0',
+    });
+
+    await client.connect(transport);
+
+    return client;
+}

src/mcp/server.ts

@@ -27,7 +27,7 @@ import {
 import { addRemoveTools, callActorGetDataset, defaultTools, getActorsAsTools } from '../tools/index.js';
 import { actorNameToToolName } from '../tools/utils.js';
 import type { ActorMcpTool, ActorTool, HelperTool, ToolEntry } from '../types.js';
-import { createMCPClient } from './client.js';
+import { connectMCPClient } from './client.js';
 import { EXTERNAL_TOOL_CALL_TIMEOUT_MSEC } from './const.js';
 import { processParamsGetTools } from './utils.js';
 
@@ -432,7 +432,7 @@ export class ActorsMcpServer {
                     const serverTool = tool.tool as ActorMcpTool;
                     let client: Client | undefined;
                     try {
-                        client = await createMCPClient(serverTool.serverUrl, apifyToken);
+                        client = await connectMCPClient(serverTool.serverUrl, apifyToken);
                         const res = await client.callTool({
                             name: serverTool.originToolName,
                             arguments: args,

src/tools/actor.ts

@@ -14,7 +14,7 @@ import {
     HelperTools,
 } from '../const.js';
 import { getActorMCPServerPath, getActorMCPServerURL } from '../mcp/actors.js';
-import { createMCPClient } from '../mcp/client.js';
+import { connectMCPClient } from '../mcp/client.js';
 import { getMCPServerTools } from '../mcp/proxy.js';
 import { actorDefinitionPrunedCache } from '../state.js';
 import type { ActorInfo, InternalTool, ToolEntry } from '../types.js';
@@ -172,7 +172,7 @@ async function getMCPServersAsTools(
 
         let client: Client | undefined;
         try {
-            client = await createMCPClient(mcpServerUrl, apifyToken);
+            client = await connectMCPClient(mcpServerUrl, apifyToken);
             const serverTools = await getMCPServerTools(actorId, client, mcpServerUrl);
             actorsMCPServerTools.push(...serverTools);
         } finally {

tests/const.ts

@@ -3,5 +3,6 @@ import { defaultTools } from '../src/tools/index.js';
 import { actorNameToToolName } from '../src/tools/utils.js';
 
 export const ACTOR_PYTHON_EXAMPLE = 'apify/python-example';
+export const ACTOR_MCP_SERVER_ACTOR_NAME = 'apify/actors-mcp-server';
 export const DEFAULT_TOOL_NAMES = defaultTools.map((tool) => tool.tool.name);
 export const DEFAULT_ACTOR_NAMES = defaults.actors.map((tool) => actorNameToToolName(tool));

tests/integration/suite.ts

@@ -6,8 +6,8 @@ import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from
 import { defaults, HelperTools } from '../../src/const.js';
 import { addRemoveTools, defaultTools } from '../../src/tools/index.js';
 import { actorNameToToolName } from '../../src/tools/utils.js';
-import { ACTOR_PYTHON_EXAMPLE, DEFAULT_ACTOR_NAMES, DEFAULT_TOOL_NAMES } from '../const.js';
-import { type McpClientOptions } from '../helpers.js';
+import { ACTOR_MCP_SERVER_ACTOR_NAME, ACTOR_PYTHON_EXAMPLE, DEFAULT_ACTOR_NAMES, DEFAULT_TOOL_NAMES } from '../const.js';
+import { addActor, type McpClientOptions } from '../helpers.js';
 
 interface IntegrationTestsSuiteOptions {
     suiteName: string;
@@ -227,6 +227,38 @@ export function createIntegrationTestsSuite(
             await client.close();
         });
 
+        it('should be able to add and call Actorized MCP server', async () => {
+            const client = await createClientFn({ enableAddingActors: true });
+
+            const toolNamesBefore = getToolNames(await client.listTools());
+            const searchToolCountBefore = toolNamesBefore.filter((name) => name.includes(HelperTools.STORE_SEARCH)).length;
+            expect(searchToolCountBefore).toBe(1);
+
+            // Add self as an Actorized MCP server
+            await addActor(client, ACTOR_MCP_SERVER_ACTOR_NAME);
+
+            const toolNamesAfter = getToolNames(await client.listTools());
+            const searchToolCountAfter = toolNamesAfter.filter((name) => name.includes(HelperTools.STORE_SEARCH)).length;
+            expect(searchToolCountAfter).toBe(2);
+
+            // Find the search tool from the Actorized MCP server
+            const actorizedMCPSearchTool = toolNamesAfter.find(
+                (name) => name.includes(HelperTools.STORE_SEARCH) && name !== HelperTools.STORE_SEARCH);
+            expect(actorizedMCPSearchTool).toBeDefined();
+
+            const result = await client.callTool({
+                name: actorizedMCPSearchTool as string,
+                arguments: {
+                    search: ACTOR_MCP_SERVER_ACTOR_NAME,
+                    limit: 1,
+                },
+            });
+            expect(result.content).toBeDefined();
+
+            await client.close();
+        });
+
+        // Session termination is only possible for streamable HTTP transport.
         it.runIf(options.transport === 'streamable-http')('should successfully terminate streamable session', async () => {
             const client = await createClientFn();
             await client.listTools();