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

/**
 * The console's endpoint: MCP over HTTP, on a machine's loopback (novox/hq ADR 0152, design 34 §2).
 *
 * **Loopback is the authority boundary.** Whoever can connect is on the machine, and whoever is on the
 * machine is the account that owns the mesh there (ADR 0034, ADR 0144). So there is no token and no
 * login here, and the one thing this file enforces is that it binds nothing else: a console reachable
 * from another machine would be authority over the mesh handed to whoever finds the port.
 *
 * The transport is the streamable-HTTP shape an agent host speaks: `POST /mcp` with one JSON-RPC
 * message, answered with one JSON body. No session, because the surface holds nothing per caller; no
 * event stream, because nothing here has anything to say unasked.
 */
import { createServer, type IncomingMessage, type ServerResponse } from "node:http";

import type { Broker } from "@novox/mesh-sdk/messaging";

import { mcpSurface, type Reply, type Request } from "./mcp.js";

/** The most a request body may be. A tool's arguments are small; a megabyte is somebody else's file. */
const BODY_LIMIT = 1 << 20;

export interface Listening {
  /** Where it listens, as `host:port`, with the port the machine actually gave. */
  address: string;
  close(): Promise<void>;
}

/** Hosts that are this machine and no other. */
const loopback = new Set(["127.0.0.1", "::1", "localhost", "[::1]"]);

/**
 * Listen on `host:port`. Refused unless the host is loopback — said before binding, so a manifest or
 * a flag that would open the console to a network is a startup failure rather than something
 * discovered by whoever finds it.
 */
export async function serveMcpHttp(bus: Broker, who: string, listen: string): Promise<Listening> {
  const at = listen.lastIndexOf(":");
  if (at < 0) {
    throw new Error(`"${listen}" is not host:port`);
  }
  const host = listen.slice(0, at);
  const port = Number(listen.slice(at + 1));
  if (!loopback.has(host)) {
    throw new Error(
      `the console listens on loopback and nowhere else (novox/hq ADR 0152): "${host}" is not this ` +
        "machine's own address — whoever is on the machine owns the mesh there, and nobody else may reach this",
    );
  }
  if (!Number.isInteger(port) || port < 0 || port > 65535) {
    throw new Error(`"${listen.slice(at + 1)}" is not a port`);
  }

  const surface = mcpSurface(bus, who);
  const server = createServer((req, res) => {
    void route(req, res, surface.handle).catch((e) => {
      json(res, 500, { jsonrpc: "2.0", id: null, error: { code: -32603, message: String(e) } });
    });
  });

  await new Promise<void>((resolve, reject) => {
    server.once("error", reject);
    server.listen(port, host.replace(/^\[|\]$/g, ""), () => resolve());
  });
  const bound = server.address();
  const address = typeof bound === "object" && bound ? `${host}:${bound.port}` : listen;
  return {
    address,
    close: () =>
      new Promise<void>((resolve) => {
        server.close(() => resolve());
      }),
  };
}

async function route(
  req: IncomingMessage,
  res: ServerResponse,
  handle: (r: Request) => Promise<Reply | undefined>,
): Promise<void> {
  const path = (req.url ?? "/").split("?")[0];
  if (path === "/") {
    res.writeHead(200, { "content-type": "text/plain; charset=utf-8" });
    res.end("the mesh's console: MCP over HTTP at POST /mcp (novox/hq design 34)\n");
    return;
  }
  if (path !== "/mcp") {
    json(res, 404, { error: "the console serves /mcp and nothing else" });
    return;
  }
  switch (req.method) {
    case "POST":
      break;
    case "DELETE":
      // A host ending a session. There is no session to end; saying so is the truthful answer.
      res.writeHead(204).end();
      return;
    case "GET":
      // A host opening an event stream. The console has nothing to say unasked.
      res.writeHead(405, { allow: "POST, DELETE" }).end();
      return;
    default:
      res.writeHead(405, { allow: "POST, DELETE" }).end();
      return;
  }

  let body: string;
  try {
    body = await read(req);
  } catch (e) {
    json(res, 413, { jsonrpc: "2.0", id: null, error: { code: -32600, message: String(e) } });
    return;
  }
  let parsed: unknown;
  try {
    parsed = JSON.parse(body);
  } catch {
    json(res, 400, { jsonrpc: "2.0", id: null, error: { code: -32700, message: "the body is not JSON" } });
    return;
  }

  // One message, or a batch of them; a batch is answered as a batch. A notification gets no reply
  // and, alone, no body: 202 is how the transport says "heard".
  if (Array.isArray(parsed)) {
    const replies = (await Promise.all(parsed.map((r) => handle(r as Request)))).filter(Boolean);
    if (replies.length === 0) {
      res.writeHead(202).end();
    } else {
      json(res, 200, replies);
    }
    return;
  }
  const reply = await handle(parsed as Request);
  if (!reply) {
    res.writeHead(202).end();
    return;
  }
  json(res, 200, reply);
}

function read(req: IncomingMessage): Promise<string> {
  return new Promise((resolve, reject) => {
    let size = 0;
    const chunks: Buffer[] = [];
    req.on("data", (chunk: Buffer) => {
      size += chunk.length;
      if (size > BODY_LIMIT) {
        reject(new Error(`the request is larger than ${BODY_LIMIT} bytes`));
        req.destroy();
        return;
      }
      chunks.push(chunk);
    });
    req.on("end", () => resolve(Buffer.concat(chunks).toString("utf8")));
    req.on("error", reject);
  });
}

function json(res: ServerResponse, status: number, body: unknown): void {
  const text = JSON.stringify(body);
  res.writeHead(status, {
    "content-type": "application/json; charset=utf-8",
    "content-length": Buffer.byteLength(text),
  });
  res.end(text);
}