From 70608b58762fc92c5b615f15685f702eaeefd536 Mon Sep 17 00:00:00 2001 From: ihrpr Date: Mon, 21 Apr 2025 11:54:10 +0100 Subject: [PATCH] Add more test clients --- .../client/multipleClientsParallel.ts | 159 +++++++++++ .../client/parallelToolCallsClient.ts | 197 ++++++++++++++ .../streamableHttpBatchMessages.test.ts | 252 ++++++++++++++++++ 3 files changed, 608 insertions(+) create mode 100644 src/examples/client/multipleClientsParallel.ts create mode 100644 src/examples/client/parallelToolCallsClient.ts create mode 100644 src/integration-tests/streamableHttpBatchMessages.test.ts diff --git a/src/examples/client/multipleClientsParallel.ts b/src/examples/client/multipleClientsParallel.ts new file mode 100644 index 00000000..90b9d4ea --- /dev/null +++ b/src/examples/client/multipleClientsParallel.ts @@ -0,0 +1,159 @@ +import { Client } from '../../client/index.js'; +import { StreamableHTTPClientTransport } from '../../client/streamableHttp.js'; +import { + CallToolRequest, + CallToolResultSchema, + LoggingMessageNotificationSchema, +} from '../../types.js'; + +/** + * Multiple Clients MCP Example + * + * This client demonstrates how to: + * 1. Create multiple MCP clients in parallel + * 2. Each client calls a single tool + * 3. Track notifications from each client independently + */ + +// Command line args processing +const args = process.argv.slice(2); +const serverUrl = args[0] || 'http://localhost:3000/mcp'; + +interface ClientConfig { + id: string; + name: string; + toolName: string; + toolArguments: Record; +} + +async function createAndRunClient(config: ClientConfig): Promise<{ id: string; result: any }> { + console.log(`[${config.id}] Creating client: ${config.name}`); + + const client = new Client({ + name: config.name, + version: '1.0.0' + }); + + const transport = new StreamableHTTPClientTransport(new URL(serverUrl)); + + // Set up client-specific error handler + client.onerror = (error) => { + console.error(`[${config.id}] Client error:`, error); + }; + + // Set up client-specific notification handler + client.setNotificationHandler(LoggingMessageNotificationSchema, (notification) => { + console.log(`[${config.id}] Notification: ${notification.params.data}`); + }); + + try { + // Connect to the server + await client.connect(transport); + console.log(`[${config.id}] Connected to MCP server`); + + // Call the specified tool + console.log(`[${config.id}] Calling tool: ${config.toolName}`); + const toolRequest: CallToolRequest = { + method: 'tools/call', + params: { + name: config.toolName, + arguments: { + ...config.toolArguments, + // Add client ID to arguments for identification in notifications + caller: config.id + } + } + }; + + const result = await client.request(toolRequest, CallToolResultSchema); + console.log(`[${config.id}] Tool call completed`); + + // Keep the connection open for a bit to receive notifications + await new Promise(resolve => setTimeout(resolve, 5000)); + + // Disconnect + await transport.close(); + console.log(`[${config.id}] Disconnected from MCP server`); + + return { id: config.id, result }; + } catch (error) { + console.error(`[${config.id}] Error:`, error); + throw error; + } +} + +async function main(): Promise { + console.log('MCP Multiple Clients Example'); + console.log('============================'); + console.log(`Server URL: ${serverUrl}`); + console.log(''); + + try { + // Define client configurations + const clientConfigs: ClientConfig[] = [ + { + id: 'client1', + name: 'basic-client-1', + toolName: 'start-notification-stream', + toolArguments: { + interval: 3, // 1 second between notifications + count: 5 // Send 5 notifications + } + }, + { + id: 'client2', + name: 'basic-client-2', + toolName: 'start-notification-stream', + toolArguments: { + interval: 2, // 2 seconds between notifications + count: 3 // Send 3 notifications + } + }, + { + id: 'client3', + name: 'basic-client-3', + toolName: 'start-notification-stream', + toolArguments: { + interval: 1, // 0.5 second between notifications + count: 8 // Send 8 notifications + } + } + ]; + + // Start all clients in parallel + console.log(`Starting ${clientConfigs.length} clients in parallel...`); + console.log(''); + + const clientPromises = clientConfigs.map(config => createAndRunClient(config)); + const results = await Promise.all(clientPromises); + + // Display results from all clients + console.log('\n=== Final Results ==='); + results.forEach(({ id, result }) => { + console.log(`\n[${id}] Tool result:`); + if (Array.isArray(result.content)) { + result.content.forEach((item: { type: string; text?: string }) => { + if (item.type === 'text' && item.text) { + console.log(` ${item.text}`); + } else { + console.log(` ${item.type} content:`, item); + } + }); + } else { + console.log(` Unexpected result format:`, result); + } + }); + + console.log('\n=== All clients completed successfully ==='); + + } catch (error) { + console.error('Error running multiple clients:', error); + process.exit(1); + } +} + +// Start the example +main().catch((error: unknown) => { + console.error('Error running MCP multiple clients example:', error); + process.exit(1); +}); \ No newline at end of file diff --git a/src/examples/client/parallelToolCallsClient.ts b/src/examples/client/parallelToolCallsClient.ts new file mode 100644 index 00000000..8a53ade9 --- /dev/null +++ b/src/examples/client/parallelToolCallsClient.ts @@ -0,0 +1,197 @@ +import { Client } from '../../client/index.js'; +import { StreamableHTTPClientTransport } from '../../client/streamableHttp.js'; +import { + ListToolsRequest, + ListToolsResultSchema, + CallToolRequest, + CallToolResultSchema, + LoggingMessageNotificationSchema, +} from '../../types.js'; + +/** + * Parallel Tool Calls MCP Client + * + * This client demonstrates how to: + * 1. Start multiple tool calls in parallel + * 2. Track notifications from each tool call using a caller parameter + */ + +// Command line args processing +const args = process.argv.slice(2); +const serverUrl = args[0] || 'http://localhost:3000/mcp'; + +async function main(): Promise { + console.log('MCP Parallel Tool Calls Client'); + console.log('=============================='); + console.log(`Connecting to server at: ${serverUrl}`); + + let client: Client; + let transport: StreamableHTTPClientTransport; + + try { + // Create client with streamable HTTP transport + client = new Client({ + name: 'parallel-tool-calls-client', + version: '1.0.0' + }); + + client.onerror = (error) => { + console.error('Client error:', error); + }; + + // Connect to the server + transport = new StreamableHTTPClientTransport(new URL(serverUrl)); + await client.connect(transport); + console.log('Successfully connected to MCP server'); + + // Set up notification handler with caller identification + client.setNotificationHandler(LoggingMessageNotificationSchema, (notification) => { + console.log(`Notification: ${notification.params.data}`); + }); + + console.log("List tools") + const toolsRequest = await listTools(client); + console.log("Tools: ", toolsRequest) + + + // 2. Start multiple notification tools in parallel + console.log('\n=== Starting Multiple Notification Streams in Parallel ==='); + const toolResults = await startParallelNotificationTools(client); + + // Log the results from each tool call + for (const [caller, result] of Object.entries(toolResults)) { + console.log(`\n=== Tool result for ${caller} ===`); + result.content.forEach((item: { type: string; text: any; }) => { + if (item.type === 'text') { + console.log(` ${item.text}`); + } else { + console.log(` ${item.type} content:`, item); + } + }); + } + + // 3. Wait for all notifications (10 seconds) + console.log('\n=== Waiting for all notifications ==='); + await new Promise(resolve => setTimeout(resolve, 10000)); + + // 4. Disconnect + console.log('\n=== Disconnecting ==='); + await transport.close(); + console.log('Disconnected from MCP server'); + + } catch (error) { + console.error('Error running client:', error); + process.exit(1); + } +} + +/** + * List available tools on the server + */ +async function listTools(client: Client): Promise { + try { + const toolsRequest: ListToolsRequest = { + method: 'tools/list', + params: {} + }; + const toolsResult = await client.request(toolsRequest, ListToolsResultSchema); + + console.log('Available tools:'); + if (toolsResult.tools.length === 0) { + console.log(' No tools available'); + } else { + for (const tool of toolsResult.tools) { + console.log(` - ${tool.name}: ${tool.description}`); + } + } + } catch (error) { + console.log(`Tools not supported by this server: ${error}`); + } +} + +/** + * Start multiple notification tools in parallel with different configurations + * Each tool call includes a caller parameter to identify its notifications + */ +async function startParallelNotificationTools(client: Client): Promise> { + try { + // Define multiple tool calls with different configurations + const toolCalls = [ + { + caller: 'fast-notifier', + request: { + method: 'tools/call', + params: { + name: 'start-notification-stream', + arguments: { + interval: 2, // 0.5 second between notifications + count: 10, // Send 10 notifications + caller: 'fast-notifier' // Identify this tool call + } + } + } + }, + { + caller: 'slow-notifier', + request: { + method: 'tools/call', + params: { + name: 'start-notification-stream', + arguments: { + interval: 5, // 2 seconds between notifications + count: 5, // Send 5 notifications + caller: 'slow-notifier' // Identify this tool call + } + } + } + }, + { + caller: 'burst-notifier', + request: { + method: 'tools/call', + params: { + name: 'start-notification-stream', + arguments: { + interval: 1, // 0.1 second between notifications + count: 3, // Send just 3 notifications + caller: 'burst-notifier' // Identify this tool call + } + } + } + } + ]; + + console.log(`Starting ${toolCalls.length} notification tools in parallel...`); + + // Start all tool calls in parallel + const toolPromises = toolCalls.map(({ caller, request }) => { + console.log(`Starting tool call for ${caller}...`); + return client.request(request, CallToolResultSchema) + .then(result => ({ caller, result })) + .catch(error => { + console.error(`Error in tool call for ${caller}:`, error); + throw error; + }); + }); + + // Wait for all tool calls to complete + const results = await Promise.all(toolPromises); + + // Organize results by caller + const resultsByTool: Record = {}; + results.forEach(({ caller, result }) => { + resultsByTool[caller] = result; + }); + + return resultsByTool; + } catch (error) { + console.error(`Error starting parallel notification tools:`, error); + throw error; + } +} + +// Start the client +main().catch((error: unknown) => { + console.error('Error running MCP client:', error); + process.exit(1); +}); \ No newline at end of file diff --git a/src/integration-tests/streamableHttpBatchMessages.test.ts b/src/integration-tests/streamableHttpBatchMessages.test.ts new file mode 100644 index 00000000..dc79e3de --- /dev/null +++ b/src/integration-tests/streamableHttpBatchMessages.test.ts @@ -0,0 +1,252 @@ +/** + * Integration tests for batch messaging in StreamableHttp transport + * + * This test suite focuses specifically on the requirement from the spec: + * "The body of the POST request MUST be one of the following: + * - A single JSON-RPC request, notification, or response + * - An array batching one or more requests and/or notifications + * - An array batching one or more responses" + */ + +import { createServer, type Server } from "node:http"; +import { AddressInfo } from "node:net"; +import { randomUUID } from "node:crypto"; +import { StreamableHTTPServerTransport } from "../server/streamableHttp.js"; +import { McpServer } from "../server/mcp.js"; +import { Client } from "../client/index.js"; +import { CallToolResult, CallToolResultSchema, JSONRPCMessage } from "../types.js"; +import { z } from "zod"; +import { StreamableHTTPClientTransport } from '../client/streamableHttp.js'; +describe("StreamableHttp Batch Messaging - SSE", () => { + let server: Server; + let client: Client; + + // Just a basic server and client for testing + beforeEach(async () => { + // Create MCP server with test tools + const mcpServer = new McpServer( + { name: "batch-test-server", version: "1.0.0" }, + { capabilities: { logging: {} } } + ); + + // Simple greeting tool + mcpServer.tool( + "greet", + "A simple greeting tool", + { name: z.string().describe("Name to greet") }, + async ({ name }): Promise => { + return { content: [{ type: "text", text: `Hello, ${name}!` }] }; + } + ); + + // Server transport with session management + const serverTransport = new StreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + }); + + // Connect transport to server + await mcpServer.connect(serverTransport); + + // Create HTTP server + server = createServer(async (req, res) => { + await serverTransport.handleRequest(req, res); + }); + + // Start server on random port + const baseUrl = await new Promise((resolve) => { + server.listen(0, '127.0.0.1', () => { + const addr = server.address() as AddressInfo; + resolve(new URL(`http://127.0.0.1:${addr.port}`)); + }); + }); + + // Create client that connects to the server + const transport = new StreamableHTTPClientTransport(baseUrl); + client = new Client({ name: "test-client", version: "1.0.0" }); + await client.connect(transport); + + }); + + afterEach(async () => { + await client.close(); + await new Promise((resolve) => server.close(() => resolve())); + }); + + /** + * 1. Test sending a single JSON-RPC request + */ + it("handles a single request", async () => { + const result = await client.callTool({ + name: "greet", + arguments: { + name: "user" + } + }); + + expect(result).toHaveProperty("content"); + expect(Array.isArray(result.content)).toBe(true); + expect(result.content).toHaveLength(1); + + // Type assertion to handle the unknown type + const content = result.content as Array<{ type: string, text: string }>; + expect(content[0]).toHaveProperty("type", "text"); + expect(content[0]).toHaveProperty("text", "Hello, user!"); + + }); + + /** + * 1. Test sending a single JSON-RPC request + */ + it("handles a single request with request", async () => { + const result = await client.request( + { + method: "tools/call", + params: { + name: "greet", + arguments: { + name: "user" + } + }, + }, + CallToolResultSchema, + ); + + + expect(result).toHaveProperty("content"); + expect(Array.isArray(result.content)).toBe(true); + expect(result.content).toHaveLength(1); + + // Type assertion to handle the unknown type + const content = result.content as Array<{ type: string, text: string }>; + expect(content[0]).toHaveProperty("type", "text"); + expect(content[0]).toHaveProperty("text", "Hello, user!"); + + }); + + // it("handles an array batching one or more requests and/or notifications", async () => { + // const result = await client.request({ + + // }); + + // }); + + +}); + +describe("StreamableHttp Batch Messaging - JSON response", () => { + let server: Server; + let client: Client; + + // Just a basic server and client for testing + beforeEach(async () => { + // Create MCP server with test tools + const mcpServer = new McpServer( + { name: "batch-test-server", version: "1.0.0" }, + { capabilities: { logging: {} } } + ); + + // Simple greeting tool + mcpServer.tool( + "greet", + "A simple greeting tool", + { name: z.string().describe("Name to greet") }, + async ({ name }): Promise => { + return { content: [{ type: "text", text: `Hello, ${name}!` }] }; + } + ); + + // Server transport with session management + const serverTransport = new StreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + enableJsonResponse: true, + + }); + + // Connect transport to server + await mcpServer.connect(serverTransport); + + // Create HTTP server + server = createServer(async (req, res) => { + await serverTransport.handleRequest(req, res); + }); + + // Start server on random port + const baseUrl = await new Promise((resolve) => { + server.listen(0, '127.0.0.1', () => { + const addr = server.address() as AddressInfo; + resolve(new URL(`http://127.0.0.1:${addr.port}`)); + }); + }); + + // Create client that connects to the server + const transport = new StreamableHTTPClientTransport(baseUrl); + client = new Client({ name: "test-client", version: "1.0.0" }); + await client.connect(transport); + + }); + + afterEach(async () => { + await client.close(); + await new Promise((resolve) => server.close(() => resolve())); + }); + + /** + * 1. Test sending a single JSON-RPC request + */ + it("handles a single request", async () => { + const result = await client.callTool({ + name: "greet", + arguments: { + name: "user" + } + }); + + expect(result).toHaveProperty("content"); + expect(Array.isArray(result.content)).toBe(true); + expect(result.content).toHaveLength(1); + + // Type assertion to handle the unknown type + const content = result.content as Array<{ type: string, text: string }>; + expect(content[0]).toHaveProperty("type", "text"); + expect(content[0]).toHaveProperty("text", "Hello, user!"); + + }); + + /** + * 1. Test sending a single JSON-RPC request + */ + it("handles a single request with request", async () => { + const result = await client.request( + { + method: "tools/call", + params: { + name: "greet", + arguments: { + name: "user" + } + }, + }, + CallToolResultSchema, + ); + + + expect(result).toHaveProperty("content"); + expect(Array.isArray(result.content)).toBe(true); + expect(result.content).toHaveLength(1); + + // Type assertion to handle the unknown type + const content = result.content as Array<{ type: string, text: string }>; + expect(content[0]).toHaveProperty("type", "text"); + expect(content[0]).toHaveProperty("text", "Hello, user!"); + + }); + + // it("handles an array batching one or more requests and/or notifications", async () => { + // const result = await client.request({ + + // }); + + // }); + + +}); \ No newline at end of file