neuron-connectd/dist/bridge.js

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import { UnauthorizedError } from "@modelcontextprotocol/sdk/client/auth.js";
import { createHash } from "node:crypto";
import { saveConfig } from "./config.js";
import { KeychainOAuthProvider, takePendingAuthUrl } from "./oauth.js";
import { kcGet, kcSet, kcDelete } from "./keychain.js";
// Namespacing: every external tool is exposed to the soul as
//   mcp__<serverId>__<toolName>
// so the soul can route a call back to the owning server by prefix alone.
const PREFIX = "mcp__";
const SEP = "__";
export function namespacedName(serverId, tool) {
    return `${PREFIX}${serverId}${SEP}${tool}`;
}
// Split "mcp__filesystem__read_file" -> { serverId:"filesystem", tool:"read_file" }.
// Tool names may themselves contain "__", so we only split on the FIRST two seps.
export function parseName(full) {
    if (!full.startsWith(PREFIX))
        return null;
    const rest = full.slice(PREFIX.length);
    const i = rest.indexOf(SEP);
    if (i < 0)
        return null;
    return { serverId: rest.slice(0, i), tool: rest.slice(i + SEP.length) };
}
export class ConnectorBridge {
    servers = new Map();
    // The live config the bridge owns and persists. UI edits mutate this and call saveConfig.
    config = { servers: {} };
    async start(config) {
        this.config = config;
        const ids = Object.keys(config.servers);
        await Promise.all(ids.map((id) => this.connectServer(id, config.servers[id])));
    }
    // ── Config CRUD (Phase 4 UI backs onto these via the soul's /api/connectors routes) ──
    // Add or replace a server, persist, and connect it. Returns its fresh status.
    async addServer(id, cfg) {
        if (!id || !/^[a-zA-Z0-9_-]+$/.test(id))
            return { ok: false, error: "invalid server id" };
        this.config.servers[id] = cfg;
        saveConfig(this.config);
        await this.reconnectServer(id);
        const s = this.servers.get(id);
        return s && s.status === "error" ? { ok: false, error: s.error } : { ok: true };
    }
    async setEnabled(id, enabled) {
        const cfg = this.config.servers[id];
        if (!cfg)
            return { ok: false, error: `unknown server: ${id}` };
        cfg.enabled = enabled;
        // Re-enabling acknowledges any flagged schema change (user is opting back in).
        if (enabled)
            delete cfg.schemaHash;
        saveConfig(this.config);
        await this.reconnectServer(id);
        return { ok: true };
    }
    async setAutoApprove(id, autoApprove) {
        const cfg = this.config.servers[id];
        if (!cfg)
            return { ok: false, error: `unknown server: ${id}` };
        cfg.autoApprove = autoApprove;
        saveConfig(this.config);
        const entry = this.servers.get(id);
        if (entry)
            entry.cfg = cfg;
        return { ok: true };
    }
    // Store a connector's API token in the Keychain (account = serverId), then reconnect.
    // Used for http auth:"token" servers; the secret never lands in connectors.json.
    async setSecret(id, secret) {
        if (!secret)
            return { ok: false, error: "empty secret" };
        const stored = await kcSet(id, secret);
        if (!stored)
            return { ok: false, error: "keychain write failed" };
        if (this.config.servers[id])
            await this.reconnectServer(id);
        return { ok: true };
    }
    async removeServer(id) {
        if (!this.config.servers[id])
            return { ok: false, error: `unknown server: ${id}` };
        await this.closeServer(id);
        delete this.config.servers[id];
        this.servers.delete(id);
        saveConfig(this.config);
        await kcDelete(id).catch(() => { }); // drop any stored secret for this connector
        return { ok: true };
    }
    // Tear down an existing client (if any) and connect fresh from current config.
    async reconnectServer(id) {
        await this.closeServer(id);
        const cfg = this.config.servers[id];
        if (cfg)
            await this.connectServer(id, cfg);
    }
    async closeServer(id) {
        const entry = this.servers.get(id);
        if (entry?.client) {
            try {
                await entry.client.close();
            }
            catch { /* best effort */ }
        }
    }
    async connectServer(id, cfg) {
        if (!cfg.enabled) {
            this.servers.set(id, { id, cfg, status: "disabled", tools: [] });
            return;
        }
        if (cfg.transport === "stdio")
            return this.connectStdio(id, cfg);
        if (cfg.transport === "http")
            return this.connectHttp(id, cfg);
        this.servers.set(id, { id, cfg, status: "error", tools: [], error: `unknown transport "${cfg.transport}"` });
    }
    async connectStdio(id, cfg) {
        try {
            const transport = new StdioClientTransport({
                command: cfg.command,
                args: cfg.args ?? [],
                env: { ...process.env, ...(cfg.env ?? {}) },
            });
            await this.finishConnect(id, cfg, transport);
        }
        catch (err) {
            this.setError(id, cfg, err);
        }
    }
    // Remote MCP over Streamable HTTP. Two auth modes:
    //   auth: "token"  → static bearer from Keychain account <serverId>
    //   auth: "oauth"  → SDK-driven PKCE; needs_auth until the user signs in
    async connectHttp(id, cfg) {
        if (!cfg.url) {
            this.servers.set(id, { id, cfg, status: "error", tools: [], error: "http transport requires a url" });
            return;
        }
        const url = new URL(cfg.url);
        if (cfg.auth === "oauth") {
            const provider = new KeychainOAuthProvider(id, cfg.scope);
            const transport = new StreamableHTTPClientTransport(url, { authProvider: provider });
            try {
                await this.finishConnect(id, cfg, transport);
            }
            catch (err) {
                if (err instanceof UnauthorizedError) {
                    // Provider stashed the auth URL; keep the transport so /oauth/callback
                    // can finishAuth on it. Surface needs_auth to the UI.
                    this.servers.set(id, {
                        id, cfg, status: "needs_auth", tools: [],
                        pendingTransport: transport, pendingProvider: provider,
                        error: "sign-in required",
                    });
                    console.error(`[connectd] ${id}: needs OAuth sign-in`);
                }
                else {
                    this.setError(id, cfg, err);
                }
            }
            return;
        }
        // token (or none)
        let requestInit;
        if (cfg.auth === "token") {
            const token = await kcGet(id);
            if (!token) {
                this.servers.set(id, { id, cfg, status: "needs_auth", tools: [], error: "no token in Keychain" });
                return;
            }
            requestInit = { headers: { Authorization: `Bearer ${token}` } };
        }
        try {
            const transport = new StreamableHTTPClientTransport(url, { requestInit });
            await this.finishConnect(id, cfg, transport);
        }
        catch (err) {
            this.setError(id, cfg, err);
        }
    }
    // Connect a client over the given transport, list tools, record as connected.
    async finishConnect(id, cfg, transport) {
        const client = new Client({ name: "neuron-connectd", version: "0.1.0" }, { capabilities: {} });
        await client.connect(transport);
        const listed = await client.listTools();
        const tools = listed.tools.map((t) => ({
            name: namespacedName(id, t.name),
            description: t.description ?? "",
            input_schema: t.inputSchema ?? { type: "object", properties: {} },
        }));
        // Tool-poisoning guard: hash name+description+schema of every tool. First connect pins the
        // hash; a later mismatch flags the server in the UI (a silently-changed tool description is a
        // prompt-injection vector) until the user re-enables it.
        const hash = schemaHashOf(tools);
        let schemaChanged = false;
        if (cfg.schemaHash && cfg.schemaHash !== hash) {
            schemaChanged = true;
        }
        else if (!cfg.schemaHash) {
            cfg.schemaHash = hash;
            this.config.servers[id] = cfg;
            saveConfig(this.config);
        }
        this.servers.set(id, { id, cfg, client, status: "connected", tools, schemaChanged });
        console.error(`[connectd] ${id}: connected (${cfg.transport}), ${tools.length} tools${schemaChanged ? " [SCHEMA CHANGED]" : ""}`);
    }
    setError(id, cfg, err) {
        const msg = err instanceof Error ? err.message : String(err);
        this.servers.set(id, { id, cfg, status: "error", tools: [], error: msg });
        console.error(`[connectd] ${id}: connect failed — ${msg}`);
    }
    // Begin an OAuth sign-in: returns the authorization URL for the UI to open.
    async startOAuth(id) {
        const entry = this.servers.get(id);
        if (!entry)
            return { ok: false, error: `unknown server: ${id}` };
        if (entry.cfg.transport !== "http" || entry.cfg.auth !== "oauth") {
            return { ok: false, error: `server ${id} is not an OAuth connector` };
        }
        // Re-drive the connect to (re)generate the auth URL if not already pending.
        if (entry.status !== "needs_auth" || !entry.pendingTransport) {
            await this.connectHttp(id, entry.cfg);
        }
        const url = takePendingAuthUrl(id);
        if (!url)
            return { ok: false, error: "no authorization URL produced (server may not require OAuth)" };
        return { ok: true, authUrl: url };
    }
    // Complete an OAuth sign-in with the authorization code from the callback.
    async finishOAuth(id, code) {
        const entry = this.servers.get(id);
        if (!entry?.pendingTransport)
            return { ok: false, error: `no pending OAuth flow for ${id}` };
        try {
            await entry.pendingTransport.finishAuth(code);
            // Tokens are now in Keychain; reconnect cleanly to list tools.
            await this.connectHttp(id, entry.cfg);
            const now = this.servers.get(id);
            return now?.status === "connected"
                ? { ok: true }
                : { ok: false, error: `post-auth status: ${now?.status} (${now?.error ?? ""})` };
        }
        catch (err) {
            return { ok: false, error: err instanceof Error ? err.message : String(err) };
        }
    }
    // Merged, namespaced tool list across all connected servers.
    allTools() {
        const out = [];
        for (const s of this.servers.values())
            out.push(...s.tools);
        return out;
    }
    // Per-server status for the UI / health endpoint.
    serverStatuses() {
        return [...this.servers.values()].map((s) => ({
            id: s.id,
            status: s.status,
            transport: s.cfg.transport,
            tools: s.tools.length,
            enabled: s.cfg.enabled,
            auth: s.cfg.auth,
            autoApprove: s.cfg.autoApprove === true,
            schemaChanged: s.schemaChanged === true,
            error: s.error,
        }));
    }
    // The set of connector tool names that are auto-approved (the soul reads this to decide
    // whether an mcp__* call skips the approval card). Only connected, enabled, opted-in servers.
    autoApprovedTools() {
        const out = [];
        for (const s of this.servers.values()) {
            if (s.cfg.autoApprove && s.status === "connected")
                out.push(...s.tools.map((t) => t.name));
        }
        return out;
    }
    // Route a namespaced call to the owning server and flatten the result to a string
    // (the soul does json_get(out,"content") then truncates at 6000 chars).
    async call(full, input) {
        const parsed = parseName(full);
        if (!parsed)
            return { ok: false, error: `not an mcp tool name: ${full}` };
        const entry = this.servers.get(parsed.serverId);
        if (!entry || !entry.client) {
            const why = entry?.status === "needs_auth" ? "needs_auth" : `server not connected: ${parsed.serverId}`;
            return { ok: false, error: why };
        }
        try {
            const res = await entry.client.callTool({
                name: parsed.tool,
                arguments: (input ?? {}),
            });
            return { ok: true, content: flattenContent(res.content) };
        }
        catch (err) {
            return { ok: false, error: err instanceof Error ? err.message : String(err) };
        }
    }
}
// Stable SHA-256 over each tool's name + description + input schema. Sorted by name so tool
// ordering doesn't churn the hash; only real content changes do.
function schemaHashOf(tools) {
    const canon = [...tools]
        .sort((a, b) => a.name.localeCompare(b.name))
        .map((t) => `${t.name}${t.description}${JSON.stringify(t.input_schema)}`)
        .join("
");
    return createHash("sha256").update(canon).digest("hex");
}
// MCP returns content as an array of typed blocks. Join text blocks; fall back to
// a compact JSON dump for non-text blocks so nothing is silently dropped.
function flattenContent(content) {
    if (!Array.isArray(content))
        return typeof content === "string" ? content : JSON.stringify(content);
    const parts = [];
    for (const block of content) {
        if (block && typeof block === "object" && block.type === "text") {
            parts.push(String(block.text ?? ""));
        }
        else {
            parts.push(JSON.stringify(block));
        }
    }
    return parts.join("\n");
}