diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8e1123d..2792f1d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -21,6 +21,9 @@ jobs: - name: Apply patches run: npx patch-package + - name: Test + run: bun run test + - name: Build run: bun run build diff --git a/package.json b/package.json index dbb701d..6ea0447 100644 --- a/package.json +++ b/package.json @@ -4,6 +4,7 @@ "main": "index.js", "scripts": { "build": "tsc && mcp-build", + "test": "node --test tests/*.test.mjs", "watch": "tsc --watch", "start": "node dist/index.js", "mcp-inspect": "mcp-inspector --transport http --server-url http://localhost:1234", diff --git a/patches/mcp-framework+0.2.18.patch b/patches/mcp-framework+0.2.18.patch index 8f6d439..b591e90 100644 --- a/patches/mcp-framework+0.2.18.patch +++ b/patches/mcp-framework+0.2.18.patch @@ -1,8 +1,7 @@ diff --git a/node_modules/mcp-framework/dist/transports/http/server.js b/node_modules/mcp-framework/dist/transports/http/server.js -index b535515..ad7f146 100644 --- a/node_modules/mcp-framework/dist/transports/http/server.js +++ b/node_modules/mcp-framework/dist/transports/http/server.js -@@ -17,6 +17,8 @@ export class HttpStreamTransport extends AbstractTransport { +@@ -17,10 +17,12 @@ export class HttpStreamTransport extends AbstractTransport { _config; _oauthMetadata; _transports = {}; @@ -11,9 +10,17 @@ index b535515..ad7f146 100644 constructor(config = {}) { super(); this._config = config; -@@ -90,6 +92,23 @@ export class HttpStreamTransport extends AbstractTransport { +- this._port = config.port || 8080; ++ this._port = config.port ?? 8080; + this._endpoint = config.endpoint || '/mcp'; + this._enableJsonResponse = config.responseMode === 'batch'; + // Initialize OAuth metadata if OAuth provider is configured +@@ -88,12 +90,33 @@ export class HttpStreamTransport extends AbstractTransport { + this._onclose?.(); + }); this._server.listen(this._port, () => { - logger.info(`HTTP server listening on port ${this._port}, endpoint ${this._endpoint}`); +- logger.info(`HTTP server listening on port ${this._port}, endpoint ${this._endpoint}`); ++ logger.info(`HTTP server listening on port ${this.port}, endpoint ${this._endpoint}`); this._isRunning = true; + // Start periodic session cleanup + const timeoutMs = this._config.session?.sessionTimeout || 300000; @@ -35,7 +42,26 @@ index b535515..ad7f146 100644 resolve(); }); }); -@@ -117,6 +136,7 @@ export class HttpStreamTransport extends AbstractTransport { + } ++ get port() { ++ const address = this._server?.address(); ++ return address && typeof address === 'object' ? address.port : this._port; ++ } + async handleMcpRequest(req, res) { + const sessionId = req.headers['mcp-session-id']; + let transport; +@@ -109,6 +132,10 @@ export class HttpStreamTransport extends AbstractTransport { + return; + authData = authResult.data || {}; + } ++ if (this._config.session?.enabled === false) { ++ await this.handleStatelessRequest(req, res, body, authData); ++ return; ++ } + // Allow re-initialization even when a stale session ID is provided. + // Clients like Cline may keep sending the old session ID header after + // a session is lost (server restart, transport error, etc.). +@@ -117,6 +144,7 @@ export class HttpStreamTransport extends AbstractTransport { if (sessionId && this._transports[sessionId]) { // Existing session transport = this._transports[sessionId]; @@ -43,7 +69,7 @@ index b535515..ad7f146 100644 logger.debug(`Reusing existing session: ${sessionId}`); } else if (isInitialize || isReInitialize) { -@@ -131,6 +151,7 @@ export class HttpStreamTransport extends AbstractTransport { +@@ -131,6 +159,7 @@ export class HttpStreamTransport extends AbstractTransport { onsessioninitialized: (sessionId) => { logger.info(`Session initialized: ${sessionId}`); this._transports[sessionId] = transport; @@ -51,7 +77,7 @@ index b535515..ad7f146 100644 }, enableJsonResponse: this._enableJsonResponse, }); -@@ -138,6 +159,7 @@ export class HttpStreamTransport extends AbstractTransport { +@@ -138,6 +167,7 @@ export class HttpStreamTransport extends AbstractTransport { if (transport.sessionId) { logger.info(`Transport closed for session: ${transport.sessionId}`); delete this._transports[transport.sessionId]; @@ -59,8 +85,58 @@ index b535515..ad7f146 100644 } }; transport.onerror = (error) => { -@@ -231,16 +253,16 @@ export class HttpStreamTransport extends AbstractTransport { - await transport.send(message); +@@ -172,6 +202,23 @@ export class HttpStreamTransport extends AbstractTransport { + await transport.handleRequest(req, res, body); + }); + } ++ async handleStatelessRequest(req, res, body, authData) { ++ const transport = new StreamableHTTPServerTransport({ ++ sessionIdGenerator: undefined, ++ enableJsonResponse: this._enableJsonResponse, ++ }); ++ transport.onerror = (error) => { ++ logger.error(`Stateless transport error: ${error}`); ++ }; ++ transport.onmessage = async (message) => { ++ if (this._onmessage) { ++ await this._onmessage(message); ++ } ++ }; ++ await requestContext.run({ ...authData, httpTransport: transport }, async () => { ++ await transport.handleRequest(req, res, body); ++ }); ++ } + async readRequestBody(req) { + return new Promise((resolve, reject) => { + let body = ''; +@@ -214,11 +261,20 @@ export class HttpStreamTransport extends AbstractTransport { + id: null, + })); + } +- async send(message) { ++ async send(message, options) { + if (!this._isRunning) { + logger.warn('Attempted to send message, but HTTP transport is not running'); + return; + } ++ if (this._config.session?.enabled === false) { ++ const transport = requestContext.getStore()?.httpTransport; ++ if (!transport) { ++ logger.warn('No active stateless request to send message to'); ++ return; ++ } ++ await transport.send(message, options); ++ return; ++ } + const activeSessions = Object.entries(this._transports); + if (activeSessions.length === 0) { + logger.warn('No active sessions to send message to'); +@@ -228,19 +284,19 @@ export class HttpStreamTransport extends AbstractTransport { + const failedSessions = []; + for (const [sessionId, transport] of activeSessions) { + try { +- await transport.send(message); ++ await transport.send(message, options); } catch (error) { - logger.error(`Error sending message to session ${sessionId}: ${error}`); @@ -82,7 +158,7 @@ index b535515..ad7f146 100644 } } async close() { -@@ -256,6 +284,11 @@ export class HttpStreamTransport extends AbstractTransport { +@@ -256,6 +312,11 @@ export class HttpStreamTransport extends AbstractTransport { } } this._transports = {}; @@ -94,3 +170,14 @@ index b535515..ad7f146 100644 if (this._server) { this._server.close(); this._server = undefined; +diff --git a/node_modules/mcp-framework/dist/transports/http/server.d.ts b/node_modules/mcp-framework/dist/transports/http/server.d.ts +--- a/node_modules/mcp-framework/dist/transports/http/server.d.ts ++++ b/node_modules/mcp-framework/dist/transports/http/server.d.ts +@@ -13,6 +13,7 @@ export declare class HttpStreamTransport extends AbstractTransport { + private _transports; + constructor(config?: HttpStreamTransportConfig); + start(): Promise; ++ get port(): number; + private handleMcpRequest; + private readRequestBody; + private setCorsHeaders; diff --git a/scripts/download-content.ts b/scripts/download-content.ts index a259fcf..f666e41 100644 --- a/scripts/download-content.ts +++ b/scripts/download-content.ts @@ -22,13 +22,11 @@ export async function downloadDocs() { } export async function downloadExamples() { - console.log(`Downloading examples from appwrite/appwrite (version: ${appwriteExamplesBranch})`); + console.log(`Downloading examples from appwrite/specs (version: ${appwriteExamplesBranch})`); const owner = "appwrite"; - const repo = "appwrite"; - const docsSubdirPath = `docs/examples/${appwriteExamplesBranch}`; - // The version-pinned example folders (docs/examples/) only live on the - // `main` branch; they were removed from the per-version branches like 1.8.x. + const repo = "specs"; + const docsSubdirPath = `examples/${appwriteExamplesBranch}`; const ref = "main"; console.log(`Downloading examples from ${owner}/${repo}/${docsSubdirPath} to ${examplesTargetDir}`); @@ -54,4 +52,4 @@ async function main() { await createTableOfContents(); } -await main(); \ No newline at end of file +await main(); diff --git a/src/index.ts b/src/index.ts index dd5f352..4a0f6c6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -22,10 +22,10 @@ const server = new MCPServer({ responseMode: "stream", // Response mode: "batch" or "stream" (default: "batch") batchTimeout: 30000, // Timeout for batch responses in ms (default: 30000) session: { - enabled: true, - headerName: "Mcp-Session-Id", - allowClientTermination: true, - sessionTimeout: 300000, // 5 minutes + // The service runs with multiple replicas. Keeping MCP sessions in process + // memory makes follow-up requests fail whenever the load balancer sends them + // to another replica, so every request must be independently routable. + enabled: false, }, cors: { // CORS configuration @@ -33,7 +33,7 @@ const server = new MCPServer({ allowMethods: "GET, POST, DELETE, OPTIONS", allowHeaders: "Content-Type, Accept, Authorization, x-api-key, Mcp-Session-Id, Last-Event-ID", - exposeHeaders: "Content-Type, Authorization, x-api-key, Mcp-Session-Id", + exposeHeaders: "Content-Type, Authorization, x-api-key", maxAge: "86400", }, }, diff --git a/tests/fixtures/tools/ping.tool.js b/tests/fixtures/tools/ping.tool.js new file mode 100644 index 0000000..bacc39a --- /dev/null +++ b/tests/fixtures/tools/ping.tool.js @@ -0,0 +1,11 @@ +import { MCPTool } from "mcp-framework"; + +export default class PingTool extends MCPTool { + name = "ping"; + description = "Return a pong response"; + schema = {}; + + async execute() { + return "pong"; + } +} diff --git a/tests/http-stateless.test.mjs b/tests/http-stateless.test.mjs new file mode 100644 index 0000000..b604758 --- /dev/null +++ b/tests/http-stateless.test.mjs @@ -0,0 +1,171 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { fileURLToPath } from "node:url"; + +import { HttpStreamTransport, MCPServer } from "mcp-framework"; + +const fixturesPath = fileURLToPath(new URL("./fixtures", import.meta.url)); + +async function createTransport(onRequest) { + const transport = new HttpStreamTransport({ + port: 0, + endpoint: "/", + responseMode: "stream", + session: { enabled: false }, + }); + transport.onmessage = (message) => onRequest(transport, message); + await transport.start(); + return { port: transport.port, transport }; +} + +async function post(port, body, headers = {}) { + return fetch(`http://127.0.0.1:${port}/`, { + method: "POST", + headers: { + accept: "application/json, text/event-stream", + "content-type": "application/json", + ...headers, + }, + body: JSON.stringify(body), + }); +} + +function parseSse(responseBody) { + const data = responseBody + .split("\n") + .find((line) => line.startsWith("data: ")); + assert(data, `Expected an SSE data event, received: ${responseBody}`); + return JSON.parse(data.slice("data: ".length)); +} + +async function createMcpServer(name) { + const server = new MCPServer({ + name, + version: "1.0.0", + basePath: fixturesPath, + transport: { + type: "http-stream", + options: { + port: 0, + endpoint: "/", + responseMode: "stream", + session: { enabled: false }, + }, + }, + }); + const running = server.start(); + + for (let attempt = 0; attempt < 50; attempt += 1) { + const port = server.transport?.port; + if (!port) { + await new Promise((resolve) => setTimeout(resolve, 20)); + continue; + } + try { + const response = await fetch(`http://127.0.0.1:${port}/`, { + method: "OPTIONS", + }); + if (response.status === 204) return { port, running, server }; + } catch { + // The listener is not ready yet. + } + await new Promise((resolve) => setTimeout(resolve, 20)); + } + + await server.stop(); + await running; + throw new Error(`MCP server ${name} did not start`); +} + +test("stateless requests can land on different server instances", async (t) => { + const instanceA = await createTransport(async (transport, message) => { + await transport.send({ + jsonrpc: "2.0", + id: message.id, + result: { + protocolVersion: "2025-03-26", + capabilities: { tools: {} }, + serverInfo: { name: "instance-a", version: "1.0.0" }, + }, + }); + }); + const instanceB = await createTransport(async (transport, message) => { + await transport.send({ + jsonrpc: "2.0", + id: message.id, + result: { tools: [{ name: "search", description: "Search docs" }] }, + }); + }); + + t.after(async () => { + await Promise.all([ + instanceA.transport.close(), + instanceB.transport.close(), + ]); + }); + + const initializeResponse = await post(instanceA.port, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-03-26", + capabilities: {}, + clientInfo: { name: "regression-test", version: "1.0.0" }, + }, + }); + + assert.equal(initializeResponse.status, 200); + assert.equal(initializeResponse.headers.get("mcp-session-id"), null); + assert.equal( + parseSse(await initializeResponse.text()).result.serverInfo.name, + "instance-a", + ); + + const listResponse = await post( + instanceB.port, + { jsonrpc: "2.0", id: 2, method: "tools/list", params: {} }, + { "mcp-protocol-version": "2025-03-26" }, + ); + + assert.equal(listResponse.status, 200); + assert.equal(listResponse.headers.get("mcp-session-id"), null); + assert.deepEqual(parseSse(await listResponse.text()).result.tools, [ + { name: "search", description: "Search docs" }, + ]); +}); + +test("MCP framework handles initialize and tools/list on different replicas", async (t) => { + const instanceA = await createMcpServer("instance-a"); + const instanceB = await createMcpServer("instance-b"); + + t.after(async () => { + await Promise.all([instanceA.server.stop(), instanceB.server.stop()]); + await Promise.all([instanceA.running, instanceB.running]); + }); + + const initializeResponse = await post(instanceA.port, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-03-26", + capabilities: {}, + clientInfo: { name: "regression-test", version: "1.0.0" }, + }, + }); + assert.equal(initializeResponse.status, 200); + assert.equal(initializeResponse.headers.get("mcp-session-id"), null); + + const listResponse = await post( + instanceB.port, + { jsonrpc: "2.0", id: 2, method: "tools/list", params: {} }, + { "mcp-protocol-version": "2025-03-26" }, + ); + assert.equal(listResponse.status, 200); + assert.equal(listResponse.headers.get("mcp-session-id"), null); + assert.deepEqual( + parseSse(await listResponse.text()).result.tools.map(({ name }) => name), + ["ping"], + ); +});