From a05be176cabeae1f933b676e3ce024bf02e2314d Mon Sep 17 00:00:00 2001 From: Paul Carleton Date: Wed, 4 Feb 2026 18:23:03 +0000 Subject: [PATCH] Merge commit from fork * fix: add transport isolation guards to prevent cross-client data leaks When a single McpServer or stateless transport is reused across multiple client connections, responses can be routed to the wrong client due to message ID collisions. This is a data leak vulnerability (CWE-362). Two guards added: - Protocol.connect() throws if already connected to a transport - Stateless transport.handleRequest() throws if called more than once Also fixes three examples that shared a single McpServer across sessions: - standaloneSseWithGetStreamableHttp.ts - ssePollingExample.ts - elicitationFormExample.ts Related: #820, #204, #243 Co-Authored-By: Claude Opus 4.5 * fix: correct misleading test name and add per-request cleanup - Rename 'should reject second SSE stream even in stateless mode' to 'should allow multiple SSE streams in stateless mode with per-request transports' since the test now asserts both streams succeed (status 200) - Add mcpServer.close() after per-request handling in stateManagementStreamableHttp.test.ts to prevent resource leaks, matching the pattern used in streamableHttp.test.ts - Remove unused mcpServers array that collected but never cleaned up per-request server instances * thread abort through sendRequest and notification to prevent crosstalk * test: remove dummy objects from stateless setupServer return In stateless mode, setupServer() was creating unused McpServer and StreamableHTTPServerTransport instances solely to satisfy the return type. Make mcpServer/serverTransport optional in the return type and simplify the stateless beforeEach/afterEach to only track server and baseUrl. * fix: use McpError for sendRequest abort guard and fix Hono example reuse - sendRequest abort guard now throws McpError(ErrorCode.ConnectionClosed) instead of plain Error for consistency with the rest of the codebase - Hono example updated to create fresh transport and server per request, fixing breakage from the stateless transport reuse guard * fix: use separate resolvers per protocol in transport isolation test Address review feedback: each protocol handler now has its own resolver returning distinct data (responseForA vs responseForB), so the test properly demonstrates that responses route to the correct transport. Uses explicit 'handler entered' promises for deterministic synchronization. --------- Co-authored-by: Claude Opus 4.5 --- src/examples/server/elicitationFormExample.ts | 579 +++++++++--------- .../server/honoWebStandardStreamableHttp.ts | 66 +- src/examples/server/ssePollingExample.ts | 162 ++--- .../standaloneSseWithGetStreamableHttp.ts | 72 ++- src/server/webStandardStreamableHttp.ts | 8 + src/shared/protocol.ts | 16 + .../stateManagementStreamableHttp.test.ts | 72 ++- test/server/streamableHttp.test.ts | 92 ++- .../protocol-transport-handling.test.ts | 202 ++++-- 9 files changed, 740 insertions(+), 529 deletions(-) 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!(); + }); });