diff --git a/src/index.ts b/src/index.ts index 275ab30..df7a0f2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -14,8 +14,8 @@ import { initializeObservability, shutdownObservability } from './observability/ import { logWithContext } from './observability/instrumentation.js'; import { logger } from './observability/logger.js'; import { registerHandlers } from './server/handlers.js'; -import { setupServer, type ServerContext } from './server/setup.js'; -import { createHttpTransport, getHttpTransportConfig, startHttpServer } from './transport/http.js'; +import { createServer, setupServer, type ServerContext } from './server/setup.js'; +import { getHttpTransportConfig, startHttpServer } from './transport/http.js'; import { startStdioTransport } from './transport/stdio.js'; import { getVersionInfo, getVersionString } from './version.js'; @@ -79,8 +79,13 @@ async function main() { dbSizeAnalyzer: serverContext.dbSizeAnalyzer, }; - const transport = createHttpTransport(config); - httpServerInstance = await startHttpServer(serverContext.server, transport, config); + // A Server binds to one transport, so HTTP needs a fresh one per + // session (stateful) / per request (stateless). + httpServerInstance = await startHttpServer(() => { + const server = createServer(); + registerHandlers({ ...serverContext, server }); + return server; + }, config); // Log REST API availability const restBasePath = process.env.REST_API_BASE_PATH || '/api/v1'; diff --git a/src/transport/http.ts b/src/transport/http.ts index bbe8cf2..bf6e61e 100644 --- a/src/transport/http.ts +++ b/src/transport/http.ts @@ -34,9 +34,17 @@ export interface HttpTransportConfig { } /** - * Create and configure HTTP transport + * Create and configure HTTP transport. + * + * One transport instance serves ONE session (stateful) or ONE request + * (stateless) - the SDK throws if a stateless transport is reused, and + * stateful sessions would collide between clients. Callers must create one + * per session/request; see startHttpServer. */ -export function createHttpTransport(config: HttpTransportConfig = {}) { +export function createHttpTransport( + config: HttpTransportConfig = {}, + onSessionInitialized?: (sessionId: string) => void +) { const sessionIdGenerator = config.stateless ? undefined : () => randomUUID(); @@ -49,6 +57,7 @@ export function createHttpTransport(config: HttpTransportConfig = {}) { enableDnsRebindingProtection: config.dnsProtection || false, onsessioninitialized: (sessionId) => { logWithContext('info', `HTTP session initialized: ${sessionId}`); + onSessionInitialized?.(sessionId); }, onsessionclosed: (sessionId) => { logWithContext('info', `HTTP session closed: ${sessionId}`); @@ -58,16 +67,52 @@ export function createHttpTransport(config: HttpTransportConfig = {}) { /** * Start HTTP server with MCP transport and Koa REST API + * + * @param createMcpServer - Factory returning a fresh MCP Server with handlers + * registered. A Server binds to exactly one transport, so we need one per + * session (stateful) / per request (stateless). */ export async function startHttpServer( - server: Server, - transport: StreamableHTTPServerTransport, + createMcpServer: () => Server, config: HttpTransportConfig = {} ): Promise { - const port = config.port || parseInt(process.env.MCP_HTTP_PORT || '3000'); + // ?? not ||: port 0 is valid (bind any free port). + const port = config.port ?? parseInt(process.env.MCP_HTTP_PORT || '3000'); const host = config.host || process.env.MCP_HTTP_HOST || '127.0.0.1'; - await server.connect(transport); + // Stateful mode: live sessions keyed by Mcp-Session-Id. Empty in stateless + // mode, where every request gets a throwaway transport. + // ponytail: in-memory map, so stateful mode needs sticky routing across + // replicas. Move to a shared store only if that becomes a real deployment. + const sessions = new Map(); + + async function handleMcpRequest( + req: http.IncomingMessage, + res: http.ServerResponse + ): Promise { + const sessionId = req.headers['mcp-session-id'] as string | undefined; + const existing = sessionId ? sessions.get(sessionId) : undefined; + + if (existing) { + return setupHttpMiddleware(req, res, existing); + } + + const transport = createHttpTransport(config, (id) => { + sessions.set(id, transport); + }); + transport.onclose = () => { + if (transport.sessionId) sessions.delete(transport.sessionId); + }; + + await createMcpServer().connect(transport); + + if (config.stateless) { + // Single-use transport: tear it down once the response is done. + res.on('close', () => void transport.close()); + } + + return setupHttpMiddleware(req, res, transport); + } // Create Koa REST API if context provided let koaApp: Koa | null = null; @@ -115,7 +160,24 @@ export async function startHttpServer( } // Fall through to MCP transport - setupHttpMiddleware(req, res, transport); + handleMcpRequest(req, res).catch((error) => { + logWithContext('error', 'MCP request handling failed', { + error_message: error instanceof Error ? error.message : String(error), + error_stack: error instanceof Error ? error.stack : undefined, + }); + if (!res.headersSent) { + res.writeHead(500, { 'Content-Type': 'application/json' }); + res.end( + JSON.stringify({ + jsonrpc: '2.0', + error: { code: -32603, message: 'Internal server error' }, + id: null, + }) + ); + } else { + res.end(); + } + }); }); // Setup server event handlers diff --git a/src/transport/middleware.ts b/src/transport/middleware.ts index 2dc6221..ec7ff04 100644 --- a/src/transport/middleware.ts +++ b/src/transport/middleware.ts @@ -16,7 +16,7 @@ export function setupHttpMiddleware( req: IncomingMessage, res: ServerResponse, transport: any -): void { +): Promise { const requestId = randomUUID(); const startTime = Date.now(); const remoteAddress = req.socket.remoteAddress || 'unknown'; @@ -124,7 +124,7 @@ export function setupHttpMiddleware( return originalEnd.call(this, chunk, encoding, cb); }; - // Delegate to transport - transport.handleRequest(req, res); + // Delegate to transport (caller handles rejection - see startHttpServer) + return transport.handleRequest(req, res); } diff --git a/tests/transport/http-lifecycle.test.ts b/tests/transport/http-lifecycle.test.ts new file mode 100644 index 0000000..4093591 --- /dev/null +++ b/tests/transport/http-lifecycle.test.ts @@ -0,0 +1,88 @@ +/** + * Regression tests for MCP-over-HTTP transport lifecycle. + * + * Guards the bug where startHttpServer reused ONE StreamableHTTPServerTransport + * for every request: the first `initialize` worked, every later request got an + * empty-body HTTP 500 (the SDK throws "Stateless transport cannot be reused + * across requests"). Stateful mode had the same shape of bug - a second client + * hit "Server already initialized". + */ + +import http from 'http'; +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import { Client } from '@modelcontextprotocol/sdk/client/index.js'; +import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'; +import { Server } from '@modelcontextprotocol/sdk/server/index.js'; +import { ListToolsRequestSchema } from '@modelcontextprotocol/sdk/types.js'; +import { startHttpServer, type HttpTransportConfig } from '../../src/transport/http.js'; + +/** Minimal MCP server - no DB, just enough to answer tools/list. */ +function createTestMcpServer(): Server { + const server = new Server( + { name: 'test-server', version: '0.0.0' }, + { capabilities: { tools: {} } } + ); + server.setRequestHandler(ListToolsRequestSchema, async () => ({ + tools: [{ name: 'ping', description: 'ping', inputSchema: { type: 'object' } }], + })); + return server; +} + +async function start(config: HttpTransportConfig): Promise<{ server: http.Server; url: URL }> { + const server = await startHttpServer(createTestMcpServer, { host: '127.0.0.1', port: 0, ...config }); + const { port } = server.address() as { port: number }; + return { server, url: new URL(`http://127.0.0.1:${port}/`) }; +} + +async function connectClient(url: URL): Promise { + const client = new Client({ name: 'test-client', version: '0.0.0' }); + await client.connect(new StreamableHTTPClientTransport(new URL(url))); + return client; +} + +// Auth and REST are configured from env; keep both out of the way. +beforeAll(() => { + process.env.API_AUTH_REQUIRED = 'false'; + process.env.REST_API_ENABLED = 'false'; +}); + +for (const stateless of [true, false]) { + for (const jsonOnly of [true, false]) { + describe(`HTTP transport lifecycle (stateless=${stateless}, jsonOnly=${jsonOnly})`, () => { + let httpServer: http.Server; + let url: URL; + + beforeAll(async () => { + ({ server: httpServer, url } = await start({ port: 0, stateless, jsonOnly })); + }); + + afterAll(() => httpServer.close()); + + it('accepts two full handshakes in sequence', async () => { + const first = await connectClient(url); + expect(first.getServerVersion()?.name).toBe('test-server'); + await first.close(); + + // This is the regression: the second initialize used to 500. + const second = await connectClient(url); + expect(second.getServerVersion()?.name).toBe('test-server'); + await second.close(); + }); + + it('serves tools/list as a separate request after initialize', async () => { + const client = await connectClient(url); + const { tools } = await client.listTools(); + expect(tools.map((t) => t.name)).toEqual(['ping']); + await client.close(); + }); + + it('serves two concurrent clients', async () => { + const [a, b] = await Promise.all([connectClient(url), connectClient(url)]); + const [ta, tb] = await Promise.all([a.listTools(), b.listTools()]); + expect(ta.tools).toHaveLength(1); + expect(tb.tools).toHaveLength(1); + await Promise.all([a.close(), b.close()]); + }); + }); + } +}