Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 |

Expand Down
5 changes: 3 additions & 2 deletions src/http/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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.

Expand Down
3 changes: 3 additions & 0 deletions src/http/FastifyTransport.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
6 changes: 3 additions & 3 deletions src/server/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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):**
Expand Down
36 changes: 9 additions & 27 deletions src/server/createMcpServer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ export async function createMcpServer(

const mode: Exclude<Mode, "ALL"> = 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 ---
Expand All @@ -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
Expand Down Expand Up @@ -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<Mode, "ALL">
options: CreateMcpServerOptions
): SessionContextResolver | undefined {
if (!options.sessionContext) return undefined;

Expand All @@ -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;
}

Expand Down Expand Up @@ -141,32 +131,24 @@ 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
*/
function createBundleFactory(
options: CreateMcpServerOptions,
mode: Exclude<Mode, "ALL">,
baseServer: McpServer,
baseOrchestrator: ServerOrchestrator,
shouldRegisterMetaTools: boolean,
notifyToolsChanged: (target: unknown) => Promise<void>
): 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(
Expand Down
44 changes: 37 additions & 7 deletions tests/createMcpServer.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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 () => {
Expand Down Expand Up @@ -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: {
Expand Down Expand Up @@ -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 () => {
Expand Down
111 changes: 111 additions & 0 deletions tests/e2e/staticMode.e2e.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@ import {
testCatalog,
extractToolNames,
extractTextContent,
sessionContextCatalog,
sessionContextModuleLoaders,
} from "./helpers.js";

describe("STATIC mode E2E", () => {
Expand Down Expand Up @@ -131,6 +133,115 @@ describe("STATIC mode E2E", () => {
);
});

describe("with concurrent clients", () => {
let server: Awaited<ReturnType<typeof createMcpServer>>;
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<ReturnType<typeof createMcpServer>>;
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<ReturnType<typeof createMcpServer>>;
let port: number;
Expand Down
Loading