diff --git a/src/examples/client/simpleStreamableHttp.ts b/src/examples/client/simpleStreamableHttp.ts index 739e1164..42ffa28e 100644 --- a/src/examples/client/simpleStreamableHttp.ts +++ b/src/examples/client/simpleStreamableHttp.ts @@ -1,5 +1,6 @@ import { Client } from '../../client/index.js'; import { StreamableHTTPClientTransport } from '../../client/streamableHttp.js'; +import { createInterface } from 'node:readline'; import { ListToolsRequest, ListToolsResultSchema, @@ -15,139 +16,434 @@ import { ResourceListChangedNotificationSchema, } from '../../types.js'; +// Create readline interface for user input +const readline = createInterface({ + input: process.stdin, + output: process.stdout +}); + +// Track received notifications for debugging resumability +let notificationCount = 0; + +// Global client and transport for interactive commands +let client: Client | null = null; +let transport: StreamableHTTPClientTransport | null = null; +let serverUrl = 'http://localhost:3000/mcp'; + async function main(): Promise { - // Create a new client with streamable HTTP transport - const client = new Client({ - name: 'example-client', - version: '1.0.0' - }); + console.log('MCP Interactive Client'); + console.log('====================='); - const transport = new StreamableHTTPClientTransport( - new URL('http://localhost:3000/mcp') - ); + // Connect to server immediately with default settings + await connect(); - // Connect the client using the transport and initialize the server - await client.connect(transport); - console.log('Connected to MCP server'); - - // Set up notification handlers for server-initiated messages - client.setNotificationHandler(LoggingMessageNotificationSchema, (notification) => { - console.log(`Notification received: ${notification.params.level} - ${notification.params.data}`); - }); - client.setNotificationHandler(ResourceListChangedNotificationSchema, async (_) => { - console.log(`Resource list changed notification received!`); - const resourcesRequest: ListResourcesRequest = { - method: 'resources/list', - params: {} - }; - const resourcesResult = await client.request(resourcesRequest, ListResourcesResultSchema); - console.log('Available resources count:', resourcesResult.resources.length); - }); - - // List and call tools - await listTools(client); - - await callGreetTool(client); - await callMultiGreetTool(client); - - - // List available prompts - try { - const promptsRequest: ListPromptsRequest = { - method: 'prompts/list', - params: {} - }; - const promptsResult = await client.request(promptsRequest, ListPromptsResultSchema); - console.log('Available prompts:', promptsResult.prompts); - } catch (error) { - console.log(`Prompts not supported by this server (${error})`); - } - - // Get a prompt - try { - const promptRequest: GetPromptRequest = { - method: 'prompts/get', - params: { - name: 'greeting-template', - arguments: { name: 'MCP User' } - } - }; - const promptResult = await client.request(promptRequest, GetPromptResultSchema); - console.log('Prompt template:', promptResult.messages[0].content.text); - } catch (error) { - console.log(`Prompt retrieval not supported by this server (${error})`); - } - - // List available resources - try { - const resourcesRequest: ListResourcesRequest = { - method: 'resources/list', - params: {} - }; - const resourcesResult = await client.request(resourcesRequest, ListResourcesResultSchema); - console.log('Available resources:', resourcesResult.resources); - } catch (error) { - console.log(`Resources not supported by this server (${error})`); - } - // Keep the connection open to receive notifications - console.log('\nKeeping connection open to receive notifications. Press Ctrl+C to exit.'); + // Print help and start the command loop + printHelp(); + commandLoop(); } -async function listTools(client: Client): Promise { +function printHelp(): void { + console.log('\nAvailable commands:'); + console.log(' connect [url] - Connect to MCP server (default: http://localhost:3000/mcp)'); + console.log(' disconnect - Disconnect from server'); + console.log(' reconnect - Reconnect to the server'); + console.log(' list-tools - List available tools'); + console.log(' call-tool [args] - Call a tool with optional JSON arguments'); + console.log(' greet [name] - Call the greet tool'); + console.log(' multi-greet [name] - Call the multi-greet tool with notifications'); + console.log(' start-notifications [interval] [count] - Start periodic notifications'); + console.log(' simulate-disconnect [delay] - Simulate network disconnection after delay seconds'); + console.log(' list-prompts - List available prompts'); + console.log(' get-prompt [name] [args] - Get a prompt with optional JSON arguments'); + console.log(' list-resources - List available resources'); + console.log(' help - Show this help'); + console.log(' quit - Exit the program'); +} + +function commandLoop(): void { + readline.question('\n> ', async (input) => { + const args = input.trim().split(/\s+/); + const command = args[0]?.toLowerCase(); + + try { + switch (command) { + case 'connect': + await connect(args[1]); + break; + + case 'disconnect': + await disconnect(); + break; + + case 'reconnect': + await reconnect(); + break; + + case 'list-tools': + await listTools(); + break; + + case 'call-tool': + if (args.length < 2) { + console.log('Usage: call-tool [args]'); + } else { + const toolName = args[1]; + let toolArgs = {}; + if (args.length > 2) { + try { + toolArgs = JSON.parse(args.slice(2).join(' ')); + } catch { + console.log('Invalid JSON arguments. Using empty args.'); + } + } + await callTool(toolName, toolArgs); + } + break; + + case 'greet': + await callGreetTool(args[1] || 'MCP User'); + break; + + case 'multi-greet': + await callMultiGreetTool(args[1] || 'MCP User'); + break; + + case 'start-notifications': { + const interval = args[1] ? parseInt(args[1], 10) : 2000; + const count = args[2] ? parseInt(args[2], 10) : 0; + await startNotifications(interval, count); + break; + } + + case 'simulate-disconnect': { + const delay = args[1] ? parseInt(args[1], 10) : 5; + simulateDisconnect(delay); + break; + } + + case 'list-prompts': + await listPrompts(); + break; + + case 'get-prompt': + if (args.length < 2) { + console.log('Usage: get-prompt [args]'); + } else { + const promptName = args[1]; + let promptArgs = {}; + if (args.length > 2) { + try { + promptArgs = JSON.parse(args.slice(2).join(' ')); + } catch { + console.log('Invalid JSON arguments. Using empty args.'); + } + } + await getPrompt(promptName, promptArgs); + } + break; + + case 'list-resources': + await listResources(); + break; + + case 'help': + printHelp(); + break; + + case 'quit': + case 'exit': + await cleanup(); + return; + + default: + if (command) { + console.log(`Unknown command: ${command}`); + } + break; + } + } catch (error) { + console.error(`Error executing command: ${error}`); + } + + // Continue the command loop + commandLoop(); + }); +} + +async function connect(url?: string): Promise { + if (client) { + console.log('Already connected. Disconnect first.'); + return; + } + + if (url) { + serverUrl = url; + } + + console.log(`Connecting to ${serverUrl}...`); + + try { + // Create a new client + client = new Client({ + name: 'example-client', + version: '1.0.0' + }); + + transport = new StreamableHTTPClientTransport( + new URL(serverUrl) + ); + + // Set up notification handlers + client.setNotificationHandler(LoggingMessageNotificationSchema, (notification) => { + notificationCount++; + console.log(`\nNotification #${notificationCount}: ${notification.params.level} - ${notification.params.data}`); + // Re-display the prompt + process.stdout.write('> '); + }); + + client.setNotificationHandler(ResourceListChangedNotificationSchema, async (_) => { + console.log(`\nResource list changed notification received!`); + try { + if (!client) { + console.log('Client disconnected, cannot fetch resources'); + return; + } + const resourcesResult = await client.request({ + method: 'resources/list', + params: {} + }, ListResourcesResultSchema); + console.log('Available resources count:', resourcesResult.resources.length); + } catch { + console.log('Failed to list resources after change notification'); + } + // Re-display the prompt + process.stdout.write('> '); + }); + + // Connect the client + await client.connect(transport); + console.log('Connected to MCP server'); + } catch (error) { + console.error('Failed to connect:', error); + client = null; + transport = null; + } +} + +async function disconnect(): Promise { + if (!client || !transport) { + console.log('Not connected.'); + return; + } + + try { + await transport.close(); + console.log('Disconnected from MCP server'); + client = null; + transport = null; + } catch (error) { + console.error('Error disconnecting:', error); + } +} + +async function reconnect(): Promise { + if (client) { + await disconnect(); + } + await connect(); +} + +async function listTools(): Promise { + if (!client) { + console.log('Not connected to server.'); + return; + } + try { const toolsRequest: ListToolsRequest = { method: 'tools/list', params: {} }; const toolsResult = await client.request(toolsRequest, ListToolsResultSchema); - console.log('Available tools:', toolsResult.tools); + + console.log('Available tools:'); if (toolsResult.tools.length === 0) { - console.log('No tools available from the server'); + 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})`); - return } } -async function callGreetTool(client: Client): Promise { - try { - const greetRequest: CallToolRequest = { - method: 'tools/call', - params: { - name: 'greet', - arguments: { name: 'MCP User' } - } - }; - const greetResult = await client.request(greetRequest, CallToolResultSchema); - console.log('Greeting result:', greetResult.content[0].text); - } catch (error) { - console.log(`Error calling greet tool: ${error}`); +async function callTool(name: string, args: Record): Promise { + if (!client) { + console.log('Not connected to server.'); + return; } -} -async function callMultiGreetTool(client: Client): Promise { try { - console.log('\nCalling multi-greet tool (with notifications)...'); - const multiGreetRequest: CallToolRequest = { + const request: CallToolRequest = { method: 'tools/call', params: { - name: 'multi-greet', - arguments: { name: 'MCP User' } + name, + arguments: args } }; - const multiGreetResult = await client.request(multiGreetRequest, CallToolResultSchema); - console.log('Multi-greet results:'); - multiGreetResult.content.forEach(item => { + + console.log(`Calling tool '${name}' with args:`, args); + const result = await client.request(request, CallToolResultSchema); + + console.log('Tool result:'); + result.content.forEach(item => { if (item.type === 'text') { - console.log(`- ${item.text}`); + console.log(` ${item.text}`); + } else { + console.log(` ${item.type} content:`, item); } }); } catch (error) { - console.log(`Error calling multi-greet tool: ${error}`); + console.log(`Error calling tool ${name}: ${error}`); } } +async function callGreetTool(name: string): Promise { + await callTool('greet', { name }); +} + +async function callMultiGreetTool(name: string): Promise { + console.log('Calling multi-greet tool with notifications...'); + await callTool('multi-greet', { name }); +} + +async function startNotifications(interval: number, count: number): Promise { + console.log(`Starting notification stream: interval=${interval}ms, count=${count || 'unlimited'}`); + await callTool('start-notification-stream', { interval, count }); +} + +function simulateDisconnect(delaySeconds: number): void { + if (!transport) { + console.log('Not connected.'); + return; + } + + console.log(`Will simulate network disconnect in ${delaySeconds} seconds...`); + setTimeout(async () => { + if (!transport) return; + + console.log('\nSimulating network disconnect...'); + await transport.close(); + console.log('Transport closed. Client should automatically attempt to reconnect...'); + + // Keep the client object but null the transport + // This simulates a network disconnection without explicit user disconnect + transport = null; + + // Re-display the prompt + process.stdout.write('> '); + }, delaySeconds * 1000); +} + +async function listPrompts(): Promise { + if (!client) { + console.log('Not connected to server.'); + return; + } + + try { + const promptsRequest: ListPromptsRequest = { + method: 'prompts/list', + params: {} + }; + const promptsResult = await client.request(promptsRequest, ListPromptsResultSchema); + console.log('Available prompts:'); + if (promptsResult.prompts.length === 0) { + console.log(' No prompts available'); + } else { + for (const prompt of promptsResult.prompts) { + console.log(` - ${prompt.name}: ${prompt.description}`); + } + } + } catch (error) { + console.log(`Prompts not supported by this server (${error})`); + } +} + +async function getPrompt(name: string, args: Record): Promise { + if (!client) { + console.log('Not connected to server.'); + return; + } + + try { + const promptRequest: GetPromptRequest = { + method: 'prompts/get', + params: { + name, + arguments: args as Record + } + }; + + const promptResult = await client.request(promptRequest, GetPromptResultSchema); + console.log('Prompt template:'); + promptResult.messages.forEach((msg, index) => { + console.log(` [${index + 1}] ${msg.role}: ${msg.content.text}`); + }); + } catch (error) { + console.log(`Error getting prompt ${name}: ${error}`); + } +} + +async function listResources(): Promise { + if (!client) { + console.log('Not connected to server.'); + return; + } + + try { + const resourcesRequest: ListResourcesRequest = { + method: 'resources/list', + params: {} + }; + const resourcesResult = await client.request(resourcesRequest, ListResourcesResultSchema); + + console.log('Available resources:'); + if (resourcesResult.resources.length === 0) { + console.log(' No resources available'); + } else { + for (const resource of resourcesResult.resources) { + console.log(` - ${resource.name}: ${resource.uri}`); + } + } + } catch (error) { + console.log(`Resources not supported by this server (${error})`); + } +} + +async function cleanup(): Promise { + if (client && transport) { + try { + await transport.close(); + } catch (error) { + console.error('Error closing transport:', error); + } + } + + readline.close(); + console.log('\nGoodbye!'); + process.exit(0); +} + +// Handle Ctrl+C +process.on('SIGINT', async () => { + console.log('\nReceived SIGINT. Cleaning up...'); + await cleanup(); +}); + +// Start the interactive client main().catch((error: unknown) => { console.error('Error running MCP client:', error); process.exit(1); diff --git a/src/examples/server/simpleStreamableHttp.ts b/src/examples/server/simpleStreamableHttp.ts index f0f74439..a45b3158 100644 --- a/src/examples/server/simpleStreamableHttp.ts +++ b/src/examples/server/simpleStreamableHttp.ts @@ -1,9 +1,75 @@ import express, { Request, Response } from 'express'; import { randomUUID } from 'node:crypto'; import { McpServer } from '../../server/mcp.js'; -import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js'; +import { EventStore, StreamableHTTPServerTransport } from '../../server/streamableHttp.js'; import { z } from 'zod'; -import { CallToolResult, GetPromptResult, ReadResourceResult } from '../../types.js'; +import { CallToolResult, GetPromptResult, JSONRPCMessage, ReadResourceResult } from '../../types.js'; + +// Create a simple in-memory EventStore for resumability +class InMemoryEventStore implements EventStore { + private events: Map = new Map(); + + /** + * Generates a unique event ID for a given stream ID + */ + generateEventId(streamId: string): string { + return `${streamId}_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`; + } + + getStreamIdFromEventId(eventId: string): string { + const parts = eventId.split('_'); + return parts.length > 0 ? parts[0] : ''; + } + + + /** + * Stores an event with a generated event ID + */ + async storeEvent(streamId: string, message: JSONRPCMessage): Promise { + const eventId = this.generateEventId(streamId); + console.log(`Storing event ${eventId} for stream ${streamId}`); + this.events.set(eventId, { streamId, message }); + return eventId; + } + + /** + * Retrieves events that occurred after a specific event + */ + async getEventsAfter(lastEventId: string): Promise> { + if (!lastEventId || !this.events.has(lastEventId)) { + return []; + } + + // Extract the stream ID from the event ID + const streamId = lastEventId.split('_')[0]; + const result: Array<{ eventId: string, message: JSONRPCMessage }> = []; + let foundLastEvent = false; + + // Sort events by eventId for chronological ordering + const sortedEvents = [...this.events.entries()].sort((a, b) => a[0].localeCompare(b[0])); + + for (const [eventId, { streamId: eventStreamId, message }] of sortedEvents) { + // Only include events from the same stream + if (eventStreamId !== streamId) { + continue; + } + + // Start collecting events after we find the lastEventId + if (eventId === lastEventId) { + foundLastEvent = true; + continue; + } + + if (foundLastEvent) { + result.push({ eventId, message }); + } + } + + console.log(`Found ${result.length} events after ${lastEventId} for replay`); + return result; + } + +} // Create an MCP server with implementation details const server = new McpServer({ @@ -92,6 +158,43 @@ server.prompt( } ); +// Register a tool specifically for testing resumability +server.tool( + 'start-notification-stream', + 'Starts sending periodic notifications for testing resumability', + { + interval: z.number().describe('Interval in milliseconds between notifications').default(100), + count: z.number().describe('Number of notifications to send (0 for 100)').default(50), + }, + async ({ interval, count }, { sendNotification }): Promise => { + const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); + let counter = 0; + + while (count === 0 || counter < count) { + counter++; + await sendNotification({ + method: "notifications/message", + params: { + level: "info", + data: `Periodic notification #${counter} at ${new Date().toISOString()}` + } + }); + + // Wait for the specified interval + await sleep(interval); + } + + return { + content: [ + { + type: 'text', + text: `Started sending periodic notifications every ${interval}ms`, + } + ], + }; + } +); + // Create a simple resource at a fixed URI server.resource( 'greeting-resource', @@ -127,8 +230,10 @@ app.post('/mcp', async (req: Request, res: Response) => { transport = transports[sessionId]; } else if (!sessionId && isInitializeRequest(req.body)) { // New initialization request + const eventStore = new InMemoryEventStore(); transport = new StreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), + eventStore, // Enable resumability }); // Connect the transport to the MCP server BEFORE handling the request @@ -182,7 +287,14 @@ app.get('/mcp', async (req: Request, res: Response) => { return; } - console.log(`Establishing SSE stream for session ${sessionId}`); + // Check for Last-Event-ID header for resumability + const lastEventId = req.headers['last-event-id'] as string | undefined; + if (lastEventId) { + console.log(`Client reconnecting with Last-Event-ID: ${lastEventId}`); + } else { + console.log(`Establishing new SSE stream for session ${sessionId}`); + } + const transport = transports[sessionId]; await transport.handleRequest(req, res); });