From 24cd6329cb0550d57bbe91cf8b23baf9763c8562 Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Fri, 30 Oct 2026 03:19:13 +0100 Subject: [PATCH] feat(ai): MCP client over stdio and streamable HTTP Co-Authored-By: Claude Opus 5.5 --- packages/feature-ai/src/index.ts | 1 + packages/feature-ai/src/mcp/client.ts | 188 ++++++++++++++++++ packages/feature-ai/test/fixtures/fake-mcp.ts | 32 +++ packages/feature-ai/test/mcp.test.ts | 57 ++++++ 4 files changed, 278 insertions(+) create mode 100644 packages/feature-ai/src/mcp/client.ts create mode 100644 packages/feature-ai/test/fixtures/fake-mcp.ts create mode 100644 packages/feature-ai/test/mcp.test.ts diff --git a/packages/feature-ai/src/index.ts b/packages/feature-ai/src/index.ts index 9489187..e7c28f0 100644 --- a/packages/feature-ai/src/index.ts +++ b/packages/feature-ai/src/index.ts @@ -6,3 +6,4 @@ export * from "./settings"; export * from "./providers"; export * from "./retrieval/course-index"; export { Bm25, tokenize } from "./retrieval/bm25"; +export * from "./mcp/client"; diff --git a/packages/feature-ai/src/mcp/client.ts b/packages/feature-ai/src/mcp/client.ts new file mode 100644 index 0000000..1bdd0af --- /dev/null +++ b/packages/feature-ai/src/mcp/client.ts @@ -0,0 +1,188 @@ +import type { McpServerConfig } from "../settings"; +import { readSse } from "../providers/sse"; + +export interface McpTool { name: string; description: string; inputSchema: Record } +export interface McpClient { + tools(): Promise; + call(name: string, args: unknown): Promise; + close(): Promise; +} + +const PROTOCOL_VERSION = "2025-06-18"; +const CLIENT_INFO = { name: "stcourses", version: "1.0.0" }; + +type RpcReply = { id?: number | string; result?: any; error?: { code: number; message: string } }; +interface Transport { request(method: string, params: unknown): Promise; notify(method: string, params?: unknown): Promise; close(): Promise } + +const withTimeout = (p: Promise, ms: number, what: string): Promise => + new Promise((resolve, reject) => { + const t = setTimeout(() => reject(new Error(`MCP ${what} timed out after ${ms} ms`)), ms); + p.then((v) => { clearTimeout(t); resolve(v); }, (e) => { clearTimeout(t); reject(e); }); + }); + +function unwrap(r: RpcReply): any { + if (r.error) throw new Error(`MCP error ${r.error.code}: ${r.error.message}`); + return r.result; +} + +function stdioTransport(cfg: McpServerConfig, timeoutMs: number): Transport { + if (!cfg.command) throw new Error(`MCP server "${cfg.name}" has no command`); + const proc = Bun.spawn([cfg.command, ...(cfg.args ?? [])], { + stdin: "pipe", + stdout: "pipe", + stderr: "ignore", + env: { ...process.env, ...(cfg.env ?? {}) }, + }); + let nextId = 1; + const pending = new Map(); + let closed = false; + const failAll = (e: Error) => { for (const p of pending.values()) p.reject(e); pending.clear(); }; + + (async () => { + const decoder = new TextDecoder(); + let buf = ""; + try { + for await (const chunk of proc.stdout as unknown as AsyncIterable) { + buf += decoder.decode(chunk, { stream: true }); + let nl: number; + while ((nl = buf.indexOf("\n")) !== -1) { + const line = buf.slice(0, nl).trim(); + buf = buf.slice(nl + 1); + if (!line) continue; + let msg: RpcReply; + try { msg = JSON.parse(line); } catch { continue; } + if (typeof msg.id === "number" && pending.has(msg.id)) { + pending.get(msg.id)!.resolve(msg); + pending.delete(msg.id); + } + } + } + } catch {} + closed = true; + failAll(new Error(`MCP server "${cfg.name}" exited`)); + })(); + proc.exited.then((code) => { closed = true; failAll(new Error(`MCP server "${cfg.name}" exited with code ${code}`)); }); + + const send = (obj: unknown) => { + if (closed) throw new Error(`MCP server "${cfg.name}" is not running`); + proc.stdin.write(JSON.stringify(obj) + "\n"); + proc.stdin.flush(); + }; + return { + async request(method, params) { + const id = nextId++; + const reply = new Promise((resolve, reject) => pending.set(id, { resolve, reject })); + send({ jsonrpc: "2.0", id, method, params }); + return unwrap(await withTimeout(reply, timeoutMs, method)); + }, + async notify(method, params) { send({ jsonrpc: "2.0", method, ...(params ? { params } : {}) }); }, + async close() { + closed = true; + try { proc.stdin.end(); } catch {} + proc.kill(); + await withTimeout(proc.exited, 2000, "shutdown").catch(() => proc.kill(9)); + }, + }; +} + +function httpTransport(cfg: McpServerConfig, timeoutMs: number, doFetch: typeof fetch): Transport { + if (!cfg.url) throw new Error(`MCP server "${cfg.name}" has no URL`); + const url = cfg.url; + let session: string | null = null; + let nextId = 1; + const post = async (body: unknown, signal: AbortSignal) => { + const headers: Record = { + "content-type": "application/json", + accept: "application/json, text/event-stream", + "mcp-protocol-version": PROTOCOL_VERSION, + ...(cfg.headers ?? {}), + }; + if (session) headers["mcp-session-id"] = session; + const res = await doFetch(url, { method: "POST", headers, body: JSON.stringify(body), signal }); + const sid = res.headers.get("mcp-session-id"); + if (sid) session = sid; + return res; + }; + return { + async request(method, params) { + const id = nextId++; + const ctl = new AbortController(); + const t = setTimeout(() => ctl.abort(), timeoutMs); + try { + const res = await post({ jsonrpc: "2.0", id, method, params }, ctl.signal); + if (!res.ok) throw new Error(`MCP HTTP ${res.status} from "${cfg.name}"`); + const type = res.headers.get("content-type") ?? ""; + if (type.includes("text/event-stream") && res.body) { + for await (const { data } of readSse(res.body)) { + let msg: RpcReply; + try { msg = JSON.parse(data); } catch { continue; } + if (msg.id === id) { ctl.abort(); return unwrap(msg); } + } + throw new Error(`MCP stream from "${cfg.name}" ended without a reply`); + } + const msg = await res.json(); + const reply = Array.isArray(msg) ? msg.find((m: RpcReply) => m.id === id) : msg; + return unwrap(reply ?? { error: { code: -1, message: "no reply" } }); + } catch (e) { + if (ctl.signal.aborted && (e as Error).name === "AbortError") throw new Error(`MCP ${method} timed out after ${timeoutMs} ms`); + throw e; + } finally { clearTimeout(t); } + }, + async notify(method, params) { + const ctl = new AbortController(); + const t = setTimeout(() => ctl.abort(), timeoutMs); + try { await post({ jsonrpc: "2.0", method, ...(params ? { params } : {}) }, ctl.signal); } finally { clearTimeout(t); } + }, + async close() { + if (!session) return; + const headers: Record = { "mcp-session-id": session, ...(cfg.headers ?? {}) }; + await doFetch(url, { method: "DELETE", headers }).catch(() => {}); + }, + }; +} + +/** Connects and initializes an MCP server (stdio or streamable HTTP). */ +export async function connectMcp(cfg: McpServerConfig, opts: { fetch?: typeof fetch; timeoutMs?: number } = {}): Promise { + const timeoutMs = opts.timeoutMs ?? 20_000; + const t = cfg.transport === "stdio" ? stdioTransport(cfg, timeoutMs) : httpTransport(cfg, timeoutMs, opts.fetch ?? fetch); + try { + await t.request("initialize", { protocolVersion: PROTOCOL_VERSION, capabilities: {}, clientInfo: CLIENT_INFO }); + await t.notify("notifications/initialized"); + } catch (e) { + await t.close().catch(() => {}); + throw e; + } + return { + async tools() { + const out: McpTool[] = []; + let cursor: string | undefined; + do { + const r = await t.request("tools/list", cursor ? { cursor } : {}); + for (const tool of r?.tools ?? []) out.push({ name: tool.name, description: tool.description ?? "", inputSchema: tool.inputSchema ?? { type: "object" } }); + cursor = r?.nextCursor; + } while (cursor); + return out; + }, + async call(name, args) { + const r = await t.request("tools/call", { name, arguments: args ?? {} }); + const text = (r?.content ?? []) + .map((c: any) => (c.type === "text" ? c.text : c.type === "resource" ? (c.resource?.text ?? `[resource ${c.resource?.uri}]`) : `[${c.type} content]`)) + .join("\n"); + const body = text || (r?.structuredContent ? JSON.stringify(r.structuredContent) : ""); + if (r?.isError) throw new Error(body || "tool failed"); + return body; + }, + close: () => t.close(), + }; +} + +/** Global servers (no courseIds) plus servers that list this course. */ +export function mcpServersFor(courseId: string, servers: McpServerConfig[]): McpServerConfig[] { + return servers.filter((s) => !s.courseIds?.length || s.courseIds.includes(courseId)); +} + +/** Provider-safe tool name (^[a-zA-Z0-9_-]{1,64}$) for an MCP tool. */ +export function mcpToolName(serverId: string, tool: string): string { + const clean = (s: string) => s.replace(/[^a-zA-Z0-9_-]/g, "_"); + return `mcp_${clean(serverId).slice(0, 20)}__${clean(tool)}`.slice(0, 64); +} diff --git a/packages/feature-ai/test/fixtures/fake-mcp.ts b/packages/feature-ai/test/fixtures/fake-mcp.ts new file mode 100644 index 0000000..8ac7e4b --- /dev/null +++ b/packages/feature-ai/test/fixtures/fake-mcp.ts @@ -0,0 +1,32 @@ +// A fake MCP server for tests. stdio: newline-delimited JSON-RPC on stdin/stdout. +// Exported `handle` is reused by the HTTP fake in mcp.test.ts. +export function handle(msg: any): any | null { + if (msg.id === undefined) return null; // notification + switch (msg.method) { + case "initialize": + return { jsonrpc: "2.0", id: msg.id, result: { protocolVersion: "2025-06-18", capabilities: { tools: {} }, serverInfo: { name: "fake", version: "1" } } }; + case "tools/list": + return { jsonrpc: "2.0", id: msg.id, result: { tools: [{ name: "echo", description: "Echo text back", inputSchema: { type: "object", properties: { text: { type: "string" } }, required: ["text"] } }] } }; + case "tools/call": + if (msg.params?.name !== "echo") return { jsonrpc: "2.0", id: msg.id, error: { code: -32602, message: `Unknown tool ${msg.params?.name}` } }; + return { jsonrpc: "2.0", id: msg.id, result: { content: [{ type: "text", text: `echo: ${msg.params.arguments?.text}` }] } }; + default: + return { jsonrpc: "2.0", id: msg.id, error: { code: -32601, message: "Method not found" } }; + } +} + +if (import.meta.main) { + if (process.argv.includes("--exit-immediately")) process.exit(3); + let buf = ""; + for await (const chunk of Bun.stdin.stream()) { + buf += new TextDecoder().decode(chunk); + let nl: number; + while ((nl = buf.indexOf("\n")) !== -1) { + const line = buf.slice(0, nl).trim(); + buf = buf.slice(nl + 1); + if (!line) continue; + const reply = handle(JSON.parse(line)); + if (reply) process.stdout.write(JSON.stringify(reply) + "\n"); + } + } +} diff --git a/packages/feature-ai/test/mcp.test.ts b/packages/feature-ai/test/mcp.test.ts new file mode 100644 index 0000000..47b78d9 --- /dev/null +++ b/packages/feature-ai/test/mcp.test.ts @@ -0,0 +1,57 @@ +import { afterAll, expect, test } from "bun:test"; +import { join } from "node:path"; +import { connectMcp, mcpServersFor, mcpToolName } from "../src/mcp/client"; +import { handle } from "./fixtures/fake-mcp"; + +const FAKE = join(import.meta.dir, "fixtures/fake-mcp.ts"); + +test("stdio: lists and calls tools", async () => { + const c = await connectMcp({ id: "fake", name: "Fake", transport: "stdio", command: process.execPath, args: [FAKE] }); + try { + expect((await c.tools()).map((t) => t.name)).toEqual(["echo"]); + expect(await c.call("echo", { text: "hi" })).toBe("echo: hi"); + await expect(c.call("nope", {})).rejects.toThrow(/Unknown tool/); + } finally { await c.close(); } +}); + +test("stdio: a server that exits rejects instead of hanging", async () => { + await expect(connectMcp({ id: "x", name: "X", transport: "stdio", command: process.execPath, args: [FAKE, "--exit-immediately"] }, { timeoutMs: 5000 })).rejects.toThrow(); +}); + +const sessions: (string | null)[] = []; +const server = Bun.serve({ + port: 0, + async fetch(req) { + if (req.method === "DELETE") return new Response(null, { status: 204 }); + const msg = await req.json(); + sessions.push(req.headers.get("mcp-session-id")); + const reply = handle(msg); + if (!reply) return new Response(null, { status: 202 }); + const headers: Record = msg.method === "initialize" ? { "mcp-session-id": "sess-42" } : {}; + if (msg.method === "tools/call") { + return new Response(`event: message\ndata: ${JSON.stringify(reply)}\n\n`, { headers: { ...headers, "content-type": "text/event-stream" } }); + } + return Response.json(reply, { headers }); + }, +}); +afterAll(() => server.stop(true)); + +test("streamable HTTP: JSON and SSE replies, session id echoed", async () => { + const c = await connectMcp({ id: "web", name: "Web", transport: "http", url: `http://127.0.0.1:${server.port}/mcp` }); + expect((await c.tools()).map((t) => t.name)).toEqual(["echo"]); + expect(await c.call("echo", { text: "over http" })).toBe("echo: over http"); + await c.close(); + expect(sessions[0]).toBeNull(); + expect(sessions.slice(1).every((s) => s === "sess-42")).toBe(true); +}); + +test("server selection and tool names", () => { + const servers = [ + { id: "g", name: "G", transport: "http" as const, url: "http://x" }, + { id: "s", name: "S", transport: "http" as const, url: "http://y", courseIds: ["sample"] }, + { id: "o", name: "O", transport: "http" as const, url: "http://z", courseIds: ["other"] }, + ]; + expect(mcpServersFor("sample", servers).map((s) => s.id)).toEqual(["g", "s"]); + expect(mcpToolName("my server", "read file!")).toBe("mcp_my_server__read_file_"); + expect(mcpToolName("a".repeat(40), "b".repeat(40)).length).toBeLessThanOrEqual(64); +});