diff --git a/packages/simplex-chat-mcp/src/protocol.ts b/packages/simplex-chat-mcp/src/protocol.ts new file mode 100644 index 0000000000..64b1910564 --- /dev/null +++ b/packages/simplex-chat-mcp/src/protocol.ts @@ -0,0 +1,50 @@ +import {CallToolResult} from "@modelcontextprotocol/server" +import {Config} from "./config" +import {ToolName} from "./tools" + +export const protocolVersion = 1 + +export interface DaemonInfo { + version: string + protocol: number +} + +export interface OwnerState { + owners: string[] + link?: string +} + +export interface SessionRequest { + id: string + name?: string + project: string + config: Config +} + +export interface SessionReply extends DaemonInfo { + number: number + displayName: string + owner: OwnerState +} + +export interface AttachRequest { + id: string +} + +export interface CallRequest { + id: string + tool: ToolName + input: unknown +} + +export type CallLine = {progress: string} | {result: CallToolResult | null} + +export interface StatusReply extends DaemonInfo { + mainGroup: string + owner: OwnerState + sessions: {number: number; name: string; attached: number}[] +} + +export interface ResetReply { + link: string +} diff --git a/packages/simplex-chat-mcp/src/socket.ts b/packages/simplex-chat-mcp/src/socket.ts new file mode 100644 index 0000000000..483a0d0669 --- /dev/null +++ b/packages/simplex-chat-mcp/src/socket.ts @@ -0,0 +1,129 @@ +import {timingSafeEqual} from "crypto" +import {existsSync, unlinkSync} from "fs" +import {IncomingMessage, Server, ServerResponse, createServer, request} from "http" +import {logError} from "./log" + +const maxBody = 1_048_576 + +export type Handler = (body: unknown, res: ServerResponse, req: IncomingMessage) => Promise + +export class SocketError extends Error { + constructor(message: string, readonly status: number) { + super(message) + } +} + +export function listen(path: string, token: string, routes: {[route: string]: Handler}): Promise { + const expected = Buffer.from(`Bearer ${token}`) + const authorized = (header: string | undefined): boolean => { + const given = Buffer.from(header ?? "") + return given.length === expected.length && timingSafeEqual(given, expected) + } + const server = createServer((req, res) => { + if (!authorized(req.headers.authorization)) return sendJson(res, 401, {error: "Unauthorized"}) + const handler = routes[`${req.method} ${req.url}`] + if (handler === undefined) return sendJson(res, 404, {error: "Not found"}) + readBody(req) + .then(body => handler(body, res, req)) + .catch(error => { + if (error instanceof SocketError) return sendJson(res, error.status, {error: error.message}) + logError(`${req.method} ${req.url}`, error) + if (!res.headersSent) sendJson(res, 500, {error: error instanceof Error ? error.message : String(error)}) + else res.destroy() + }) + }) + server.requestTimeout = 0 + server.headersTimeout = 60_000 + server.keepAliveTimeout = 5_000 + server.timeout = 0 + if (process.platform !== "win32" && existsSync(path)) unlinkSync(path) + return new Promise((resolve, reject) => { + server.once("error", reject) + server.listen(path, () => { + server.off("error", reject) + resolve(server) + }) + }) +} + +export function closeServer(server: Server): Promise { + const closed = new Promise(resolve => server.close(() => resolve())) + server.closeAllConnections() + return closed +} + +export function sendJson(res: ServerResponse, status: number, value: unknown): void { + res.writeHead(status, {"content-type": "application/json"}) + res.end(JSON.stringify(value)) +} + +export function startLines(res: ServerResponse): (line: unknown) => void { + res.writeHead(200, {"content-type": "application/x-ndjson"}) + return line => { + res.write(`${JSON.stringify(line)}\n`) + } +} + +export function isNotRunning(error: unknown): boolean { + const code = (error as NodeJS.ErrnoException | undefined)?.code + return code === "ENOENT" || code === "ECONNREFUSED" +} + +export async function requestJson(path: string, token: string, method: string, route: string, body?: unknown): Promise { + const res = await send(path, token, method, route, body) + const text = await readText(res) + if (res.statusCode !== 200) throw new SocketError(errorOf(text), res.statusCode ?? 0) + return JSON.parse(text) as T +} + +export async function streamLines(path: string, token: string, route: string, body: unknown, onLine: (line: unknown) => void, signal?: AbortSignal): Promise { + const res = await send(path, token, "POST", route, body, signal) + if (res.statusCode !== 200) throw new SocketError(errorOf(await readText(res)), res.statusCode ?? 0) + let buffer = "" + for await (const chunk of res) { + buffer += chunk + for (let i = buffer.indexOf("\n"); i >= 0; i = buffer.indexOf("\n")) { + onLine(JSON.parse(buffer.slice(0, i))) + buffer = buffer.slice(i + 1) + } + } +} + +function send(path: string, token: string, method: string, route: string, body: unknown, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + const req = request({socketPath: path, method, path: route, headers: {authorization: `Bearer ${token}`, "content-type": "application/json"}, signal}, resolve) + req.on("error", reject) + req.end(body === undefined ? undefined : JSON.stringify(body)) + }) +} + +async function readBody(req: IncomingMessage): Promise { + let text = "" + let tooLarge = false + for await (const chunk of req) { + if (tooLarge) continue + text += chunk + tooLarge = text.length > maxBody + } + if (tooLarge) throw new SocketError("Body too large", 413) + if (text === "") return undefined + try { + return JSON.parse(text) + } catch { + throw new SocketError("Invalid JSON", 400) + } +} + +async function readText(res: IncomingMessage): Promise { + let text = "" + for await (const chunk of res) text += chunk + return text +} + +function errorOf(text: string): string { + try { + return (JSON.parse(text) as {error?: string}).error ?? text + } catch { + return text + } +} diff --git a/packages/simplex-chat-mcp/test/unit/socket.test.ts b/packages/simplex-chat-mcp/test/unit/socket.test.ts new file mode 100644 index 0000000000..24ca633fb1 --- /dev/null +++ b/packages/simplex-chat-mcp/test/unit/socket.test.ts @@ -0,0 +1,71 @@ +import {once} from "events" +import {existsSync, mkdtempSync, writeFileSync} from "fs" +import {Server} from "http" +import {tmpdir} from "os" +import {join} from "path" +import {SocketError, closeServer, isNotRunning, listen, requestJson, sendJson, startLines, streamLines} from "../../src/socket" + +let dir: string +let path: string +let server: Server | undefined +const token = "t0ken" + +beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "simplex-mcp-socket-")) + path = join(dir, "daemon.sock") +}) + +afterEach(async () => { + if (server) await closeServer(server) + server = undefined +}) + +describe("socket", () => { + it("checks the token, the route and the body", async () => { + server = await listen(path, token, { + "POST /echo": async (body, res) => sendJson(res, 200, body), + "POST /fail": async () => { + throw new Error("broken") + }, + }) + expect(await requestJson(path, token, "POST", "/echo", {a: 1})).toEqual({a: 1}) + await expect(requestJson(path, "wrong", "POST", "/echo", {})).rejects.toEqual(new SocketError("Unauthorized", 401)) + await expect(requestJson(path, token, "GET", "/echo")).rejects.toEqual(new SocketError("Not found", 404)) + await expect(requestJson(path, token, "POST", "/fail", {})).rejects.toEqual(new SocketError("broken", 500)) + await expect(requestJson(path, token, "POST", "/echo", "x".repeat(1_048_577))).rejects.toEqual(new SocketError("Body too large", 413)) + }) + + it("streams lines and aborts the request on the signal", async () => { + let closedEarly: Promise | undefined + server = await listen(path, token, { + "POST /lines": async (_body, res) => { + const write = startLines(res) + write({progress: "one"}) + write({progress: "two"}) + closedEarly = once(res, "close") + }, + }) + const lines: unknown[] = [] + const controller = new AbortController() + const streaming = streamLines(path, token, "/lines", {}, line => { + lines.push(line) + if (lines.length === 2) controller.abort() + }, controller.signal) + await expect(streaming).rejects.toThrow() + expect(lines).toEqual([{progress: "one"}, {progress: "two"}]) + await closedEarly + }) + + it("reports a daemon that is not running by the error code", async () => { + const error = await requestJson(path, token, "GET", "/status").catch(e => e) + expect(isNotRunning(error)).toBe(true) + expect(isNotRunning(new SocketError("x", 500))).toBe(false) + }) + + it("replaces a stale socket file", async () => { + writeFileSync(path, "") + server = await listen(path, token, {"GET /ok": async (_body, res) => sendJson(res, 200, {})}) + expect(existsSync(path)).toBe(true) + expect(await requestJson(path, token, "GET", "/ok")).toEqual({}) + }) +})