diff --git a/src/examples/server/elicitationFormExample.ts b/src/examples/server/elicitationFormExample.ts index 6c080094..d220806d 100644 --- a/src/examples/server/elicitationFormExample.ts +++ b/src/examples/server/elicitationFormExample.ts @@ -14,308 +14,314 @@ import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js'; import { isInitializeRequest } from '../../types.js'; import { createMcpExpressApp } from '../../server/express.js'; -// Create MCP server - it will automatically use AjvJsonSchemaValidator with sensible defaults -// The validator supports format validation (email, date, etc.) if ajv-formats is installed -const mcpServer = new McpServer( - { - name: 'form-elicitation-example-server', - version: '1.0.0' - }, - { - capabilities: {} - } -); - -/** - * Example 1: Simple user registration tool - * Collects username, email, and password from the user - */ -mcpServer.registerTool( - 'register_user', - { - description: 'Register a new user account by collecting their information', - inputSchema: {} - }, - async () => { - try { - // Request user information through form elicitation - const result = await mcpServer.server.elicitInput({ - mode: 'form', - message: 'Please provide your registration information:', - requestedSchema: { - type: 'object', - properties: { - username: { - type: 'string', - title: 'Username', - description: 'Your desired username (3-20 characters)', - minLength: 3, - maxLength: 20 - }, - email: { - type: 'string', - title: 'Email', - description: 'Your email address', - format: 'email' - }, - password: { - type: 'string', - title: 'Password', - description: 'Your password (min 8 characters)', - minLength: 8 - }, - newsletter: { - type: 'boolean', - title: 'Newsletter', - description: 'Subscribe to newsletter?', - default: false - } - }, - required: ['username', 'email', 'password'] - } - }); - - // Handle the different possible actions - if (result.action === 'accept' && result.content) { - const { username, email, newsletter } = result.content as { - username: string; - email: string; - password: string; - newsletter?: boolean; - }; - - return { - content: [ - { - type: 'text', - text: `Registration successful!\n\nUsername: ${username}\nEmail: ${email}\nNewsletter: ${newsletter ? 'Yes' : 'No'}` - } - ] - }; - } else if (result.action === 'decline') { - return { - content: [ - { - type: 'text', - text: 'Registration cancelled by user.' - } - ] - }; - } else { - return { - content: [ - { - type: 'text', - text: 'Registration was cancelled.' - } - ] - }; - } - } catch (error) { - return { - content: [ - { - type: 'text', - text: `Registration failed: ${error instanceof Error ? error.message : String(error)}` - } - ], - isError: true - }; +// Factory to create a new MCP server per session. +// Each session needs its own server+transport pair to avoid cross-session contamination. +const getServer = () => { + // Create MCP server - it will automatically use AjvJsonSchemaValidator with sensible defaults + // The validator supports format validation (email, date, etc.) if ajv-formats is installed + const mcpServer = new McpServer( + { + name: 'form-elicitation-example-server', + version: '1.0.0' + }, + { + capabilities: {} } - } -); + ); -/** - * Example 2: Multi-step workflow with multiple form elicitation requests - * Demonstrates how to collect information in multiple steps - */ -mcpServer.registerTool( - 'create_event', - { - description: 'Create a calendar event by collecting event details', - inputSchema: {} - }, - async () => { - try { - // Step 1: Collect basic event information - const basicInfo = await mcpServer.server.elicitInput({ - mode: 'form', - message: 'Step 1: Enter basic event information', - requestedSchema: { - type: 'object', - properties: { - title: { - type: 'string', - title: 'Event Title', - description: 'Name of the event', - minLength: 1 + /** + * Example 1: Simple user registration tool + * Collects username, email, and password from the user + */ + mcpServer.registerTool( + 'register_user', + { + description: 'Register a new user account by collecting their information', + inputSchema: {} + }, + async () => { + try { + // Request user information through form elicitation + const result = await mcpServer.server.elicitInput({ + mode: 'form', + message: 'Please provide your registration information:', + requestedSchema: { + type: 'object', + properties: { + username: { + type: 'string', + title: 'Username', + description: 'Your desired username (3-20 characters)', + minLength: 3, + maxLength: 20 + }, + email: { + type: 'string', + title: 'Email', + description: 'Your email address', + format: 'email' + }, + password: { + type: 'string', + title: 'Password', + description: 'Your password (min 8 characters)', + minLength: 8 + }, + newsletter: { + type: 'boolean', + title: 'Newsletter', + description: 'Subscribe to newsletter?', + default: false + } }, - description: { - type: 'string', - title: 'Description', - description: 'Event description (optional)' - } - }, - required: ['title'] - } - }); - - if (basicInfo.action !== 'accept' || !basicInfo.content) { - return { - content: [{ type: 'text', text: 'Event creation cancelled.' }] - }; - } - - // Step 2: Collect date and time - const dateTime = await mcpServer.server.elicitInput({ - mode: 'form', - message: 'Step 2: Enter date and time', - requestedSchema: { - type: 'object', - properties: { - date: { - type: 'string', - title: 'Date', - description: 'Event date', - format: 'date' - }, - startTime: { - type: 'string', - title: 'Start Time', - description: 'Event start time (HH:MM)' - }, - duration: { - type: 'integer', - title: 'Duration', - description: 'Duration in minutes', - minimum: 15, - maximum: 480 - } - }, - required: ['date', 'startTime', 'duration'] - } - }); - - if (dateTime.action !== 'accept' || !dateTime.content) { - return { - content: [{ type: 'text', text: 'Event creation cancelled.' }] - }; - } - - // Combine all collected information - const event = { - ...basicInfo.content, - ...dateTime.content - }; - - return { - content: [ - { - type: 'text', - text: `Event created successfully!\n\n${JSON.stringify(event, null, 2)}` + required: ['username', 'email', 'password'] } - ] - }; - } catch (error) { - return { - content: [ - { - type: 'text', - text: `Event creation failed: ${error instanceof Error ? error.message : String(error)}` - } - ], - isError: true - }; - } - } -); + }); -/** - * Example 3: Collecting address information - * Demonstrates validation with patterns and optional fields - */ -mcpServer.registerTool( - 'update_shipping_address', - { - description: 'Update shipping address with validation', - inputSchema: {} - }, - async () => { - try { - const result = await mcpServer.server.elicitInput({ - mode: 'form', - message: 'Please provide your shipping address:', - requestedSchema: { - type: 'object', - properties: { - name: { - type: 'string', - title: 'Full Name', - description: 'Recipient name', - minLength: 1 - }, - street: { - type: 'string', - title: 'Street Address', - minLength: 1 - }, - city: { - type: 'string', - title: 'City', - minLength: 1 - }, - state: { - type: 'string', - title: 'State/Province', - minLength: 2, - maxLength: 2 - }, - zipCode: { - type: 'string', - title: 'ZIP/Postal Code', - description: '5-digit ZIP code' - }, - phone: { - type: 'string', - title: 'Phone Number (optional)', - description: 'Contact phone number' - } - }, - required: ['name', 'street', 'city', 'state', 'zipCode'] + // Handle the different possible actions + if (result.action === 'accept' && result.content) { + const { username, email, newsletter } = result.content as { + username: string; + email: string; + password: string; + newsletter?: boolean; + }; + + return { + content: [ + { + type: 'text', + text: `Registration successful!\n\nUsername: ${username}\nEmail: ${email}\nNewsletter: ${newsletter ? 'Yes' : 'No'}` + } + ] + }; + } else if (result.action === 'decline') { + return { + content: [ + { + type: 'text', + text: 'Registration cancelled by user.' + } + ] + }; + } else { + return { + content: [ + { + type: 'text', + text: 'Registration was cancelled.' + } + ] + }; } - }); - - if (result.action === 'accept' && result.content) { + } catch (error) { return { content: [ { type: 'text', - text: `Address updated successfully!\n\n${JSON.stringify(result.content, null, 2)}` + text: `Registration failed: ${error instanceof Error ? error.message : String(error)}` + } + ], + isError: true + }; + } + } + ); + + /** + * Example 2: Multi-step workflow with multiple form elicitation requests + * Demonstrates how to collect information in multiple steps + */ + mcpServer.registerTool( + 'create_event', + { + description: 'Create a calendar event by collecting event details', + inputSchema: {} + }, + async () => { + try { + // Step 1: Collect basic event information + const basicInfo = await mcpServer.server.elicitInput({ + mode: 'form', + message: 'Step 1: Enter basic event information', + requestedSchema: { + type: 'object', + properties: { + title: { + type: 'string', + title: 'Event Title', + description: 'Name of the event', + minLength: 1 + }, + description: { + type: 'string', + title: 'Description', + description: 'Event description (optional)' + } + }, + required: ['title'] + } + }); + + if (basicInfo.action !== 'accept' || !basicInfo.content) { + return { + content: [{ type: 'text', text: 'Event creation cancelled.' }] + }; + } + + // Step 2: Collect date and time + const dateTime = await mcpServer.server.elicitInput({ + mode: 'form', + message: 'Step 2: Enter date and time', + requestedSchema: { + type: 'object', + properties: { + date: { + type: 'string', + title: 'Date', + description: 'Event date', + format: 'date' + }, + startTime: { + type: 'string', + title: 'Start Time', + description: 'Event start time (HH:MM)' + }, + duration: { + type: 'integer', + title: 'Duration', + description: 'Duration in minutes', + minimum: 15, + maximum: 480 + } + }, + required: ['date', 'startTime', 'duration'] + } + }); + + if (dateTime.action !== 'accept' || !dateTime.content) { + return { + content: [{ type: 'text', text: 'Event creation cancelled.' }] + }; + } + + // Combine all collected information + const event = { + ...basicInfo.content, + ...dateTime.content + }; + + return { + content: [ + { + type: 'text', + text: `Event created successfully!\n\n${JSON.stringify(event, null, 2)}` } ] }; - } else if (result.action === 'decline') { + } catch (error) { return { - content: [{ type: 'text', text: 'Address update cancelled by user.' }] - }; - } else { - return { - content: [{ type: 'text', text: 'Address update was cancelled.' }] + content: [ + { + type: 'text', + text: `Event creation failed: ${error instanceof Error ? error.message : String(error)}` + } + ], + isError: true }; } - } catch (error) { - return { - content: [ - { - type: 'text', - text: `Address update failed: ${error instanceof Error ? error.message : String(error)}` - } - ], - isError: true - }; } - } -); + ); + + /** + * Example 3: Collecting address information + * Demonstrates validation with patterns and optional fields + */ + mcpServer.registerTool( + 'update_shipping_address', + { + description: 'Update shipping address with validation', + inputSchema: {} + }, + async () => { + try { + const result = await mcpServer.server.elicitInput({ + mode: 'form', + message: 'Please provide your shipping address:', + requestedSchema: { + type: 'object', + properties: { + name: { + type: 'string', + title: 'Full Name', + description: 'Recipient name', + minLength: 1 + }, + street: { + type: 'string', + title: 'Street Address', + minLength: 1 + }, + city: { + type: 'string', + title: 'City', + minLength: 1 + }, + state: { + type: 'string', + title: 'State/Province', + minLength: 2, + maxLength: 2 + }, + zipCode: { + type: 'string', + title: 'ZIP/Postal Code', + description: '5-digit ZIP code' + }, + phone: { + type: 'string', + title: 'Phone Number (optional)', + description: 'Contact phone number' + } + }, + required: ['name', 'street', 'city', 'state', 'zipCode'] + } + }); + + if (result.action === 'accept' && result.content) { + return { + content: [ + { + type: 'text', + text: `Address updated successfully!\n\n${JSON.stringify(result.content, null, 2)}` + } + ] + }; + } else if (result.action === 'decline') { + return { + content: [{ type: 'text', text: 'Address update cancelled by user.' }] + }; + } else { + return { + content: [{ type: 'text', text: 'Address update was cancelled.' }] + }; + } + } catch (error) { + return { + content: [ + { + type: 'text', + text: `Address update failed: ${error instanceof Error ? error.message : String(error)}` + } + ], + isError: true + }; + } + } + ); + + return mcpServer; +}; async function main() { const PORT = process.env.PORT ? parseInt(process.env.PORT, 10) : 3000; @@ -357,7 +363,8 @@ async function main() { } }; - // Connect the transport to the MCP server BEFORE handling the request + // Create a new server per session and connect it to the transport + const mcpServer = getServer(); await mcpServer.connect(transport); await transport.handleRequest(req, res, req.body); diff --git a/src/examples/server/honoWebStandardStreamableHttp.ts b/src/examples/server/honoWebStandardStreamableHttp.ts index ba8805ea..5d4510f4 100644 --- a/src/examples/server/honoWebStandardStreamableHttp.ts +++ b/src/examples/server/honoWebStandardStreamableHttp.ts @@ -15,29 +15,30 @@ import { McpServer } from '../../server/mcp.js'; import { WebStandardStreamableHTTPServerTransport } from '../../server/webStandardStreamableHttp.js'; import { CallToolResult } from '../../types.js'; -// Create the MCP server -const server = new McpServer({ - name: 'hono-webstandard-mcp-server', - version: '1.0.0' -}); +// Factory function to create a new MCP server per request (stateless mode) +const getServer = () => { + const server = new McpServer({ + name: 'hono-webstandard-mcp-server', + version: '1.0.0' + }); -// Register a simple greeting tool -server.registerTool( - 'greet', - { - title: 'Greeting Tool', - description: 'A simple greeting tool', - inputSchema: { name: z.string().describe('Name to greet') } - }, - async ({ name }): Promise => { - return { - content: [{ type: 'text', text: `Hello, ${name}! (from Hono + WebStandard transport)` }] - }; - } -); + // Register a simple greeting tool + server.registerTool( + 'greet', + { + title: 'Greeting Tool', + description: 'A simple greeting tool', + inputSchema: { name: z.string().describe('Name to greet') } + }, + async ({ name }): Promise => { + return { + content: [{ type: 'text', text: `Hello, ${name}! (from Hono + WebStandard transport)` }] + }; + } + ); -// Create a stateless transport (no options = no session management) -const transport = new WebStandardStreamableHTTPServerTransport(); + return server; +}; // Create the Hono app const app = new Hono(); @@ -56,19 +57,22 @@ app.use( // Health check endpoint app.get('/health', c => c.json({ status: 'ok' })); -// MCP endpoint -app.all('/mcp', c => transport.handleRequest(c.req.raw)); +// MCP endpoint - create a fresh transport and server per request (stateless) +app.all('/mcp', async (c) => { + const transport = new WebStandardStreamableHTTPServerTransport(); + const server = getServer(); + await server.connect(transport); + return transport.handleRequest(c.req.raw); +}); // Start the server const PORT = process.env.MCP_PORT ? parseInt(process.env.MCP_PORT, 10) : 3000; -server.connect(transport).then(() => { - console.log(`Starting Hono MCP server on port ${PORT}`); - console.log(`Health check: http://localhost:${PORT}/health`); - console.log(`MCP endpoint: http://localhost:${PORT}/mcp`); +console.log(`Starting Hono MCP server on port ${PORT}`); +console.log(`Health check: http://localhost:${PORT}/health`); +console.log(`MCP endpoint: http://localhost:${PORT}/mcp`); - serve({ - fetch: app.fetch, - port: PORT - }); +serve({ + fetch: app.fetch, + port: PORT }); diff --git a/src/examples/server/ssePollingExample.ts b/src/examples/server/ssePollingExample.ts index bbecf2fd..04d9f775 100644 --- a/src/examples/server/ssePollingExample.ts +++ b/src/examples/server/ssePollingExample.ts @@ -12,7 +12,7 @@ * Run with: npx tsx src/examples/server/ssePollingExample.ts * Test with: curl or the MCP Inspector */ -import { Request, Response } from 'express'; +import { type Request, type Response } from 'express'; import { randomUUID } from 'node:crypto'; import { McpServer } from '../../server/mcp.js'; import { createMcpExpressApp } from '../../server/express.js'; @@ -21,87 +21,92 @@ import { CallToolResult } from '../../types.js'; import { InMemoryEventStore } from '../shared/inMemoryEventStore.js'; import cors from 'cors'; -// Create the MCP server -const server = new McpServer( - { - name: 'sse-polling-example', - version: '1.0.0' - }, - { - capabilities: { logging: {} } - } -); - -// Register a long-running tool that demonstrates server-initiated disconnect -server.tool( - 'long-task', - 'A long-running task that sends progress updates. Server will disconnect mid-task to demonstrate polling.', - {}, - async (_args, extra): Promise => { - const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); - - console.log(`[${extra.sessionId}] Starting long-task...`); - - // Send first progress notification - await server.sendLoggingMessage( - { - level: 'info', - data: 'Progress: 25% - Starting work...' - }, - extra.sessionId - ); - await sleep(1000); - - // Send second progress notification - await server.sendLoggingMessage( - { - level: 'info', - data: 'Progress: 50% - Halfway there...' - }, - extra.sessionId - ); - await sleep(1000); - - // Server decides to disconnect the client to free resources - // Client will reconnect via GET with Last-Event-ID after the transport's retryInterval - // Use extra.closeSSEStream callback - available when eventStore is configured - if (extra.closeSSEStream) { - console.log(`[${extra.sessionId}] Closing SSE stream to trigger client polling...`); - extra.closeSSEStream(); +// Factory to create a new MCP server per session. +// Each session needs its own server+transport pair to avoid cross-session contamination. +const getServer = () => { + const server = new McpServer( + { + name: 'sse-polling-example', + version: '1.0.0' + }, + { + capabilities: { logging: {} } } + ); - // Continue processing while client is disconnected - // Events are stored in eventStore and will be replayed on reconnect - await sleep(500); - await server.sendLoggingMessage( - { - level: 'info', - data: 'Progress: 75% - Almost done (sent while client disconnected)...' - }, - extra.sessionId - ); + // Register a long-running tool that demonstrates server-initiated disconnect + server.tool( + 'long-task', + 'A long-running task that sends progress updates. Server will disconnect mid-task to demonstrate polling.', + {}, + async (_args, extra): Promise => { + const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); - await sleep(500); - await server.sendLoggingMessage( - { - level: 'info', - data: 'Progress: 100% - Complete!' - }, - extra.sessionId - ); + console.log(`[${extra.sessionId}] Starting long-task...`); - console.log(`[${extra.sessionId}] Task complete`); - - return { - content: [ + // Send first progress notification + await server.sendLoggingMessage( { - type: 'text', - text: 'Long task completed successfully!' - } - ] - }; - } -); + level: 'info', + data: 'Progress: 25% - Starting work...' + }, + extra.sessionId + ); + await sleep(1000); + + // Send second progress notification + await server.sendLoggingMessage( + { + level: 'info', + data: 'Progress: 50% - Halfway there...' + }, + extra.sessionId + ); + await sleep(1000); + + // Server decides to disconnect the client to free resources + // Client will reconnect via GET with Last-Event-ID after the transport's retryInterval + // Use extra.closeSSEStream callback - available when eventStore is configured + if (extra.closeSSEStream) { + console.log(`[${extra.sessionId}] Closing SSE stream to trigger client polling...`); + extra.closeSSEStream(); + } + + // Continue processing while client is disconnected + // Events are stored in eventStore and will be replayed on reconnect + await sleep(500); + await server.sendLoggingMessage( + { + level: 'info', + data: 'Progress: 75% - Almost done (sent while client disconnected)...' + }, + extra.sessionId + ); + + await sleep(500); + await server.sendLoggingMessage( + { + level: 'info', + data: 'Progress: 100% - Complete!' + }, + extra.sessionId + ); + + console.log(`[${extra.sessionId}] Task complete`); + + return { + content: [ + { + type: 'text', + text: 'Long task completed successfully!' + } + ] + }; + } + ); + + return server; +}; // Set up Express app const app = createMcpExpressApp(); @@ -131,7 +136,8 @@ app.all('/mcp', async (req: Request, res: Response) => { } }); - // Connect the MCP server to the transport + // Create a new server per session and connect it to the transport + const server = getServer(); await server.connect(transport); } diff --git a/src/examples/server/standaloneSseWithGetStreamableHttp.ts b/src/examples/server/standaloneSseWithGetStreamableHttp.ts index 225ef1f3..97882874 100644 --- a/src/examples/server/standaloneSseWithGetStreamableHttp.ts +++ b/src/examples/server/standaloneSseWithGetStreamableHttp.ts @@ -5,36 +5,47 @@ import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js'; import { isInitializeRequest, ReadResourceResult } from '../../types.js'; import { createMcpExpressApp } from '../../server/express.js'; -// Create an MCP server with implementation details -const server = new McpServer({ - name: 'resource-list-changed-notification-server', - version: '1.0.0' -}); +// Factory to create a new MCP server per session. +// Each session needs its own server+transport pair to avoid cross-session contamination. +const getServer = () => { + const server = new McpServer({ + name: 'resource-list-changed-notification-server', + version: '1.0.0' + }); + + const addResource = (name: string, content: string) => { + const uri = `https://mcp-example.com/dynamic/${encodeURIComponent(name)}`; + server.registerResource( + name, + uri, + { mimeType: 'text/plain', description: `Dynamic resource: ${name}` }, + async (): Promise => { + return { + contents: [{ uri, text: content }] + }; + } + ); + }; + + addResource('example-resource', 'Initial content for example-resource'); + + // Periodically add new resources to demonstrate notifications + const resourceChangeInterval = setInterval(() => { + const name = randomUUID(); + addResource(name, `Content for ${name}`); + }, 5000); + + // Clean up the interval when the server closes + server.server.onclose = () => { + clearInterval(resourceChangeInterval); + }; + + return server; +}; // Store transports by session ID to send notifications const transports: { [sessionId: string]: StreamableHTTPServerTransport } = {}; -const addResource = (name: string, content: string) => { - const uri = `https://mcp-example.com/dynamic/${encodeURIComponent(name)}`; - server.registerResource( - name, - uri, - { mimeType: 'text/plain', description: `Dynamic resource: ${name}` }, - async (): Promise => { - return { - contents: [{ uri, text: content }] - }; - } - ); -}; - -addResource('example-resource', 'Initial content for example-resource'); - -const resourceChangeInterval = setInterval(() => { - const name = randomUUID(); - addResource(name, `Content for ${name}`); -}, 5000); // Change resources every 5 seconds for testing - const app = createMcpExpressApp(); app.post('/mcp', async (req: Request, res: Response) => { @@ -59,7 +70,8 @@ app.post('/mcp', async (req: Request, res: Response) => { } }); - // Connect the transport to the MCP server + // Create a new server per session and connect it to the transport + const server = getServer(); await server.connect(transport); // Handle the request - the onsessioninitialized callback will store the transport @@ -121,7 +133,9 @@ app.listen(PORT, error => { // Handle server shutdown process.on('SIGINT', async () => { console.log('Shutting down server...'); - clearInterval(resourceChangeInterval); - await server.close(); + for (const sessionId in transports) { + await transports[sessionId].close(); + delete transports[sessionId]; + } process.exit(0); }); diff --git a/src/server/webStandardStreamableHttp.ts b/src/server/webStandardStreamableHttp.ts index 3ae9846c..c811c54c 100644 --- a/src/server/webStandardStreamableHttp.ts +++ b/src/server/webStandardStreamableHttp.ts @@ -210,6 +210,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // when sessionId is not set (undefined), it means the transport is in stateless mode private sessionIdGenerator: (() => string) | undefined; private _started: boolean = false; + private _hasHandledRequest: boolean = false; private _streamMapping: Map = new Map(); private _requestToStreamMapping: Map = new Map(); private _requestResponseMap: Map = new Map(); @@ -319,6 +320,13 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { * Returns a Response object (Web Standard) */ async handleRequest(req: Request, options?: HandleRequestOptions): Promise { + // In stateless mode (no sessionIdGenerator), each request must use a fresh transport. + // Reusing a stateless transport causes message ID collisions between clients. + if (!this.sessionIdGenerator && this._hasHandledRequest) { + throw new Error('Stateless transport cannot be reused across requests. Create a new transport per request.'); + } + this._hasHandledRequest = true; + // Validate request headers for DNS rebinding protection const validationError = this.validateRequestHeaders(req); if (validationError) { diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index aa242a64..5bb6d62e 100644 --- a/src/shared/protocol.ts +++ b/src/shared/protocol.ts @@ -605,6 +605,12 @@ export abstract class Protocol { + if (this._transport) { + throw new Error( + 'Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection.' + ); + } + this._transport = transport; const _onclose = this.transport?.onclose; this._transport.onclose = () => { @@ -642,6 +648,12 @@ export abstract class Protocol { + if (abortController.signal.aborted) return; // Include related-task metadata if this request is part of a task const notificationOptions: NotificationOptions = { relatedRequestId: request.id }; if (relatedTaskId) { @@ -727,6 +740,9 @@ export abstract class Protocol { + if (abortController.signal.aborted) { + throw new McpError(ErrorCode.ConnectionClosed, 'Request was cancelled'); + } // Include related-task metadata if this request is part of a task const requestOptions: RequestOptions = { ...options, relatedRequestId: request.id }; if (relatedTaskId && !requestOptions.relatedTask) { diff --git a/test/integration-tests/stateManagementStreamableHttp.test.ts b/test/integration-tests/stateManagementStreamableHttp.test.ts index d79d95c7..311aa23c 100644 --- a/test/integration-tests/stateManagementStreamableHttp.test.ts +++ b/test/integration-tests/stateManagementStreamableHttp.test.ts @@ -17,9 +17,11 @@ import { listenOnRandomPort } from '../helpers/http.js'; describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => { const { z } = entry; describe('Streamable HTTP Transport Session Management', () => { - // Function to set up the server with optional session management - async function setupServer(withSessionManagement: boolean) { - const server: Server = createServer(); + /** + * Helper to create and configure a fresh McpServer instance with standard + * resources, prompts, and tools for testing. + */ + function createMcpServer(): McpServer { const mcpServer = new McpServer( { name: 'test-server', version: '1.0.0' }, { @@ -67,43 +69,67 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => { } ); - // Create transport with or without session management - const serverTransport = new StreamableHTTPServerTransport({ - sessionIdGenerator: withSessionManagement - ? () => randomUUID() // With session management, generate UUID - : undefined // Without session management, return undefined - }); + return mcpServer; + } - await mcpServer.connect(serverTransport); + // Function to set up the server with optional session management + async function setupServer(withSessionManagement: boolean): Promise<{ + server: Server; + mcpServer?: McpServer; + serverTransport?: StreamableHTTPServerTransport; + baseUrl: URL; + }> { + const server: Server = createServer(); - server.on('request', async (req, res) => { - await serverTransport.handleRequest(req, res); - }); + if (withSessionManagement) { + // Stateful mode: single transport + server for the session + const mcpServer = createMcpServer(); + const serverTransport = new StreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID() + }); - // Start the server on a random port - const baseUrl = await listenOnRandomPort(server); + await mcpServer.connect(serverTransport); - return { server, mcpServer, serverTransport, baseUrl }; + server.on('request', async (req, res) => { + await serverTransport.handleRequest(req, res); + }); + + // Start the server on a random port + const baseUrl = await listenOnRandomPort(server); + + return { server, mcpServer, serverTransport, baseUrl }; + } else { + // Stateless mode: create a fresh transport + server per request + // to comply with the guard that stateless transports cannot be reused. + server.on('request', async (req, res) => { + const mcpServer = createMcpServer(); + const serverTransport = new StreamableHTTPServerTransport({ + sessionIdGenerator: undefined + }); + await mcpServer.connect(serverTransport); + await serverTransport.handleRequest(req, res); + // Close the per-request mcpServer after handling to avoid leaks + await mcpServer.close(); + }); + + // Start the server on a random port + const baseUrl = await listenOnRandomPort(server); + + return { server, baseUrl }; + } } describe('Stateless Mode', () => { let server: Server; - let mcpServer: McpServer; - let serverTransport: StreamableHTTPServerTransport; let baseUrl: URL; beforeEach(async () => { const setup = await setupServer(false); server = setup.server; - mcpServer = setup.mcpServer; - serverTransport = setup.serverTransport; baseUrl = setup.baseUrl; }); afterEach(async () => { - // Clean up resources - await mcpServer.close().catch(() => {}); - await serverTransport.close().catch(() => {}); server.close(); }); diff --git a/test/server/streamableHttp.test.ts b/test/server/streamableHttp.test.ts index ae77dd4e..1c4d5ed8 100644 --- a/test/server/streamableHttp.test.ts +++ b/test/server/streamableHttp.test.ts @@ -1529,20 +1529,56 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => { // Test stateless mode describe('StreamableHTTPServerTransport in stateless mode', () => { let server: Server; - let transport: StreamableHTTPServerTransport; let baseUrl: URL; + // In stateless mode, each request must use a fresh transport + server pair. + // The HTTP server creates these per-request and delegates accordingly. beforeEach(async () => { - const result = await createTestServer({ sessionIdGenerator: undefined }); - server = result.server; - transport = result.transport; - baseUrl = result.baseUrl; + server = createServer(async (req, res) => { + try { + const { transport, mcpServer } = await createStatelessHandler(); + await transport.handleRequest(req, res); + // Close the per-request mcpServer after handling to avoid leaks + await mcpServer.close(); + } catch (error) { + console.error('Error handling request:', error); + if (!res.headersSent) res.writeHead(500).end(); + } + }); + baseUrl = await listenOnRandomPort(server); }); afterEach(async () => { - await stopTestServer({ server, transport }); + server.close(); }); + /** + * Creates a fresh transport + mcpServer pair for a single stateless request. + */ + async function createStatelessHandler(): Promise<{ + transport: StreamableHTTPServerTransport; + mcpServer: McpServer; + }> { + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }, { capabilities: { logging: {} } }); + + mcpServer.tool( + 'greet', + 'A simple greeting tool', + { name: z.string().describe('Name to greet') }, + async ({ name }): Promise => { + return { content: [{ type: 'text', text: `Hello, ${name}!` }] }; + } + ); + + const transport = new StreamableHTTPServerTransport({ + sessionIdGenerator: undefined + }); + + await mcpServer.connect(transport); + + return { transport, mcpServer }; + } + it('should operate without session ID validation', async () => { // Initialize the server first const initResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.initialize); @@ -1552,6 +1588,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => { expect(initResponse.headers.get('mcp-session-id')).toBeNull(); // Try request without session ID - should work in stateless mode + // (a fresh transport is created per request) const toolsResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.toolsList); expect(toolsResponse.status).toBe(200); @@ -1585,14 +1622,14 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => { expect(response2.status).toBe(200); }); - it('should reject second SSE stream even in stateless mode', async () => { - // Despite no session ID requirement, the transport still only allows - // one standalone SSE stream at a time + it('should allow multiple SSE streams in stateless mode with per-request transports', async () => { + // Each request gets its own transport, so multiple SSE streams can + // coexist since they are handled by separate transport instances // Initialize the server first await sendPostRequest(baseUrl, TEST_MESSAGES.initialize); - // Open first SSE stream + // Open first SSE stream - this uses its own per-request transport const stream1 = await fetch(baseUrl, { method: 'GET', headers: { @@ -1602,7 +1639,8 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => { }); expect(stream1.status).toBe(200); - // Open second SSE stream - should still be rejected, stateless mode still only allows one + // Open second SSE stream - also gets its own per-request transport, + // so it should also succeed (each transport only handles one request) const stream2 = await fetch(baseUrl, { method: 'GET', headers: { @@ -1610,7 +1648,9 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => { 'mcp-protocol-version': '2025-11-25' } }); - expect(stream2.status).toBe(409); // Conflict - only one stream allowed + // With per-request transports in stateless mode, each GET gets its own + // transport, so the second one also succeeds + expect(stream2.status).toBe(200); }); }); @@ -2868,17 +2908,20 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => { describe('Combined validations', () => { it('should validate both host and origin when both are configured', async () => { - const result = await createTestServerWithDnsProtection({ + // In stateless mode, each request needs a fresh transport, so we + // test invalid and valid origins with separate server instances. + + // Test with invalid origin + const result1 = await createTestServerWithDnsProtection({ sessionIdGenerator: undefined, allowedHosts: ['localhost'], allowedOrigins: ['http://localhost:3001'], enableDnsRebindingProtection: true }); - server = result.server; - transport = result.transport; - baseUrl = result.baseUrl; + server = result1.server; + transport = result1.transport; + baseUrl = result1.baseUrl; - // Test with invalid origin (host will be automatically correct via fetch) const response1 = await fetch(baseUrl, { method: 'POST', headers: { @@ -2893,7 +2936,20 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => { const body1 = await response1.json(); expect(body1.error.message).toBe('Invalid Origin header: http://evil.com'); - // Test with valid origin + // Clean up first server + await stopTestServer({ server, transport }); + + // Test with valid origin using a fresh server+transport + const result2 = await createTestServerWithDnsProtection({ + sessionIdGenerator: undefined, + allowedHosts: ['localhost'], + allowedOrigins: ['http://localhost:3001'], + enableDnsRebindingProtection: true + }); + server = result2.server; + transport = result2.transport; + baseUrl = result2.baseUrl; + const response2 = await fetch(baseUrl, { method: 'POST', headers: { diff --git a/test/shared/protocol-transport-handling.test.ts b/test/shared/protocol-transport-handling.test.ts index 60eff5c2..14863946 100644 --- a/test/shared/protocol-transport-handling.test.ts +++ b/test/shared/protocol-transport-handling.test.ts @@ -27,30 +27,36 @@ class MockTransport implements Transport { } } -describe('Protocol transport handling bug', () => { - let protocol: Protocol; +function createProtocol(): Protocol { + return new (class extends Protocol { + protected assertCapabilityForMethod(): void {} + protected assertNotificationCapability(): void {} + protected assertRequestHandlerCapability(): void {} + protected assertTaskCapability(): void {} + protected assertTaskHandlerCapability(): void {} + })(); +} + +describe('Protocol transport handling', () => { let transportA: MockTransport; let transportB: MockTransport; beforeEach(() => { - protocol = new (class extends Protocol { - protected assertCapabilityForMethod(): void {} - protected assertNotificationCapability(): void {} - protected assertRequestHandlerCapability(): void {} - protected assertTaskCapability(): void {} - protected assertTaskHandlerCapability(): void {} - })(); - transportA = new MockTransport('A'); transportB = new MockTransport('B'); }); - test('should send response to the correct transport when multiple clients are connected', async () => { - // Set up a request handler that simulates processing time - let resolveHandler: (value: Result) => void; - const handlerPromise = new Promise(resolve => { - resolveHandler = resolve; - }); + test('should send response to the correct transport when using separate protocol instances', async () => { + const protocolA = createProtocol(); + const protocolB = createProtocol(); + + // Each protocol gets its own resolver so we can verify responses route correctly + let resolveA: (value: Result) => void; + let resolveB: (value: Result) => void; + let handlerAEnteredResolve: () => void; + let handlerBEnteredResolve: () => void; + const handlerAEntered = new Promise(resolve => { handlerAEnteredResolve = resolve; }); + const handlerBEntered = new Promise(resolve => { handlerBEnteredResolve = resolve; }); const TestRequestSchema = z.object({ method: z.literal('test/method'), @@ -61,13 +67,22 @@ describe('Protocol transport handling bug', () => { .optional() }); - protocol.setRequestHandler(TestRequestSchema, async request => { - console.log(`Processing request from ${request.params?.from}`); - return handlerPromise; + protocolA.setRequestHandler(TestRequestSchema, async () => { + return new Promise(resolve => { + resolveA = resolve; + handlerAEnteredResolve(); + }); + }); + + protocolB.setRequestHandler(TestRequestSchema, async () => { + return new Promise(resolve => { + resolveB = resolve; + handlerBEnteredResolve(); + }); }); // Client A connects and sends a request - await protocol.connect(transportA); + await protocolA.connect(transportA); const requestFromA = { jsonrpc: '2.0' as const, @@ -79,9 +94,8 @@ describe('Protocol transport handling bug', () => { // Simulate client A sending a request transportA.onmessage?.(requestFromA); - // While A's request is being processed, client B connects - // This overwrites the transport reference in the protocol - await protocol.connect(transportB); + // Client B connects to a separate protocol instance + await protocolB.connect(transportB); const requestFromB = { jsonrpc: '2.0' as const, @@ -93,19 +107,18 @@ describe('Protocol transport handling bug', () => { // Client B sends its own request transportB.onmessage?.(requestFromB); - // Now complete A's request - resolveHandler!({ data: 'responseForA' } as Result); + // Wait for both handlers to be invoked so resolvers are captured + await handlerAEntered; + await handlerBEntered; - // Wait for async operations to complete + // Resolve each handler with distinct data + resolveA!({ data: 'responseForA' } as Result); + resolveB!({ data: 'responseForB' } as Result); + + // Wait for response delivery (transport.send is async) await new Promise(resolve => setTimeout(resolve, 10)); - // Check where the responses went - console.log('Transport A received:', transportA.sentMessages); - console.log('Transport B received:', transportB.sentMessages); - - // FIXED: Each transport now receives its own response - - // Transport A should receive response for request ID 1 + // Each transport receives its own response expect(transportA.sentMessages.length).toBe(1); expect(transportA.sentMessages[0]).toMatchObject({ jsonrpc: '2.0', @@ -113,18 +126,17 @@ describe('Protocol transport handling bug', () => { result: { data: 'responseForA' } }); - // Transport B should only receive its own response (when implemented) expect(transportB.sentMessages.length).toBe(1); expect(transportB.sentMessages[0]).toMatchObject({ jsonrpc: '2.0', id: 2, - result: { data: 'responseForA' } // Same handler result in this test + result: { data: 'responseForB' } }); }); - test('demonstrates the timing issue with multiple rapid connections', async () => { - const delays: number[] = []; - const results: { transport: string; response: JSONRPCMessage[] }[] = []; + test('demonstrates isolation with separate protocol instances for rapid connections', async () => { + const protocolA = createProtocol(); + const protocolB = createProtocol(); const DelayedRequestSchema = z.object({ method: z.literal('test/delayed'), @@ -136,21 +148,20 @@ describe('Protocol transport handling bug', () => { .optional() }); - // Set up handler with variable delay - protocol.setRequestHandler(DelayedRequestSchema, async (request, extra) => { - const delay = request.params?.delay || 0; - delays.push(delay); + // Set up handler with variable delay on each protocol + for (const protocol of [protocolA, protocolB]) { + protocol.setRequestHandler(DelayedRequestSchema, async (request, extra) => { + const delay = request.params?.delay || 0; + await new Promise(resolve => setTimeout(resolve, delay)); + return { + processedBy: `handler-${extra.requestId}`, + delay: delay + } as Result; + }); + } - await new Promise(resolve => setTimeout(resolve, delay)); - - return { - processedBy: `handler-${extra.requestId}`, - delay: delay - } as Result; - }); - - // Rapid succession of connections and requests - await protocol.connect(transportA); + // Connect and send requests + await protocolA.connect(transportA); transportA.onmessage?.({ jsonrpc: '2.0' as const, method: 'test/delayed', @@ -160,7 +171,7 @@ describe('Protocol transport handling bug', () => { // Connect B while A is processing setTimeout(async () => { - await protocol.connect(transportB); + await protocolB.connect(transportB); transportB.onmessage?.({ jsonrpc: '2.0' as const, method: 'test/delayed', @@ -172,18 +183,81 @@ describe('Protocol transport handling bug', () => { // Wait for all processing await new Promise(resolve => setTimeout(resolve, 100)); - // Collect results - if (transportA.sentMessages.length > 0) { - results.push({ transport: 'A', response: transportA.sentMessages }); - } - if (transportB.sentMessages.length > 0) { - results.push({ transport: 'B', response: transportB.sentMessages }); - } - - console.log('Timing test results:', results); - - // FIXED: Each transport receives its own responses + // Each transport receives its own responses expect(transportA.sentMessages.length).toBe(1); expect(transportB.sentMessages.length).toBe(1); }); + + test('connect guard throws when calling connect() twice without closing', async () => { + const protocol = createProtocol(); + + await protocol.connect(transportA); + + await expect(protocol.connect(transportB)).rejects.toThrow( + 'Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection.' + ); + }); + + test('connect succeeds after calling close() first', async () => { + const protocol = createProtocol(); + + await protocol.connect(transportA); + await protocol.close(); + + // Should succeed without error + await expect(protocol.connect(transportB)).resolves.toBeUndefined(); + }); + + test('close() aborts in-flight request handlers', async () => { + const protocol = createProtocol(); + + const SlowRequestSchema = z.object({ + method: z.literal('test/slow') + }); + + let capturedSignal: AbortSignal | undefined; + let capturedSendNotification: ((notification: Notification) => Promise) | undefined; + let resolveHandler: () => void; + const handlerBlocking = new Promise(resolve => { + resolveHandler = resolve; + }); + + protocol.setRequestHandler(SlowRequestSchema, async (_request, extra) => { + capturedSignal = extra.signal; + capturedSendNotification = extra.sendNotification; + // Block the handler until we release it + await handlerBlocking; + return {} as Result; + }); + + await protocol.connect(transportA); + + // Send a request to trigger the handler + transportA.onmessage?.({ + jsonrpc: '2.0' as const, + method: 'test/slow', + id: 1 + }); + + // Wait for the handler to start and capture the signal + await new Promise(resolve => setTimeout(resolve, 10)); + expect(capturedSignal).toBeDefined(); + expect(capturedSignal!.aborted).toBe(false); + + // Close the protocol while the handler is still in-flight + await protocol.close(); + + // The signal should now be aborted + expect(capturedSignal!.aborted).toBe(true); + + // sendNotification should be a no-op after close (no error thrown) + await expect(capturedSendNotification!({ method: 'notifications/test' } as Notification)).resolves.toBeUndefined(); + + // No notification should have been sent to the transport + const notifications = transportA.sentMessages.filter((m: JSONRPCMessage) => 'method' in m && m.method === 'notifications/test'); + expect(notifications).toHaveLength(0); + + // Release the handler so the promise chain completes + resolveHandler!(); + }); });