mesh-tools/node-tools/src/mcp.ts

/**
 * The mesh's tools as an MCP server (novox/hq design 25 §7, design 34).
 *
 * **A thin adapter and nothing more.** Every tool an agent sees is one a module answered for and one
 * this account may call; the schema is the module's own; the answer is the module's own. Nothing here
 * decides anything, which is why it is short — an MCP surface that reshaped arguments or summarised
 * answers would be a second definition of what a tool is, and the module's code is the first.
 *
 * Implemented against the protocol directly rather than through a library: the surface is three
 * methods and one framing, and a dependency here would be a dependency on every machine.
 *
 * One handler, two transports. Over stdio for a program a person starts (`mesh mcp`), over HTTP on a
 * machine's loopback for the console the mesh assigns there (`mesh serve`, http.ts). The handler does
 * not know which asked.
 */
import type { Broker } from "@novox/mesh-sdk/messaging";

import { callTool, seatsIn, toolKey, toolsOn, whyItFailed, type Listing, type Seats } from "./client.js";

/** The protocol version this speaks. Stated, because a host that wants another should be told so
 *  rather than discovering it through a shape it did not expect. */
export const PROTOCOL = "2025-03-26";

export interface Request {
  jsonrpc: string;
  id?: number | string | null;
  method: string;
  params?: Record<string, unknown>;
}

export interface Reply {
  jsonrpc: "2.0";
  id: Request["id"];
  result?: unknown;
  error?: { code: number; message: string };
}

/** How long a fetched tool list is kept before the modules are asked again. An agent asks on every
 *  turn; the mesh changes on the order of minutes. */
export const LISTING_KEPT_MS = 30_000;

export interface Surface {
  /** Answer one request; undefined for a notification, which expects none. */
  handle(request: Request): Promise<Reply | undefined>;
}

/**
 * The surface over one bus connection, as one account.
 *
 * The tool list is fetched when first asked and kept for a short while (design 34 §3): asking every
 * module on every `tools/list` would fan out on every agent turn for something nobody changed, and
 * never refreshing would hide a module assigned a moment ago.
 */
export function mcpSurface(bus: Broker, who: string): Surface {
  let known: { listing: Listing; at: number } | undefined;

  const answer = (id: Request["id"], result: unknown): Reply => ({ jsonrpc: "2.0", id, result });
  const refuse = (id: Request["id"], code: number, message: string): Reply => ({
    jsonrpc: "2.0",
    id,
    error: { code, message },
  });

  const listing = async (): Promise<Listing> => {
    if (!known || Date.now() - known.at > LISTING_KEPT_MS) {
      known = { listing: await toolsOn(bus), at: Date.now() };
    }
    return known.listing;
  };
  // The roles the last listing knew, so `<seat>.<verb>` resolves to the seat. Fetched once if a
  // call arrives before any list did; a listing that failed leaves no roles, and the name is then
  // a module's, which is the right fallback for a mesh whose control plane is away.
  const roles = async (): Promise<Seats | undefined> => {
    try {
      return seatsIn(await listing());
    } catch {
      return undefined;
    }
  };

  return {
    async handle(request) {
      // A notification has no id and expects no answer; `initialized` is the one every host sends.
      const notification = request.id === undefined || request.id === null;

      switch (request.method) {
        case "initialize":
          return answer(request.id, {
            protocolVersion: PROTOCOL,
            capabilities: { tools: {} },
            serverInfo: { name: "mesh", version: "1" },
            // Said in the handshake, because an agent that knows whose authority it is acting under
            // can say so when a call is refused — and a refusal is the one thing here that is not
            // the mesh's fault or the tool's.
            instructions:
              `These are the tools of a Novox mesh, reached as ${who}. Every call goes to the module ` +
              `that serves it; what may be called was fixed when this account was issued, so a ` +
              `refusal means the account, not the tool. The list is what the running modules ` +
              `answered, plus every role's tools from the mesh's records — the mesh's own verbs ` +
              `(mesh-controller.status, .push, .assign …) among them; a module that did not answer ` +
              `is named in the list's _meta and can still be called by <module>.<tool>.`,
          });

        case "notifications/initialized":
          return undefined;

        case "ping":
          return notification ? undefined : answer(request.id, {});

        case "tools/list": {
          let have: Listing;
          try {
            have = await listing();
          } catch (e) {
            return refuse(request.id, -32603, whyItFailed("mesh-catalog.catalog_modules", e));
          }
          return answer(request.id, {
            tools: have.tools.map((t) => ({
              name: `${t.module}.${t.name}`,
              description: t.description ?? `${t.name}, served by ${t.module}`,
              // The module's own schema, passed through — with `node`, the machine to ask when
              // the module runs on several (novox/hq ADR 0159); a seat's verb takes none, the
              // seat's scope decides. An empty object is a tool that takes nothing, which is a
              // real answer and not a missing one.
              inputSchema: t.seat && t.scope !== "node" ? asSchema(t.input)
                : t.seat ? withNode(asSchema(t.input), "the machine whose seat answers; required, the seat is held once per machine", true)
                : withNode(asSchema(t.input)),
            })),
            // Silence, named (design 34 §3): the modules the catalogue holds and nothing answered
            // for. Not a tool, so not in `tools`; not dropped either.
            _meta: { notAnswering: have.notAnswering },
          });
        }

        case "tools/call": {
          const given = String(request.params?.name ?? "");
          const args = { ...((request.params?.arguments as Record<string, unknown> | undefined) ?? {}) };
          // The machine, when the caller names one, travels in the subject and never reaches the
          // module's arguments (novox/hq ADR 0159) — for a module's tool. A seat's verb takes no
          // machine from the console (the seat's scope decides), so a `node` among its arguments
          // is the verb's own, as `push` and `assign` take one, and is handed through untouched.
          const have = await listing().catch(() => undefined);
          const roles = have && seatsIn(have);
          const bare = given.split("@", 1)[0];
          const isSeatVerb = roles ? toolKey(bare, roles).startsWith("seat:") : false;
          // A node-scoped seat's verb is asked of one machine (design 33 §4, ADR 0170): `node`
          // names it and travels in the subject, as for a module's tool.
          const nodeScoped = isSeatVerb && (have?.tools.some((t) => t.seat && t.scope === "node" &&
            `${t.module}.${t.name}` === bare) ?? false);
          const takesNode = !isSeatVerb || nodeScoped;
          const node = takesNode && typeof args.node === "string" && args.node !== "" ? args.node : "";
          if (takesNode) delete args.node;
          if (nodeScoped && !node && !given.includes("@")) {
            return refuse(request.id, -32602, `${given} is a machine's seat's verb: name the machine with \`node\``);
          }
          const name = node && !given.includes("@") ? `${given}@${node}` : given;
          try {
            const { result, node: answeredBy } = await callTool(bus, name, args, roles, have);
            // Text, because that is what every host renders. The content is the module's answer
            // as JSON, unshaped: an adapter that flattened it would be deciding what matters in
            // somebody else's answer. Which machine answered follows it as its own line.
            const content: { type: string; text: string }[] = [
              { type: "text", text: JSON.stringify(result, null, 2) },
            ];
            if (answeredBy) content.push({ type: "text", text: `answered by ${answeredBy}` });
            return answer(request.id, { content });
          } catch (e) {
            // **An error the agent can act on, not a stack.** isError rather than a protocol
            // failure, because the call was well-formed and the mesh answered it — with a refusal,
            // an absence or a fault, and the words say which.
            return answer(request.id, {
              content: [{ type: "text", text: whyItFailed(name, e) }],
              isError: true,
            });
          }
        }

        default:
          return notification ? undefined : refuse(request.id, -32601, `mesh's MCP surface has no ${request.method}`);
      }
    },
  };
}

/**
 * A module's declared input as a JSON schema an agent can read.
 *
 * The sdk keeps a tool's input opaque, and the catalogue's modules write it as a bare map of
 * property to description — `{ module: { type, description } }` — which is the `properties` of a
 * schema rather than a schema. Wrapped here when that is what arrived; passed through when a module
 * already wrote a schema; an empty object when it declared nothing. The module's words are kept
 * either way.
 */
export function asSchema(input: unknown): Record<string, unknown> {
  if (!input || typeof input !== "object" || Array.isArray(input)) {
    return { type: "object", properties: {} };
  }
  const given = input as Record<string, unknown>;
  if (given.type === "object" || "properties" in given) return given;
  if (Object.keys(given).length === 0) return { type: "object", properties: {} };
  return { type: "object", properties: given };
}

/**
 * Serve over stdio until stdin closes, which is how a host ends a session.
 */
export async function serveMcp(bus: Broker, who: string): Promise<void> {
  const surface = mcpSurface(bus, who);
  const say = (message: unknown) => {
    process.stdout.write(`${JSON.stringify(message)}\n`);
  };
  for await (const line of lines()) {
    let request: Request;
    try {
      request = JSON.parse(line) as Request;
    } catch {
      // Unparseable, and with no id there is nobody to tell. Skipped rather than answered, because a
      // reply to a request that was never framed is noise on the same channel.
      continue;
    }
    const reply = await surface.handle(request);
    if (reply) say(reply);
  }
}

/** stdin as newline-framed messages, which is what MCP over stdio is. */
async function* lines(): AsyncGenerator<string> {
  let buffered = "";
  for await (const chunk of process.stdin) {
    buffered += (chunk as Buffer).toString("utf8");
    let at: number;
    while ((at = buffered.indexOf("\n")) >= 0) {
      const line = buffered.slice(0, at).trim();
      buffered = buffered.slice(at + 1);
      if (line !== "") yield line;
    }
  }
  if (buffered.trim() !== "") yield buffered.trim();
}

/** Every module tool takes an optional `node`: the machine to ask when the module runs on several
 *  (novox/hq ADR 0159). Added to the listing, stripped before the call, never seen by the module. */
function withNode(schema: Record<string, unknown>, description?: string, required = false): Record<string, unknown> {
  const properties = { ...((schema.properties as Record<string, unknown> | undefined) ?? {}) };
  if (!("node" in properties)) {
    properties.node = {
      type: "string",
      description: description ??
        "the machine to ask, when this module runs on several; else whichever answers, and the answer says which",
    };
  }
  const out: Record<string, unknown> = { ...schema, type: "object", properties };
  if (required) {
    const have = Array.isArray(schema.required) ? (schema.required as string[]) : [];
    out.required = have.includes("node") ? have : [...have, "node"];
  }
  return out;
}