feat(ai): MCP client over stdio and streamable HTTP
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Lucas Winther
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
50e1e49b6d
commit
24cd6329cb
@@ -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";
|
||||
|
||||
@@ -0,0 +1,188 @@
|
||||
import type { McpServerConfig } from "../settings";
|
||||
import { readSse } from "../providers/sse";
|
||||
|
||||
export interface McpTool { name: string; description: string; inputSchema: Record<string, unknown> }
|
||||
export interface McpClient {
|
||||
tools(): Promise<McpTool[]>;
|
||||
call(name: string, args: unknown): Promise<string>;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
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<any>; notify(method: string, params?: unknown): Promise<void>; close(): Promise<void> }
|
||||
|
||||
const withTimeout = <T>(p: Promise<T>, ms: number, what: string): Promise<T> =>
|
||||
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<number, { resolve(v: RpcReply): void; reject(e: Error): void }>();
|
||||
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<Uint8Array>) {
|
||||
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<RpcReply>((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<string, string> = {
|
||||
"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<string, string> = { "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<McpClient> {
|
||||
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);
|
||||
}
|
||||
+32
@@ -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");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<string, string> = 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);
|
||||
});
|
||||
Reference in New Issue
Block a user