feat(broker): MCP client manager — connect, discover, callTool (M5a-1)

.changeset/broker-mcp-client-manager.md

@@ -0,0 +1,27 @@
+---
+"@boardstate/broker": minor
+---
+
+New package `@boardstate/broker` — the MCP CLIENT manager (M5a-1, epic #37). It connects
+outward to the external MCP servers an operator declares in a `boardstate.connectors.json`
+config, discovers their tools into a namespaced `ToolManifest`, and calls them behind a
+narrow API the host/server layers consume.
+
+- **Config is operator-authored only**: `loadConnectorsConfig(path)` + a programmatic
+  `new McpBroker(config)`. A connector name not in the config is inert. `env` values are
+  process-env var NAMES (references), never literal secrets — validated up front and never
+  echoed into errors or logs.
+- **Transports**: `StdioClientTransport` for local servers, `StreamableHTTPClientTransport`
+  with SSE fallback for remotes. Lazy connect on first use, warm/pooled clients, capped
+  exponential-backoff reconnect, clean close.
+- **`listTools()` → `ToolManifest`**: `connector:tool` ids AND provider-safe
+  `connector__tool` names (both inside a 64-char budget, collisions fail loud), input
+  schemas, `readOnlyHint` honored (absent ⇒ treated as a mutation — fail-safe), and a
+  stable manifest hash over sorted (id + canonical input-schema) pairs — the anti-rug-pull
+  snapshot the grant lifecycle (M5b-2) pins to.
+- **`callTool(id, args, { timeout })`**: resolves the client, strips the namespace,
+  enforces a hard timeout, and normalizes `isError: true` results into a typed
+  `BrokerToolError`.
+
+Depends only on `@modelcontextprotocol/sdk` and `@boardstate/server` (types only). Ships
+an in-repo fake MCP server fixture (stdio child + in-process HTTP) so CI needs no network.

packages/broker/package.json

@@ -0,0 +1,46 @@
+{
+  "name": "@boardstate/broker",
+  "version": "0.1.0",
+  "description": "The MCP client manager: connect outward to external MCP servers, discover their tools, and call them behind a narrow, provider-safe API",
+  "type": "module",
+  "license": "MIT",
+  "homepage": "https://github.com/100yenadmin/boardstate#readme",
+  "bugs": "https://github.com/100yenadmin/boardstate/issues",
+  "keywords": [
+    "mcp",
+    "model-context-protocol",
+    "client",
+    "broker",
+    "tools",
+    "boardstate"
+  ],
+  "exports": {
+    ".": {
+      "types": "./dist/index.d.ts",
+      "import": "./dist/index.js"
+    }
+  },
+  "bin": {
+    "boardstate-fake-mcp": "./dist/fixture/stdio-entry.js"
+  },
+  "files": [
+    "dist"
+  ],
+  "scripts": {
+    "build": "tsdown",
+    "test": "vitest run"
+  },
+  "dependencies": {
+    "@boardstate/server": "workspace:*",
+    "@modelcontextprotocol/sdk": "^1.29.0"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/100yenadmin/boardstate.git",
+    "directory": "packages/broker"
+  },
+  "publishConfig": {
+    "access": "public",
+    "provenance": true
+  }
+}

packages/broker/src/broker.test.ts

@@ -0,0 +1,183 @@
+// Broker behavior over an in-memory transport (fast, deterministic): config-only
+// refusal, namespaced + provider-name calling, isError normalization, hard timeout,
+// backoff on a flaky connect, and warm-client reconnect after a transport drop.
+
+import { afterEach, describe, expect, it } from "vitest";
+import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";
+import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js";
+import { McpBroker } from "./broker.js";
+import { parseConnectorsConfig } from "./config.js";
+import { BrokerTimeoutError, BrokerToolError, BrokerUnknownConnectorError } from "./errors.js";
+import { buildFakeMcpServer, type FakeCatalogState } from "./fixture/fake-mcp-server.js";
+
+/**
+ * An in-memory transport factory wired to a fresh fake server per connect. Exposes the
+ * connect count and the latest server-side transport so tests can force a drop.
+ */
+function inMemoryFactory(state: FakeCatalogState = { mutated: false }) {
+  let connects = 0;
+  let lastServerTransport: InMemoryTransport | undefined;
+  const factory = (): Transport => {
+    connects += 1;
+    const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair();
+    lastServerTransport = serverTransport;
+    const server = buildFakeMcpServer(state);
+    void server.connect(serverTransport);
+    return clientTransport;
+  };
+  return {
+    factory,
+    connects: () => connects,
+    dropLast: async () => {
+      await lastServerTransport?.close();
+      // Let the propagated onclose settle before the next call.
+      await new Promise((resolve) => setTimeout(resolve, 5));
+    },
+  };
+}
+
+function makeBroker(state?: FakeCatalogState) {
+  const wiring = inMemoryFactory(state);
+  const config = parseConnectorsConfig({
+    connectors: [{ name: "office", transport: "stdio", command: "unused-in-memory" }],
+  });
+  const broker = new McpBroker(config, {
+    transportFactory: wiring.factory,
+    initialBackoffMs: 1,
+    maxBackoffMs: 4,
+  });
+  return { broker, wiring };
+}
+
+describe("McpBroker", () => {
+  let open: McpBroker | null = null;
+  afterEach(async () => {
+    await open?.close();
+    open = null;
+  });
+
+  it("refuses a connector that is not in the operator config", async () => {
+    const { broker } = makeBroker();
+    open = broker;
+    await expect(broker.callTool("ghost:do_thing")).rejects.toBeInstanceOf(
+      BrokerUnknownConnectorError,
+    );
+  });
+
+  it("discovers a namespaced manifest with the fail-safe readOnly flag", async () => {
+    const { broker } = makeBroker();
+    open = broker;
+    const manifest = await broker.listTools();
+    const ids = manifest.tools.map((t) => t.id);
+    expect(ids).toContain("office:echo");
+    expect(ids).toContain("office:write_note");
+    expect(manifest.tools.find((t) => t.id === "office:echo")?.readOnly).toBe(true);
+    // write_note has no readOnlyHint → mutation.
+    expect(manifest.tools.find((t) => t.id === "office:write_note")?.readOnly).toBe(false);
+    expect(manifest.idToProvider.get("office:echo")).toBe("office__echo");
+  });
+
+  it("calls a tool by its manifest id and by its provider-safe name", async () => {
+    const { broker } = makeBroker();
+    open = broker;
+    const manifest = await broker.listTools();
+
+    const byId = await broker.callTool("office:add", { a: 2, b: 3 });
+    expect(byId.content).toEqual([{ type: "text", text: JSON.stringify({ sum: 5 }) }]);
+
+    const byProvider = await broker.callTool(
+      "office__add",
+      { a: 4, b: 1 },
+      { providerToId: manifest.providerToId },
+    );
+    expect(byProvider.content).toEqual([{ type: "text", text: JSON.stringify({ sum: 5 }) }]);
+  });
+
+  it("normalizes an isError result into a BrokerToolError carrying the server text", async () => {
+    const { broker } = makeBroker();
+    open = broker;
+    let thrown: unknown;
+    try {
+      await broker.callTool("office:boom");
+    } catch (error) {
+      thrown = error;
+    }
+    expect(thrown).toBeInstanceOf(BrokerToolError);
+    expect((thrown as BrokerToolError).toolId).toBe("office:boom");
+    expect((thrown as Error).message).toContain("boom");
+  });
+
+  it("enforces a hard timeout", async () => {
+    const { broker } = makeBroker();
+    open = broker;
+    await expect(
+      broker.callTool("office:sleep", { ms: 500 }, { timeout: 20 }),
+    ).rejects.toBeInstanceOf(BrokerTimeoutError);
+  });
+
+  it("retries a flaky connect with backoff", async () => {
+    let attempts = 0;
+    const config = parseConnectorsConfig({
+      connectors: [{ name: "office", transport: "stdio", command: "x" }],
+    });
+    // A transport whose start() rejects (a transient connect failure) for the first two
+    // attempts, then a live in-memory transport — the broker must back off and retry.
+    const broken = (): Transport => ({
+      async start() {
+        throw new Error("connect refused");
+      },
+      async send() {},
+      async close() {},
+    });
+    const broker = new McpBroker(config, {
+      initialBackoffMs: 1,
+      maxBackoffMs: 2,
+      transportFactory: () => {
+        attempts += 1;
+        if (attempts < 3) {
+          return broken();
+        }
+        const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair();
+        void buildFakeMcpServer().connect(serverTransport);
+        return clientTransport;
+      },
+    });
+    open = broker;
+    const manifest = await broker.listTools();
+    expect(manifest.tools.length).toBeGreaterThan(0);
+    expect(attempts).toBe(3);
+  });
+
+  it("reconnects a warm client after the transport drops", async () => {
+    const { broker, wiring } = makeBroker();
+    open = broker;
+    await broker.callTool("office:echo", { text: "hi" });
+    expect(wiring.connects()).toBe(1);
+
+    await wiring.dropLast();
+    // Next use must transparently reconnect (a second connect), not call a dead client.
+    const result = await broker.callTool("office:echo", { text: "again" });
+    expect(result.content).toEqual([{ type: "text", text: JSON.stringify({ text: "again" }) }]);
+    expect(wiring.connects()).toBe(2);
+  });
+
+  it("pools a warm client across calls (one connect for many calls)", async () => {
+    const { broker, wiring } = makeBroker();
+    open = broker;
+    await broker.callTool("office:echo", { text: "a" });
+    await broker.callTool("office:echo", { text: "b" });
+    await broker.listTools();
+    expect(wiring.connects()).toBe(1);
+  });
+
+  it("detects a rug-pull: the manifest hash changes when the catalog mutates", async () => {
+    const state: FakeCatalogState = { mutated: false };
+    const { broker } = makeBroker(state);
+    open = broker;
+    const before = await broker.listTools();
+    state.mutated = true;
+    const after = await broker.listTools();
+    expect(after.hash).not.toBe(before.hash);
+    expect(after.tools.map((t) => t.id)).toContain("office:extra");
+  });
+});