diff --git a/AGENTS.md b/AGENTS.md index f6c90db..da938fe 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -24,7 +24,7 @@ src/ | | DYNAMIC | STATIC | |---|---|---| | Toolsets | Enabled at runtime via meta-tools | Pre-loaded at startup | -| Server instances | Per-client | Shared singleton | +| Server instances | Per-client | Per-client (startup toolsets preloaded) | | Meta-tools | All 5 registered | `list_tools` only | | Use case | Task-specific, lazy loading | Fixed pipelines | diff --git a/src/http/AGENTS.md b/src/http/AGENTS.md index eb4c1ac..e9628a2 100644 --- a/src/http/AGENTS.md +++ b/src/http/AGENTS.md @@ -33,8 +33,9 @@ Provides Fastify-based HTTP transport for the MCP protocol. Handles SSE streams, 1. POST /mcp (initialize) → isInitializeRequest(body) detects first contact → drainExistingSessions() — closes sessions in this bundle's map (same-client reconnect) - → disconnectServer() — closes server's current transport (STATIC cross-client case) + → disconnectServer() — closes the bundle server's current transport (stale same-client connection) → Set transport.onclose BEFORE server.connect() (SDK 1.26+ chain requirement) + → await bundle.orchestrator.isReady() — STATIC bundles preload tools async; SDK refuses capability registration after connect → server.connect(transport) — throws "Already connected" if above steps missed → Generate session ID, store in bundle.sessions via onsessioninitialized @@ -56,7 +57,7 @@ Provides Fastify-based HTTP transport for the MCP protocol. Handles SSE streams, `Protocol.connect()` throws `"Already connected"` if `this._transport` is still set. Two private methods handle the pre-connect cleanup: - **`drainExistingSessions(sessions)`** — covers same-bundle reconnects where sessions are still in the map (e.g. client aborted without DELETE but same `clientId`) -- **`disconnectServer(server)`** — covers cross-bundle cases in STATIC mode where different clients share one `McpServer`; `StreamableHTTPClientTransport.close()` does NOT send DELETE (it only aborts connections), so `_transport` can remain set after the first client leaves +- **`disconnectServer(server)`** — clears a bundle server still attached to a transport that left without DELETE; `StreamableHTTPClientTransport.close()` does NOT send DELETE (it only aborts connections), so `_transport` can remain set. Bundles never share a server, so this only ever closes the same client's stale connection **Critical ordering**: `transport.onclose` must be assigned BEFORE `server.connect(transport)`. `Protocol.connect()` reads the existing `onclose` and wraps it in a chain that also calls `Protocol._onclose()` (which clears `_transport`). If set after `connect()`, the chain is broken and `_transport` is never cleared on disconnect. diff --git a/src/http/FastifyTransport.ts b/src/http/FastifyTransport.ts index 060d2a3..29f88b0 100644 --- a/src/http/FastifyTransport.ts +++ b/src/http/FastifyTransport.ts @@ -238,6 +238,9 @@ export class FastifyTransport { bundle!.sessions.delete(transport.sessionId); }; try { + // STATIC bundles preload toolsets asynchronously; the SDK refuses to + // register tool capabilities once connected, so wait for them first. + await bundle.orchestrator.isReady(); await bundle.server.connect(transport); } catch (error) { reply.code(500); diff --git a/src/server/AGENTS.md b/src/server/AGENTS.md index 437b9c2..27ee851 100644 --- a/src/server/AGENTS.md +++ b/src/server/AGENTS.md @@ -14,10 +14,10 @@ Factory functions for creating MCP servers. Provides standard and permission-bas ## Invariants 1. **Startup config validated via Zod `.strict()`** — Catches typos in config keys (e.g., `initialToolsets` instead of `toolsets`). -2. **STATIC + sessionContext = warning** — Session context has limited effect in STATIC mode since all clients share one server. +2. **Every client gets its own server, in both modes** — The MCP SDK binds one `McpServer` to one transport, so a shared STATIC server made each new client disconnect the previous one. STATIC client orchestrators preload the startup toolsets with the client's merged session context. The base server/orchestrator only serves validation at startup and the `/tools` status endpoint. 3. **Permission-based servers reject `startup` field** — Throws if `startup` provided. Toolsets come from permissions, not config. 4. **Permission-based sanitizes exposure policy** — `sanitizeExposurePolicyForPermissions` strips allowlist/denylist/maxActiveToolsets with warnings. -5. **`ensureReady()` before transport start** — STATIC mode waits for toolset initialization to complete. +5. **`ensureReady()` before transport start** — STATIC mode waits for the base toolset initialization to complete (fail-fast). Per-client readiness is awaited by the transport before `connect()`. 6. **Orchestrator built via conditional builder calls** — Optional fields (`exposurePolicy`, `startup`) passed to builder only when defined. Never use `!` to coerce. ## Notifier Pattern @@ -29,7 +29,7 @@ Factory functions for creating MCP servers. Provides standard and permission-bas **Standard (createMcpServer):** ``` DYNAMIC: Per-client ServerOrchestrator + MCP Server (fresh per bundle) -STATIC: Shared singleton — all clients reuse one server + orchestrator +STATIC: Per-client ServerOrchestrator + MCP Server, startup toolsets preloaded per bundle ``` **Permission-based (createPermissionBasedMcpServer):** diff --git a/src/server/createMcpServer.ts b/src/server/createMcpServer.ts index 8a191d5..c319960 100644 --- a/src/server/createMcpServer.ts +++ b/src/server/createMcpServer.ts @@ -22,7 +22,7 @@ export async function createMcpServer( const mode: Exclude = options.startup?.mode ?? "DYNAMIC"; const shouldRegisterMetaTools = resolveMetaToolsFlag(options.registerMetaTools, mode); - const sessionContextResolver = buildSessionContextResolver(options, mode); + const sessionContextResolver = buildSessionContextResolver(options); const notifyToolsChanged = createToolsChangedNotifier(); // --- Build base server & orchestrator --- @@ -37,7 +37,7 @@ export async function createMcpServer( // --- Build transport --- const bundleFactory = createBundleFactory( - options, mode, baseServer, baseOrchestrator, shouldRegisterMetaTools, notifyToolsChanged + options, mode, shouldRegisterMetaTools, notifyToolsChanged ); const transport = buildTransport( options, baseOrchestrator.getManager(), bundleFactory, sessionContextResolver @@ -69,16 +69,13 @@ function validateOptions(options: CreateMcpServerOptions): void { } /** - * Validates session context config, builds the resolver, and warns about - * limited utility in STATIC mode. + * Validates session context config and builds the resolver. * * @param options - Server creation options containing sessionContext config - * @param mode - The resolved server mode * @returns A SessionContextResolver if configured, otherwise undefined */ function buildSessionContextResolver( - options: CreateMcpServerOptions, - mode: Exclude + options: CreateMcpServerOptions ): SessionContextResolver | undefined { if (!options.sessionContext) return undefined; @@ -91,13 +88,6 @@ function buildSessionContextResolver( .merge(options.sessionContext.merge ?? "shallow") .build(); - if (mode === "STATIC" && options.sessionContext.enabled !== false) { - console.warn( - "sessionContext has limited effect in STATIC mode: all clients share the same server instance with base context. " + - "Use DYNAMIC mode for per-session context isolation." - ); - } - return resolver; } @@ -141,13 +131,13 @@ function buildOrchestrator( /** * Creates the bundle factory callback for the transport layer. - * In STATIC mode all clients share one server + orchestrator. - * In DYNAMIC mode a fresh server + orchestrator is created per client. + * Every client gets a fresh server + orchestrator in both modes: the MCP SDK + * binds one McpServer to one transport, so a shared server would force each + * new client to disconnect the previous one. In STATIC mode the per-client + * orchestrator preloads the startup toolsets with the client's merged context. * * @param options - Server creation options - * @param mode - STATIC reuses base bundle; DYNAMIC creates fresh per client - * @param baseServer - The shared base server instance - * @param baseOrchestrator - The shared base orchestrator + * @param mode - The resolved server mode * @param shouldRegisterMetaTools - Pre-resolved meta-tools flag * @param notifyToolsChanged - Notifier function for tool list changes * @returns Bundle factory callback for the transport layer @@ -155,18 +145,10 @@ function buildOrchestrator( function createBundleFactory( options: CreateMcpServerOptions, mode: Exclude, - baseServer: McpServer, - baseOrchestrator: ServerOrchestrator, shouldRegisterMetaTools: boolean, notifyToolsChanged: (target: unknown) => Promise ): CreateBundleCallback { return (mergedContext?: unknown) => { - if (mode === "STATIC") { - // STATIC: all clients share one server + orchestrator - return { server: baseServer, orchestrator: baseOrchestrator }; - } - - // DYNAMIC: fresh server + orchestrator per client const effectiveContext = mergedContext ?? options.context; const clientServer: McpServer = options.createServer(); const clientOrchestrator = buildOrchestrator( diff --git a/tests/createMcpServer.test.ts b/tests/createMcpServer.test.ts index 0e8957b..719b788 100644 --- a/tests/createMcpServer.test.ts +++ b/tests/createMcpServer.test.ts @@ -113,7 +113,7 @@ describe("createMcpServer", () => { expect(args?.[3]).toEqual(configSchema); }); - it("reuses a single instance in STATIC mode bundles", async () => { + it("creates a fresh instance per bundle in STATIC mode", async () => { const f = makeFakeServerFactory(); await createMcpServer({ catalog, @@ -123,7 +123,9 @@ describe("createMcpServer", () => { const bundleFactory = (FastifyTransportMock as any).lastArgs?.[1]; const b1 = bundleFactory(); const b2 = bundleFactory(); - expect(b1.server).toBe(b2.server); + expect(f.created.length).toBe(3); + expect(b1.server).not.toBe(b2.server); + expect(b1.server).not.toBe(f.created[0].server); }); it("creates a fresh instance per bundle in DYNAMIC mode", async () => { @@ -152,7 +154,7 @@ describe("createMcpServer", () => { expect(s2.calls.includes("list_tools")).toBe(true); }); - it("does not re-register tools in STATIC mode across multiple clients", async () => { + it("preloads startup toolsets once on each STATIC client server", async () => { const f = makeFakeServerFactory(); const staticCatalog = { core: { @@ -180,12 +182,40 @@ describe("createMcpServer", () => { expect(base.calls.filter((n) => n === "core.ping").length).toBe(1); const bundleFactory = (FastifyTransportMock as any).lastArgs?.[1]; - // Simulate two more clients (bundles) - bundleFactory(); - bundleFactory(); + // Simulate two clients (bundles) + const b1 = bundleFactory(); + const b2 = bundleFactory(); + await b1.orchestrator.ensureReady(); + await b2.orchestrator.ensureReady(); - // No additional registrations should have occurred on the shared server + // The base server is untouched; each client server registers the tool once expect(base.calls.filter((n) => n === "core.ping").length).toBe(1); + for (const client of f.created.slice(1)) { + expect(client.calls.filter((n) => n === "core.ping").length).toBe(1); + } + }); + + it("passes the merged session context to STATIC client module loaders", async () => { + const f = makeFakeServerFactory(); + const seen: unknown[] = []; + await createMcpServer({ + catalog: { tenant: { name: "Tenant", description: "", modules: ["tenant"] } } as any, + moduleLoaders: { + tenant: async (ctx?: unknown) => { + seen.push(ctx); + return []; + }, + }, + context: { base: true }, + startup: { mode: "STATIC", toolsets: ["tenant"] }, + createServer: f.createServer, + }); + expect(seen).toEqual([{ base: true }]); + + const bundleFactory = (FastifyTransportMock as any).lastArgs?.[1]; + const bundle = bundleFactory({ base: true, API_TOKEN: "client-token" }); + await bundle.orchestrator.ensureReady(); + expect(seen[1]).toEqual({ base: true, API_TOKEN: "client-token" }); }); it("ensures tools are ready immediately in STATIC mode (waits for async module loaders)", async () => { diff --git a/tests/e2e/staticMode.e2e.test.ts b/tests/e2e/staticMode.e2e.test.ts index fd059d2..e27232a 100644 --- a/tests/e2e/staticMode.e2e.test.ts +++ b/tests/e2e/staticMode.e2e.test.ts @@ -8,6 +8,8 @@ import { testCatalog, extractToolNames, extractTextContent, + sessionContextCatalog, + sessionContextModuleLoaders, } from "./helpers.js"; describe("STATIC mode E2E", () => { @@ -131,6 +133,115 @@ describe("STATIC mode E2E", () => { ); }); + describe("with concurrent clients", () => { + let server: Awaited>; + let port: number; + + beforeAll(async () => { + port = await getAvailablePort(); + server = await createMcpServer({ + createServer: () => + new McpServer({ + name: "test-static-concurrent", + version: "1.0.0", + }), + catalog: testCatalog, + startup: { mode: "STATIC", toolsets: ["core"] }, + http: { port }, + }); + await server.start(); + }); + + afterAll(async () => { + await server.close(); + }); + + it("keeps the first client connected when a second client connects", async () => { + const connect = async (clientId: string) => { + const transport = new StreamableHTTPClientTransport( + new URL(`http://localhost:${port}/mcp`), + { requestInit: { headers: { "mcp-client-id": clientId } } } + ); + const client = new Client({ name: "test", version: "1.0.0" }); + await client.connect(transport); + return client; + }; + + const client1 = await connect("concurrent-client-1"); + const client2 = await connect("concurrent-client-2"); + + // Both sessions stay usable at the same time + const result1 = await client1.callTool({ name: "core.ping", arguments: {} } as any); + const result2 = await client2.callTool({ name: "core.ping", arguments: {} } as any); + expect(extractTextContent(result1)).toBe("pong"); + expect(extractTextContent(result2)).toBe("pong"); + + await client1.close(); + await client2.close(); + }); + }); + + describe("with session context", () => { + let server: Awaited>; + let port: number; + + beforeAll(async () => { + port = await getAvailablePort(); + server = await createMcpServer({ + createServer: () => + new McpServer({ + name: "test-static-session-context", + version: "1.0.0", + }), + catalog: sessionContextCatalog, + moduleLoaders: sessionContextModuleLoaders, + context: { baseValue: "shared" }, + sessionContext: { + enabled: true, + queryParam: { + name: "config", + encoding: "base64", + allowedKeys: ["API_TOKEN"], + }, + merge: "shallow", + }, + startup: { mode: "STATIC", toolsets: "ALL" }, + http: { port }, + }); + await server.start(); + }); + + afterAll(async () => { + await server.close(); + }); + + it("passes per-client session context to preloaded tools", async () => { + const connect = async (clientId: string, token: string) => { + const config = Buffer.from(JSON.stringify({ API_TOKEN: token })).toString("base64"); + const transport = new StreamableHTTPClientTransport( + new URL(`http://localhost:${port}/mcp?config=${config}`), + { requestInit: { headers: { "mcp-client-id": clientId } } } + ); + const client = new Client({ name: "test", version: "1.0.0" }); + await client.connect(transport); + return client; + }; + const readContext = async (client: Client) => { + const result = await client.callTool({ name: "tenant.get_context", arguments: {} } as any); + return JSON.parse(extractTextContent(result)); + }; + + const clientA = await connect("static-session-a", "token-a"); + const clientB = await connect("static-session-b", "token-b"); + + expect(await readContext(clientA)).toEqual({ API_TOKEN: "token-a", USER_ID: null, baseValue: "shared" }); + expect(await readContext(clientB)).toEqual({ API_TOKEN: "token-b", USER_ID: null, baseValue: "shared" }); + + await clientA.close(); + await clientB.close(); + }); + }); + describe("with ALL toolsets", () => { let server: Awaited>; let port: number;