joakimp
37cc49e06f
- New RemoteMcpClient implementing MCP streamable-HTTP per spec 2025-03-26:
POST JSON-RPC, parse application/json or text/event-stream responses,
round-trip optional Mcp-Session-Id, optional auth via 'headers' config.
- Refactor StdioMcpClient to share an IMcpClient interface with the remote
client; extension entry dispatches on cfg.type. Drops the v1 'remote
skipped with warning' code path.
- Bump MCP_PROTOCOL_VERSION to 2025-11-25 (single constant, both clients).
- 404 self-heal: when a remote returns 404 to a request carrying our
Mcp-Session-Id, drop the id, re-initialize, retry the request once
(per spec 2025-11-25 \u00a72.2). allowReinitOn404=false on the retry path
prevents recursion. Verified via mock-server smoke test.
- Sanitize pi-facing tool names to ^[A-Za-z][A-Za-z0-9_]{0,63}$. Anthropic
allows hyphens but Bedrock's Anthropic shim rejects them, causing entire
turns to 4xx silently when context7's hyphenated tools (resolve-library-id,
query-docs) were registered. Original MCP-side names are preserved in the
tool-execute closure, so sanitization is purely pi-facing.
- /mcp slash command: drop 'remote (skipped)' status label.
- Docs: README and AGENTS updated for transports, headers config, 404
self-heal, tool-name sanitization rationale, OAuth limitation.
End-to-end verified: context7 connects through pi, returns useful docs
(Bun streaming/SSE example fetched successfully).
836 lines
30 KiB
TypeScript
836 lines
30 KiB
TypeScript
/**
|
|
* MCP Loader Extension
|
|
*
|
|
* Reads an `mcp` block from pi's settings.json (same shape as opencode and
|
|
* Claude Desktop) and connects to each declared server, exposing all of
|
|
* their tools to pi as native tools.
|
|
*
|
|
* Settings.json shape:
|
|
*
|
|
* {
|
|
* "mcp": {
|
|
* "searxng": {
|
|
* "type": "local",
|
|
* "command": ["uvx", "mcp-searxng"],
|
|
* "env": { "SEARXNG_URL": "https://searxng.your-host.lan" }
|
|
* },
|
|
* "gitea": {
|
|
* "type": "local",
|
|
* "command": ["gitea-mcp", "-t", "stdio"],
|
|
* "enabled": false,
|
|
* "env": { "GITEA_ACCESS_TOKEN": "...", "GITEA_HOST": "..." }
|
|
* },
|
|
* "context7": {
|
|
* "type": "remote",
|
|
* "url": "https://mcp.context7.com/mcp",
|
|
* "headers": { "Authorization": "Bearer …" }
|
|
* }
|
|
* }
|
|
* }
|
|
*
|
|
* Per-server keys:
|
|
* type "local" (stdio subprocess) | "remote" (streamable-http)
|
|
* command argv array — first element is the executable, rest are args.
|
|
* For local servers only.
|
|
* url remote MCP endpoint URL. For remote servers only.
|
|
* headers optional object of HTTP headers to send with every request to
|
|
* a remote server (e.g. Authorization, X-API-Key). Remote only.
|
|
* enabled default true. Set false to disable a server without
|
|
* removing it from settings.json.
|
|
* env optional object of environment variables to inject into
|
|
* the spawned subprocess. Inherits parent env first, then
|
|
* overlays these keys. Local only.
|
|
*
|
|
* Tool naming
|
|
*
|
|
* Each MCP tool is registered with pi as `<server-name>_<tool-name>` to
|
|
* avoid collisions across servers. If the tool name already begins with
|
|
* `<server-name>_` (e.g. mempalace's tools are `mempalace_search` etc.)
|
|
* the prefix is not duplicated.
|
|
*
|
|
* Lifecycle
|
|
*
|
|
* • Servers are connected at extension load time (pi startup or /reload).
|
|
* • Each server's `tools/list` is awaited before pi finishes registering.
|
|
* • Local subprocess stderr is silenced unless PI_MCP_LOADER_DEBUG=1.
|
|
* • Local subprocesses receive SIGTERM at session_shutdown; remote
|
|
* servers receive an MCP DELETE if a session id was issued.
|
|
* • A server that fails to start (binary missing, init handshake error,
|
|
* remote 4xx/5xx) logs a single line to stderr and is skipped. Other
|
|
* servers continue. Pi keeps working without that server's tools.
|
|
*
|
|
* Transports
|
|
*
|
|
* local — stdio JSON-RPC subprocess (mcp-searxng, gitea-mcp, …).
|
|
* remote — streamable-HTTP per MCP spec 2025-03-26: POST JSON-RPC, server
|
|
* replies either application/json or text/event-stream. Optional
|
|
* Mcp-Session-Id header round-tripped if the server issues one.
|
|
* No GET-stream subscription (we don't consume server-initiated
|
|
* notifications).
|
|
*
|
|
* Limitations
|
|
*
|
|
* • No stdio reconnect: if a stdio subprocess dies mid-session, its tools
|
|
* become unavailable until pi is reloaded. The mempalace.ts extension
|
|
* has the same limitation today.
|
|
* • Remote sessions self-heal on 404: if a streamable-HTTP server
|
|
* forgets our session id (e.g. server restart), the next request gets
|
|
* HTTP 404; the client transparently re-initializes and retries the
|
|
* request once. Persistent 404s after refresh surface as errors.
|
|
* • No OAuth flow: remote servers requiring OAuth must be accessed with
|
|
* a pre-issued bearer token via `headers`.
|
|
*
|
|
* Coexistence with mempalace.ts
|
|
*
|
|
* The mempalace.ts extension already spawns mempalace-mcp directly with
|
|
* bespoke handling (agent identity injection from $MEMPALACE_AGENT_NAME,
|
|
* wake-up context). This loader does NOT touch mempalace — leave the
|
|
* bespoke extension in place. If you also list mempalace under the `mcp`
|
|
* block in settings.json the loader will register a parallel set of
|
|
* tools, which is harmless but redundant. Don't.
|
|
*
|
|
* Slash command
|
|
*
|
|
* `/mcp` lists configured MCP servers with their runtime status
|
|
* (running / failed / disabled / invalid) and lets
|
|
* you toggle the `enabled` flag in settings.json. Same UX as `/ext`:
|
|
* space stages, enter applies + reloads, esc cancels.
|
|
*/
|
|
|
|
import { type ChildProcessWithoutNullStreams, spawn } from "node:child_process";
|
|
import * as fs from "node:fs";
|
|
import * as os from "node:os";
|
|
import * as path from "node:path";
|
|
import {
|
|
type ExtensionAPI,
|
|
getSettingsListTheme,
|
|
} from "@mariozechner/pi-coding-agent";
|
|
import {
|
|
Container,
|
|
Key,
|
|
matchesKey,
|
|
type SettingItem,
|
|
SettingsList,
|
|
} from "@mariozechner/pi-tui";
|
|
import { Type } from "typebox";
|
|
|
|
// ── MCP types ────────────────────────────────────────────────────────────────
|
|
|
|
type McpTool = {
|
|
name: string;
|
|
description?: string;
|
|
inputSchema?: Record<string, unknown>;
|
|
};
|
|
|
|
type LocalServerConfig = {
|
|
type?: "local";
|
|
command: string[];
|
|
env?: Record<string, string>;
|
|
enabled?: boolean;
|
|
};
|
|
|
|
type RemoteServerConfig = {
|
|
type: "remote";
|
|
url: string;
|
|
headers?: Record<string, string>;
|
|
enabled?: boolean;
|
|
};
|
|
|
|
type ServerConfig = LocalServerConfig | RemoteServerConfig;
|
|
|
|
const SETTINGS_PATH = path.join(os.homedir(), ".pi", "agent", "settings.json");
|
|
const DEBUG = process.env.PI_MCP_LOADER_DEBUG === "1";
|
|
|
|
// MCP protocol version we advertise. Servers we've tested:
|
|
// - searxng (mcp-searxng) accepts both 2024-11-05 and 2025-11-25
|
|
// - context7 accepts both
|
|
// - mempalace bridge: not touched by this loader
|
|
// Bump as the spec evolves; both stdio and remote clients use this.
|
|
const MCP_PROTOCOL_VERSION = "2025-11-25";
|
|
|
|
function dlog(msg: string) {
|
|
if (DEBUG) process.stderr.write(`[mcp-loader] ${msg}\n`);
|
|
}
|
|
function warn(msg: string) {
|
|
process.stderr.write(`[mcp-loader] ${msg}\n`);
|
|
}
|
|
|
|
// ── Settings.json reader ─────────────────────────────────────────────────────
|
|
|
|
function readMcpServers(): Record<string, ServerConfig> {
|
|
if (!fs.existsSync(SETTINGS_PATH)) {
|
|
dlog(`no settings.json at ${SETTINGS_PATH}`);
|
|
return {};
|
|
}
|
|
let raw: string;
|
|
try {
|
|
raw = fs.readFileSync(SETTINGS_PATH, "utf8");
|
|
} catch (err) {
|
|
warn(`could not read ${SETTINGS_PATH}: ${(err as Error).message}`);
|
|
return {};
|
|
}
|
|
let parsed: any;
|
|
try {
|
|
parsed = JSON.parse(raw);
|
|
} catch (err) {
|
|
warn(`settings.json parse error: ${(err as Error).message}`);
|
|
return {};
|
|
}
|
|
const block = parsed?.mcp;
|
|
if (!block || typeof block !== "object") {
|
|
dlog("no `mcp` block in settings.json");
|
|
return {};
|
|
}
|
|
return block as Record<string, ServerConfig>;
|
|
}
|
|
|
|
// ── Stdio MCP client ─────────────────────────────────────────────────────────
|
|
|
|
interface IMcpClient {
|
|
readonly serverName: string;
|
|
tools: McpTool[];
|
|
start(): Promise<void>;
|
|
callTool(name: string, args: Record<string, unknown>): Promise<any>;
|
|
stop(): void | Promise<void>;
|
|
}
|
|
|
|
class StdioMcpClient implements IMcpClient {
|
|
private proc: ChildProcessWithoutNullStreams | null = null;
|
|
private nextId = 1;
|
|
private pending = new Map<number, { resolve: (v: any) => void; reject: (e: Error) => void }>();
|
|
private stdoutBuf = "";
|
|
public readonly serverName: string;
|
|
private command: string;
|
|
private args: string[];
|
|
private env: Record<string, string> | undefined;
|
|
public tools: McpTool[] = [];
|
|
|
|
constructor(serverName: string, command: string, args: string[], env?: Record<string, string>) {
|
|
this.serverName = serverName;
|
|
this.command = command;
|
|
this.args = args;
|
|
this.env = env;
|
|
}
|
|
|
|
async start(): Promise<void> {
|
|
const childEnv = { ...process.env, ...(this.env ?? {}) };
|
|
this.proc = spawn(this.command, this.args, { stdio: ["pipe", "pipe", "pipe"], env: childEnv });
|
|
|
|
this.proc.on("error", (err) => this.failAll(err));
|
|
this.proc.on("exit", (code, signal) => {
|
|
this.failAll(new Error(`${this.serverName}: subprocess exited (code=${code}, signal=${signal})`));
|
|
});
|
|
|
|
this.proc.stdout.setEncoding("utf8");
|
|
this.proc.stdout.on("data", (chunk: string) => this.onStdout(chunk));
|
|
|
|
this.proc.stderr.setEncoding("utf8");
|
|
if (DEBUG) {
|
|
this.proc.stderr.on("data", (chunk: string) => {
|
|
process.stderr.write(`[${this.serverName} stderr] ${chunk}`);
|
|
});
|
|
} else {
|
|
this.proc.stderr.resume(); // drain without logging
|
|
}
|
|
|
|
// MCP initialize handshake
|
|
await this.request("initialize", {
|
|
protocolVersion: MCP_PROTOCOL_VERSION,
|
|
capabilities: {},
|
|
clientInfo: { name: "pi-mcp-loader", version: "0.2.0" },
|
|
});
|
|
this.notify("notifications/initialized", {});
|
|
|
|
const listed = await this.request("tools/list", {});
|
|
this.tools = (listed?.tools as McpTool[]) ?? [];
|
|
}
|
|
|
|
private onStdout(chunk: string) {
|
|
this.stdoutBuf += chunk;
|
|
let nl: number;
|
|
while ((nl = this.stdoutBuf.indexOf("\n")) !== -1) {
|
|
const line = this.stdoutBuf.slice(0, nl).trim();
|
|
this.stdoutBuf = this.stdoutBuf.slice(nl + 1);
|
|
if (!line) continue;
|
|
let msg: any;
|
|
try {
|
|
msg = JSON.parse(line);
|
|
} catch {
|
|
continue;
|
|
}
|
|
if (typeof msg.id === "number" && this.pending.has(msg.id)) {
|
|
const { resolve, reject } = this.pending.get(msg.id)!;
|
|
this.pending.delete(msg.id);
|
|
if (msg.error) reject(new Error(msg.error.message ?? "MCP error"));
|
|
else resolve(msg.result);
|
|
}
|
|
// notifications (no id) ignored.
|
|
}
|
|
}
|
|
|
|
private failAll(err: Error) {
|
|
for (const { reject } of this.pending.values()) reject(err);
|
|
this.pending.clear();
|
|
}
|
|
|
|
private write(obj: unknown) {
|
|
if (!this.proc) throw new Error(`${this.serverName}: not started`);
|
|
this.proc.stdin.write(`${JSON.stringify(obj)}\n`);
|
|
}
|
|
|
|
private notify(method: string, params: unknown) {
|
|
this.write({ jsonrpc: "2.0", method, params });
|
|
}
|
|
|
|
request(method: string, params: unknown): Promise<any> {
|
|
const id = this.nextId++;
|
|
return new Promise((resolve, reject) => {
|
|
this.pending.set(id, { resolve, reject });
|
|
this.write({ jsonrpc: "2.0", id, method, params });
|
|
});
|
|
}
|
|
|
|
async callTool(name: string, args: Record<string, unknown>): Promise<any> {
|
|
return this.request("tools/call", { name, arguments: args });
|
|
}
|
|
|
|
stop() {
|
|
if (this.proc) {
|
|
try {
|
|
this.proc.kill("SIGTERM");
|
|
} catch {}
|
|
this.proc = null;
|
|
}
|
|
}
|
|
}
|
|
|
|
// ── Streamable-HTTP MCP client (remote) ──────────────────────────────────────
|
|
//
|
|
// Per MCP spec 2025-03-26: a single endpoint URL accepts POST with a
|
|
// JSON-RPC body. The server responds either with application/json (a
|
|
// single response) or text/event-stream (an SSE stream containing the
|
|
// response, possibly preceded by notifications, terminated by stream
|
|
// close). We do NOT open a separate GET subscription stream — we don't
|
|
// consume server-initiated notifications today.
|
|
//
|
|
// Session id: if the server returns an Mcp-Session-Id header on
|
|
// initialize, we round-trip it on every subsequent request and DELETE
|
|
// it on stop().
|
|
|
|
class RemoteMcpClient implements IMcpClient {
|
|
public readonly serverName: string;
|
|
private url: string;
|
|
private extraHeaders: Record<string, string>;
|
|
private sessionId: string | null = null;
|
|
private nextId = 1;
|
|
public tools: McpTool[] = [];
|
|
|
|
constructor(serverName: string, url: string, headers?: Record<string, string>) {
|
|
this.serverName = serverName;
|
|
this.url = url;
|
|
this.extraHeaders = headers ?? {};
|
|
}
|
|
|
|
async start(): Promise<void> {
|
|
await this.request("initialize", {
|
|
protocolVersion: MCP_PROTOCOL_VERSION,
|
|
capabilities: {},
|
|
clientInfo: { name: "pi-mcp-loader", version: "0.2.0" },
|
|
});
|
|
await this.notify("notifications/initialized", {});
|
|
const listed = await this.request("tools/list", {});
|
|
this.tools = (listed?.tools as McpTool[]) ?? [];
|
|
}
|
|
|
|
async callTool(name: string, args: Record<string, unknown>): Promise<any> {
|
|
return this.request("tools/call", { name, arguments: args });
|
|
}
|
|
|
|
async stop(): Promise<void> {
|
|
if (!this.sessionId) return;
|
|
try {
|
|
await fetch(this.url, { method: "DELETE", headers: this.buildHeaders() });
|
|
} catch {}
|
|
}
|
|
|
|
private buildHeaders(): Record<string, string> {
|
|
const h: Record<string, string> = {
|
|
"Content-Type": "application/json",
|
|
Accept: "application/json, text/event-stream",
|
|
"MCP-Protocol-Version": MCP_PROTOCOL_VERSION,
|
|
...this.extraHeaders,
|
|
};
|
|
if (this.sessionId) h["Mcp-Session-Id"] = this.sessionId;
|
|
return h;
|
|
}
|
|
|
|
/**
|
|
* Re-establish a session after the server returned 404 to a request that
|
|
* carried our `Mcp-Session-Id`. Per spec 2025-11-25, the client MUST start
|
|
* a new session by sending a fresh InitializeRequest without a session id.
|
|
*
|
|
* Caller must clear `this.sessionId` BEFORE invoking this so the new
|
|
* initialize POST goes out without the stale id.
|
|
*/
|
|
private async reinitialize(): Promise<void> {
|
|
dlog(`${this.serverName}: session lost (404) — reinitializing`);
|
|
await this.request("initialize", {
|
|
protocolVersion: MCP_PROTOCOL_VERSION,
|
|
capabilities: {},
|
|
clientInfo: { name: "pi-mcp-loader", version: "0.2.0" },
|
|
}, /*allowReinitOn404=*/ false);
|
|
await this.notify("notifications/initialized", {});
|
|
}
|
|
|
|
private async request(method: string, params: unknown, allowReinitOn404 = true): Promise<any> {
|
|
const id = this.nextId++;
|
|
const body = JSON.stringify({ jsonrpc: "2.0", id, method, params });
|
|
const sessionAtStart = this.sessionId;
|
|
let res: Response;
|
|
try {
|
|
res = await fetch(this.url, { method: "POST", headers: this.buildHeaders(), body });
|
|
} catch (err) {
|
|
throw new Error(`${this.serverName}: fetch failed on ${method}: ${(err as Error).message}`);
|
|
}
|
|
|
|
const sid = res.headers.get("Mcp-Session-Id") ?? res.headers.get("mcp-session-id");
|
|
if (sid) this.sessionId = sid;
|
|
|
|
// 404 + we sent a session id → server forgot us. Drop our id, re-init,
|
|
// retry once. `allowReinitOn404=false` is passed by the recovery path to
|
|
// prevent infinite recursion if the server is permanently broken.
|
|
if (res.status === 404 && sessionAtStart && allowReinitOn404) {
|
|
try { await res.arrayBuffer(); } catch {}
|
|
this.sessionId = null;
|
|
await this.reinitialize();
|
|
return this.request(method, params, /*allowReinitOn404=*/ false);
|
|
}
|
|
|
|
if (!res.ok) {
|
|
let detail = "";
|
|
try { detail = (await res.text()).slice(0, 200); } catch {}
|
|
throw new Error(`${this.serverName}: HTTP ${res.status} on ${method}${detail ? `: ${detail}` : ""}`);
|
|
}
|
|
|
|
const ct = (res.headers.get("content-type") ?? "").toLowerCase();
|
|
if (ct.includes("text/event-stream")) {
|
|
return await this.readSseForResponse(res, id);
|
|
}
|
|
if (ct.includes("application/json")) {
|
|
const json: any = await res.json();
|
|
if (json.error) throw new Error(json.error.message ?? "MCP error");
|
|
return json.result;
|
|
}
|
|
if (res.status === 202) return undefined;
|
|
throw new Error(`${this.serverName}: unexpected content-type "${ct}" on ${method}`);
|
|
}
|
|
|
|
private async notify(method: string, params: unknown): Promise<void> {
|
|
const body = JSON.stringify({ jsonrpc: "2.0", method, params });
|
|
let res: Response;
|
|
try {
|
|
res = await fetch(this.url, { method: "POST", headers: this.buildHeaders(), body });
|
|
} catch (err) {
|
|
throw new Error(`${this.serverName}: fetch failed on notify ${method}: ${(err as Error).message}`);
|
|
}
|
|
try { await res.arrayBuffer(); } catch {}
|
|
if (!res.ok && res.status !== 202) {
|
|
throw new Error(`${this.serverName}: HTTP ${res.status} on notify ${method}`);
|
|
}
|
|
}
|
|
|
|
private async readSseForResponse(res: Response, expectedId: number): Promise<any> {
|
|
if (!res.body) throw new Error(`${this.serverName}: SSE response had no body`);
|
|
const reader = res.body.getReader();
|
|
const decoder = new TextDecoder();
|
|
let buf = "";
|
|
let dataLines: string[] = [];
|
|
|
|
const tryDispatch = (): { matched: boolean; result?: any } => {
|
|
if (dataLines.length === 0) return { matched: false };
|
|
const data = dataLines.join("\n");
|
|
dataLines = [];
|
|
let msg: any;
|
|
try {
|
|
msg = JSON.parse(data);
|
|
} catch {
|
|
return { matched: false };
|
|
}
|
|
if (typeof msg.id === "number" && msg.id === expectedId) {
|
|
if (msg.error) throw new Error(msg.error.message ?? "MCP error");
|
|
return { matched: true, result: msg.result };
|
|
}
|
|
return { matched: false };
|
|
};
|
|
|
|
try {
|
|
while (true) {
|
|
const { done, value } = await reader.read();
|
|
if (done) break;
|
|
buf += decoder.decode(value, { stream: true });
|
|
let nl: number;
|
|
while ((nl = buf.indexOf("\n")) !== -1) {
|
|
const rawLine = buf.slice(0, nl);
|
|
buf = buf.slice(nl + 1);
|
|
const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine;
|
|
if (line === "") {
|
|
const out = tryDispatch();
|
|
if (out.matched) return out.result;
|
|
} else if (line.startsWith(":")) {
|
|
// SSE comment — ignore (often a keepalive)
|
|
} else if (line.startsWith("data:")) {
|
|
dataLines.push(line.slice(5).replace(/^ /, ""));
|
|
}
|
|
// event:, id:, retry: — ignored; only `data` matters for our use
|
|
}
|
|
}
|
|
const out = tryDispatch();
|
|
if (out.matched) return out.result;
|
|
} finally {
|
|
try { reader.cancel(); } catch {}
|
|
}
|
|
throw new Error(`${this.serverName}: SSE stream closed without response for id=${expectedId}`);
|
|
}
|
|
}
|
|
|
|
// ── MCP result → pi tool result helpers ──────────────────────────────────────
|
|
|
|
function extractText(mcpResult: any): string {
|
|
if (!mcpResult) return "";
|
|
const content = mcpResult.content;
|
|
if (!Array.isArray(content)) return JSON.stringify(mcpResult);
|
|
return content
|
|
.map((c: any) => {
|
|
if (c?.type === "text" && typeof c.text === "string") return c.text;
|
|
return JSON.stringify(c);
|
|
})
|
|
.join("\n");
|
|
}
|
|
|
|
function namespacedToolName(serverName: string, toolName: string): string {
|
|
// Avoid double-prefix: if the MCP server already names its tools
|
|
// `<serverName>_<...>`, leave them alone.
|
|
const prefix = `${serverName}_`;
|
|
const raw = toolName.startsWith(prefix) ? toolName : `${prefix}${toolName}`;
|
|
// Sanitize for the strictest tool-name regex we have to satisfy.
|
|
// Anthropic's Messages API allows `^[a-zA-Z0-9_-]{1,64}$` (hyphens OK),
|
|
// but AWS Bedrock's Anthropic shim rejects names containing `-` outright
|
|
// — it accepts `^[a-zA-Z][a-zA-Z0-9_]{0,63}$`. Hyphenated MCP tool names
|
|
// (context7's `resolve-library-id`, `query-docs`) caused entire turns to
|
|
// 4xx silently on Bedrock, manifesting as no output at all once the
|
|
// server was enabled. Replace any non-[A-Za-z0-9_] with `_`, then prefix
|
|
// a leading underscore with `t` if the result starts with a non-letter.
|
|
// Truncate to 64 to also satisfy Anthropic's length cap.
|
|
let sanitized = raw.replace(/[^A-Za-z0-9_]/g, "_");
|
|
if (!/^[A-Za-z]/.test(sanitized)) sanitized = `t_${sanitized}`;
|
|
if (sanitized.length > 64) sanitized = sanitized.slice(0, 64);
|
|
return sanitized;
|
|
}
|
|
|
|
// ── Settings.json writer ─────────────────────────────────────────────────────
|
|
|
|
/**
|
|
* Flip the `enabled` flag for a server in settings.json and write back.
|
|
* Preserves all other keys verbatim by round-tripping through JSON with
|
|
* 2-space indent (matches the file's existing style).
|
|
*/
|
|
function setServerEnabled(serverName: string, enabled: boolean): void {
|
|
const raw = fs.readFileSync(SETTINGS_PATH, "utf8");
|
|
const parsed = JSON.parse(raw);
|
|
if (!parsed.mcp || !parsed.mcp[serverName]) {
|
|
throw new Error(`server '${serverName}' not in settings.json`);
|
|
}
|
|
if (enabled) {
|
|
// Default is true — clear the explicit flag if present, keep the file tidy.
|
|
delete parsed.mcp[serverName].enabled;
|
|
} else {
|
|
parsed.mcp[serverName].enabled = false;
|
|
}
|
|
fs.writeFileSync(SETTINGS_PATH, `${JSON.stringify(parsed, null, 2)}\n`, "utf8");
|
|
}
|
|
|
|
// ── Per-server runtime state (for /mcp display) ──────────────────────────────
|
|
|
|
type ServerRuntimeState =
|
|
| { kind: "running"; toolCount: number }
|
|
| { kind: "failed"; message: string }
|
|
| { kind: "disabled" }
|
|
| { kind: "invalid"; message: string };
|
|
|
|
// ── Extension entry ──────────────────────────────────────────────────────────
|
|
|
|
export default async function mcpLoaderExtension(pi: ExtensionAPI) {
|
|
const servers = readMcpServers();
|
|
const runtime = new Map<string, ServerRuntimeState>();
|
|
const clients: IMcpClient[] = [];
|
|
|
|
// Always register /mcp — even when no servers are configured, so users
|
|
// discover the surface. Handler reads settings.json fresh on each invoke.
|
|
registerMcpCommand(pi, runtime);
|
|
|
|
if (Object.keys(servers).length === 0) {
|
|
dlog("no MCP servers configured — nothing to load");
|
|
return;
|
|
}
|
|
|
|
for (const [name, cfg] of Object.entries(servers)) {
|
|
if (cfg.enabled === false) {
|
|
dlog(`${name}: disabled in settings.json, skipping`);
|
|
runtime.set(name, { kind: "disabled" });
|
|
continue;
|
|
}
|
|
|
|
let client: IMcpClient;
|
|
if (cfg.type === "remote") {
|
|
const remote = cfg as RemoteServerConfig;
|
|
if (typeof remote.url !== "string" || remote.url.length === 0) {
|
|
warn(`${name}: remote server missing \`url\`, skipping`);
|
|
runtime.set(name, { kind: "invalid", message: "remote server missing url" });
|
|
continue;
|
|
}
|
|
client = new RemoteMcpClient(name, remote.url, remote.headers);
|
|
} else {
|
|
const local = cfg as LocalServerConfig;
|
|
if (!Array.isArray(local.command) || local.command.length === 0) {
|
|
warn(`${name}: invalid \`command\` (must be non-empty array), skipping`);
|
|
runtime.set(name, { kind: "invalid", message: "command must be non-empty array" });
|
|
continue;
|
|
}
|
|
const [bin, ...args] = local.command;
|
|
client = new StdioMcpClient(name, bin, args, local.env);
|
|
}
|
|
|
|
try {
|
|
await client.start();
|
|
} catch (err) {
|
|
const message = (err as Error).message;
|
|
warn(`${name}: failed to start — ${message}`);
|
|
runtime.set(name, { kind: "failed", message });
|
|
try { await client.stop(); } catch {}
|
|
continue;
|
|
}
|
|
|
|
dlog(`${name}: connected, ${client.tools.length} tools`);
|
|
runtime.set(name, { kind: "running", toolCount: client.tools.length });
|
|
clients.push(client);
|
|
|
|
// Register each MCP tool as a pi tool.
|
|
for (const tool of client.tools) {
|
|
const piName = namespacedToolName(name, tool.name);
|
|
const schema =
|
|
tool.inputSchema && typeof tool.inputSchema === "object"
|
|
? (Type.Unsafe<Record<string, unknown>>(
|
|
tool.inputSchema as object,
|
|
) as unknown as ReturnType<typeof Type.Object>)
|
|
: Type.Object({}, { additionalProperties: true });
|
|
|
|
pi.registerTool({
|
|
name: piName,
|
|
label: piName,
|
|
description: tool.description ?? `MCP tool from ${name}`,
|
|
parameters: schema,
|
|
async execute(_toolCallId, params) {
|
|
try {
|
|
const result = await client.callTool(tool.name, params as Record<string, unknown>);
|
|
const text = extractText(result);
|
|
return {
|
|
content: [{ type: "text", text }],
|
|
details: { server: name, tool: tool.name, raw: result },
|
|
};
|
|
} catch (err) {
|
|
return {
|
|
content: [
|
|
{
|
|
type: "text",
|
|
text: `MCP call failed (${name}/${tool.name}): ${(err as Error).message}`,
|
|
},
|
|
],
|
|
details: { server: name, tool: tool.name, error: (err as Error).message },
|
|
isError: true,
|
|
};
|
|
}
|
|
},
|
|
});
|
|
}
|
|
}
|
|
|
|
// Tear down clients (stdio subprocesses, remote sessions) on session shutdown.
|
|
pi.on("session_shutdown", async () => {
|
|
await Promise.allSettled(clients.map((c) => Promise.resolve(c.stop())));
|
|
});
|
|
}
|
|
|
|
// ── /mcp slash command ──────────────────────────────────────────────────────
|
|
|
|
const VAL_ON = "● enabled";
|
|
const VAL_OFF = "○ disabled";
|
|
const VAL_RO = "[read-only]";
|
|
|
|
function statusLabel(state: ServerRuntimeState | undefined, configEnabled: boolean): string {
|
|
if (!state) return configEnabled ? "unknown" : "disabled";
|
|
switch (state.kind) {
|
|
case "running":
|
|
return `running · ${state.toolCount} tool${state.toolCount === 1 ? "" : "s"}`;
|
|
case "failed":
|
|
return `failed: ${state.message}`;
|
|
case "disabled":
|
|
return "disabled in settings";
|
|
case "invalid":
|
|
return `invalid: ${state.message}`;
|
|
}
|
|
}
|
|
|
|
function registerMcpCommand(pi: ExtensionAPI, runtime: Map<string, ServerRuntimeState>) {
|
|
pi.registerCommand("mcp", {
|
|
description: "List and toggle MCP servers configured in ~/.pi/agent/settings.json",
|
|
handler: async (_args, ctx) => {
|
|
const servers = readMcpServers();
|
|
const names = Object.keys(servers);
|
|
if (names.length === 0) {
|
|
ctx.ui.notify(
|
|
`No MCP servers configured. Add an \`mcp\` block to ${SETTINGS_PATH}.`,
|
|
"info",
|
|
);
|
|
return;
|
|
}
|
|
|
|
// Snapshot config-level enabled state (default true if unset).
|
|
const configEnabled = new Map<string, boolean>();
|
|
for (const [n, cfg] of Object.entries(servers)) {
|
|
configEnabled.set(n, cfg.enabled !== false);
|
|
}
|
|
|
|
// Staged state: name → enabled (initialised from configEnabled).
|
|
const staged = new Map(configEnabled);
|
|
|
|
await ctx.ui.custom<void>((tui, theme, _kb, done) => {
|
|
let statusText = "";
|
|
|
|
const items: SettingItem[] = names.map((n) => {
|
|
const cfg = servers[n];
|
|
const cfgEn = configEnabled.get(n) ?? true;
|
|
const state = runtime.get(n);
|
|
const isRemote = cfg.type === "remote";
|
|
const isInvalid = state?.kind === "invalid";
|
|
const transport = isRemote ? "remote" : "local";
|
|
const desc = `${transport} · ${statusLabel(state, cfgEn)}`;
|
|
|
|
if (isInvalid) {
|
|
return {
|
|
id: n,
|
|
label: n,
|
|
currentValue: VAL_RO,
|
|
description: desc,
|
|
};
|
|
}
|
|
return {
|
|
id: n,
|
|
label: n,
|
|
currentValue: cfgEn ? VAL_ON : VAL_OFF,
|
|
values: [VAL_ON, VAL_OFF],
|
|
description: desc,
|
|
};
|
|
});
|
|
|
|
const container = new Container();
|
|
|
|
container.addChild({
|
|
render(_w: number) {
|
|
return [
|
|
theme.fg("accent", theme.bold("MCP servers — settings.json `mcp` block")),
|
|
theme.fg(
|
|
"muted",
|
|
" space: toggle (stage) · enter: apply + reload · esc: cancel",
|
|
),
|
|
"",
|
|
];
|
|
},
|
|
invalidate() {},
|
|
});
|
|
|
|
const settingsList = new SettingsList(
|
|
items,
|
|
Math.min(items.length + 2, 20),
|
|
getSettingsListTheme(),
|
|
(id, newValue) => {
|
|
const isInvalid = runtime.get(id)?.kind === "invalid";
|
|
if (isInvalid) {
|
|
settingsList.updateValue(id, VAL_RO);
|
|
statusText = `⊘ ${id}: invalid config — fix in settings.json`;
|
|
tui.requestRender();
|
|
return;
|
|
}
|
|
const stagingEnabled = newValue === VAL_ON;
|
|
staged.set(id, stagingEnabled);
|
|
|
|
const drift: string[] = [];
|
|
for (const n of names) {
|
|
const s = staged.get(n);
|
|
const c = configEnabled.get(n);
|
|
if (s !== undefined && s !== c) drift.push(`${n}→${s ? "on" : "off"}`);
|
|
}
|
|
statusText = drift.length ? `pending: ${drift.join(", ")}` : "";
|
|
tui.requestRender();
|
|
},
|
|
() => {
|
|
done(undefined);
|
|
},
|
|
);
|
|
|
|
container.addChild(settingsList);
|
|
|
|
container.addChild({
|
|
render(_w: number) {
|
|
if (!statusText) return [""];
|
|
const colored = statusText.startsWith("⊘")
|
|
? theme.fg("warning", statusText)
|
|
: theme.fg("muted", statusText);
|
|
return ["", colored];
|
|
},
|
|
invalidate() {},
|
|
});
|
|
|
|
return {
|
|
render(width: number) {
|
|
return container.render(width);
|
|
},
|
|
invalidate() {
|
|
container.invalidate();
|
|
},
|
|
handleInput(data: string) {
|
|
if (matchesKey(data, Key.enter)) {
|
|
const errors: string[] = [];
|
|
let applied = 0;
|
|
for (const n of names) {
|
|
const s = staged.get(n);
|
|
const c = configEnabled.get(n);
|
|
if (s === undefined || s === c) continue;
|
|
try {
|
|
setServerEnabled(n, s);
|
|
applied++;
|
|
} catch (err) {
|
|
const msg = err instanceof Error ? err.message : String(err);
|
|
errors.push(`${n}: ${msg}`);
|
|
}
|
|
}
|
|
done(undefined);
|
|
if (errors.length) {
|
|
ctx.ui.notify(`mcp: ${errors.join(" | ")}`, "error");
|
|
} else if (applied > 0) {
|
|
ctx.ui.notify(`mcp: applied ${applied} change(s); reloading…`, "info");
|
|
ctx.reload().catch((err) => {
|
|
const msg = err instanceof Error ? err.message : String(err);
|
|
ctx.ui.notify(`reload failed: ${msg}`, "error");
|
|
});
|
|
}
|
|
return;
|
|
}
|
|
settingsList.handleInput?.(data);
|
|
tui.requestRender();
|
|
},
|
|
};
|
|
});
|
|
},
|
|
});
|
|
}
|