theseus/bundled-addons/claude-connector/lib/mcp-server.cjs

// MCP over Streamable HTTP, loopback only.
//
//   POST   /mcp  JSON-RPC request(s) → application/json response
//   GET    /mcp  SSE stream for server notifications (also the "a client is
//                still here" signal behind the toolbar indicator)
//   DELETE /mcp  end the session
//
// Every request needs `Authorization: Bearer <token>`. Requests with an
// Origin header (a web page) or a Host other than this loopback address
// (DNS rebinding) are refused before the token is even looked at.

const http = require("node:http");
const crypto = require("node:crypto");

const PROTOCOL_VERSIONS = ["2025-06-18", "2025-03-26", "2024-11-05"];
const MAX_BODY = 32 * 1024 * 1024;
// A session counts as connected while its SSE stream is open, or for this
// long after its last request.
const IDLE_MS = 2 * 60 * 1000;
const SESSION_TTL_MS = 12 * 60 * 60 * 1000;

const sha = (s) => crypto.createHash("sha256").update(String(s)).digest();
function tokenMatches(given, expected) {
  if (!given || !expected) return false;
  return crypto.timingSafeEqual(sha(given), sha(expected));
}

class McpServer {
  // tools: { list(): [toolDef], call(name, args, session) => Promise<result> }
  constructor({ getToken, tools, serverInfo, instructions, onStatus, log }) {
    this.getToken = getToken;
    this.tools = tools;
    this.serverInfo = serverInfo || { name: "theseus", version: "0.0.0" };
    this.instructions = instructions || "";
    this.onStatus = typeof onStatus === "function" ? onStatus : () => {};
    this.log = typeof log === "function" ? log : () => {};
    this.sessions = new Map(); // id -> { id, client, protocolVersion, created, lastSeen, streams:Set<res> }
    this.server = null;
    this.port = null;
    this._lastStatus = "";
    this._sweep = null;
  }

  get url() { return this.port ? `http://127.0.0.1:${this.port}/mcp` : null; }

  listen(port) {
    return new Promise((resolve, reject) => {
      const srv = http.createServer((req, res) => this._handle(req, res).catch((e) => {
        this.log("request failed:", e?.message || e);
        if (!res.headersSent) this._json(res, 500, { jsonrpc: "2.0", id: null, error: { code: -32603, message: "internal error" } });
        else try { res.end(); } catch {}
      }));
      srv.keepAliveTimeout = 5000;
      srv.requestTimeout = 0; // tool calls may wait on the user (site approval)
      srv.headersTimeout = 20000;
      const fail = (e) => { srv.removeListener("listening", ok); reject(e); };
      const ok = () => {
        srv.removeListener("error", fail);
        this.server = srv;
        this.port = srv.address().port;
        this._sweep = setInterval(() => this._expire(), 15000);
        this._sweep.unref?.();
        resolve(this.port);
      };
      srv.once("error", fail);
      srv.once("listening", ok);
      srv.listen(port, "127.0.0.1");
    });
  }

  async close() {
    clearInterval(this._sweep);
    this._sweep = null;
    this.dropAllSessions();
    const srv = this.server;
    this.server = null;
    this.port = null;
    if (srv) await new Promise((r) => { srv.close(() => r()); srv.closeAllConnections?.(); });
    this._emitStatus();
  }

  dropAllSessions() {
    for (const s of this.sessions.values()) for (const r of s.streams) { try { r.end(); } catch {} }
    this.sessions.clear();
    this._emitStatus();
  }

  // Tell every connected client that the tool list changed (evaluate on/off).
  notifyToolsChanged() {
    const msg = `event: message\ndata: ${JSON.stringify({ jsonrpc: "2.0", method: "notifications/tools/list_changed" })}\n\n`;
    for (const s of this.sessions.values()) for (const r of s.streams) { try { r.write(msg); } catch {} }
  }

  status() {
    const now = Date.now();
    const live = [...this.sessions.values()].filter((s) => s.streams.size > 0 || now - s.lastSeen < IDLE_MS);
    return {
      listening: !!this.server,
      port: this.port,
      connected: live.length > 0,
      clients: [...new Set(live.map((s) => s.client))],
      sessions: live.map((s) => ({ client: s.client, since: new Date(s.created).toISOString(), lastSeen: new Date(s.lastSeen).toISOString(), streaming: s.streams.size > 0 })),
    };
  }

  _emitStatus() {
    const st = this.status();
    const key = JSON.stringify([st.listening, st.port, st.connected, st.clients]);
    if (key === this._lastStatus) return;
    this._lastStatus = key;
    this.onStatus(st);
  }

  _expire() {
    const now = Date.now();
    for (const [id, s] of this.sessions) {
      if (s.streams.size === 0 && now - s.lastSeen > SESSION_TTL_MS) this.sessions.delete(id);
    }
    this._emitStatus();
  }

  _json(res, status, body, headers = {}) {
    const data = body === undefined ? "" : JSON.stringify(body);
    res.writeHead(status, { "content-type": "application/json", "cache-control": "no-store", ...headers });
    res.end(data);
  }

  _refuse(res, status, message) {
    res.writeHead(status, { "content-type": "text/plain; charset=utf-8", "cache-control": "no-store", ...(status === 401 ? { "www-authenticate": 'Bearer realm="theseus"' } : {}) });
    res.end(message);
  }

  async _handle(req, res) {
    const ra = req.socket.remoteAddress || "";
    if (!/^(?:127\.|::1$|::ffff:127\.)/.test(ra)) return this._refuse(res, 403, "loopback only");
    const host = String(req.headers.host || "").toLowerCase();
    if (host !== `127.0.0.1:${this.port}` && host !== `localhost:${this.port}`) return this._refuse(res, 403, "bad host");
    // Browsers always send Origin on cross-origin requests; MCP clients do
    // not. A page in any browser on this computer gets nothing here.
    if (req.headers.origin) return this._refuse(res, 403, "browser requests are not accepted");
    const url = new URL(req.url, `http://${host}`);
    if (url.pathname !== "/mcp") return this._refuse(res, 404, "not found");
    const auth = String(req.headers.authorization || "");
    const m = /^Bearer\s+(\S+)$/i.exec(auth);
    if (!m || !tokenMatches(m[1], this.getToken())) return this._refuse(res, 401, "missing or wrong token");

    const sid = req.headers["mcp-session-id"] ? String(req.headers["mcp-session-id"]) : null;
    if (req.method === "DELETE") {
      const s = sid && this.sessions.get(sid);
      if (s) { for (const r of s.streams) { try { r.end(); } catch {} } this.sessions.delete(sid); this._emitStatus(); }
      return this._json(res, 200, undefined);
    }
    if (req.method === "GET") {
      if (!String(req.headers.accept || "").includes("text/event-stream")) return this._refuse(res, 406, "GET needs Accept: text/event-stream");
      const s = sid && this.sessions.get(sid);
      if (!s) return this._refuse(res, sid ? 404 : 400, sid ? "unknown session" : "missing Mcp-Session-Id");
      res.writeHead(200, { "content-type": "text/event-stream", "cache-control": "no-store", connection: "keep-alive" });
      res.write(": theseus\n\n");
      s.streams.add(res);
      s.lastSeen = Date.now();
      const ping = setInterval(() => { try { res.write(": ping\n\n"); } catch {} }, 20000);
      ping.unref?.();
      const done = () => { clearInterval(ping); s.streams.delete(res); s.lastSeen = Date.now(); this._emitStatus(); };
      req.on("close", done);
      res.on("close", done);
      this._emitStatus();
      return;
    }
    if (req.method !== "POST") return this._refuse(res, 405, "method not allowed");

    const body = await readBody(req);
    let msg;
    try { msg = JSON.parse(body); } catch { return this._json(res, 400, { jsonrpc: "2.0", id: null, error: { code: -32700, message: "parse error" } }); }
    const batch = Array.isArray(msg);
    const msgs = batch ? msg : [msg];
    if (!msgs.length) return this._json(res, 400, { jsonrpc: "2.0", id: null, error: { code: -32600, message: "empty batch" } });

    const init = msgs.find((x) => x && x.method === "initialize");
    let session = sid ? this.sessions.get(sid) : null;
    const headers = {};
    if (init) {
      const id = crypto.randomBytes(16).toString("hex");
      const ci = (init.params && init.params.clientInfo) || {};
      session = { id, client: String(ci.name || "MCP client").slice(0, 60), clientVersion: String(ci.version || ""), protocolVersion: null, created: Date.now(), lastSeen: Date.now(), streams: new Set() };
      this.sessions.set(id, session);
      headers["mcp-session-id"] = id;
      this.log(`client connected: ${session.client} ${session.clientVersion}`.trim());
    } else if (!session) {
      // Unknown id: the client re-initializes (spec: 404 means "start over").
      return this._json(res, sid ? 404 : 400, { jsonrpc: "2.0", id: null, error: { code: -32000, message: sid ? "unknown session; initialize again" : "missing Mcp-Session-Id; initialize first" } });
    }
    session.lastSeen = Date.now();
    this._emitStatus();

    const replies = [];
    for (const m1 of msgs) {
      if (!m1 || m1.jsonrpc !== "2.0" || typeof m1.method !== "string") {
        if (m1 && m1.id !== undefined && (m1.result !== undefined || m1.error !== undefined)) continue; // a client's reply to us
        replies.push({ jsonrpc: "2.0", id: m1 && m1.id !== undefined ? m1.id : null, error: { code: -32600, message: "invalid request" } });
        continue;
      }
      const isNotification = m1.id === undefined || m1.id === null;
      const r = await this._dispatch(m1, session);
      if (!isNotification) replies.push(r);
    }
    session.lastSeen = Date.now();
    if (!replies.length) { res.writeHead(202, headers); return res.end(); }
    return this._json(res, 200, batch ? replies : replies[0], headers);
  }

  async _dispatch(m, session) {
    const ok = (result) => ({ jsonrpc: "2.0", id: m.id, result });
    const err = (code, message) => ({ jsonrpc: "2.0", id: m.id, error: { code, message } });
    try {
      switch (m.method) {
        case "initialize": {
          const asked = m.params && m.params.protocolVersion;
          const v = PROTOCOL_VERSIONS.includes(asked) ? asked : PROTOCOL_VERSIONS[0];
          session.protocolVersion = v;
          return ok({ protocolVersion: v, capabilities: { tools: { listChanged: true } }, serverInfo: this.serverInfo, instructions: this.instructions });
        }
        case "notifications/initialized":
        case "notifications/cancelled":
        case "notifications/roots/list_changed":
          return null;
        case "ping":
          return ok({});
        case "tools/list":
          return ok({ tools: this.tools.list() });
        case "tools/call": {
          const name = m.params && m.params.name;
          const args = (m.params && m.params.arguments) || {};
          if (!this.tools.list().some((t) => t.name === name)) return err(-32602, `unknown tool: ${name}`);
          try {
            return ok(await this.tools.call(name, args, session));
          } catch (e) {
            return ok({ content: [{ type: "text", text: `Error: ${e?.message || e}` }], isError: true });
          }
        }
        default:
          if (m.method.startsWith("notifications/")) return null;
          return err(-32601, `method not found: ${m.method}`);
      }
    } catch (e) {
      return err(-32603, e?.message || String(e));
    }
  }
}

function readBody(req) {
  return new Promise((resolve, reject) => {
    const chunks = [];
    let n = 0;
    req.on("data", (c) => {
      n += c.length;
      if (n > MAX_BODY) { reject(new Error("request too large")); req.destroy(); return; }
      chunks.push(c);
    });
    req.on("end", () => resolve(Buffer.concat(chunks).toString("utf8")));
    req.on("error", reject);
  });
}

// First free port in the connector's range, else any free port.
async function pickPort(preferred) {
  const tryPort = (p) => new Promise((resolve) => {
    const s = http.createServer();
    s.once("error", () => resolve(null));
    s.listen(p, "127.0.0.1", () => { const got = s.address().port; s.close(() => resolve(got)); });
  });
  if (preferred) { const p = await tryPort(preferred); if (p) return p; }
  for (let p = 47800; p < 47900; p++) { const got = await tryPort(p); if (got) return got; }
  return tryPort(0);
}

module.exports = { McpServer, pickPort, tokenMatches, PROTOCOL_VERSIONS };