feat(mcp): Streamable HTTP MCP server on the capability registry

POST/GET/DELETE /api/v1/mcp with API Key auth; tools are the capabilities visible to the key's scope (download__submit naming), every call goes through registry.invoke (authorize, audit, idempotency key, ledger) with source=mcp. Resources: settings summary, pending decisions, subscriptions overview; prompts: triage_pending, acquire_title, diagnose_task. Sessions bound to the API key, idle reaping, per-key rate limits, enable switch and key allowlist. Also GET /capabilities/stats (from audit_event) and /capabilities/tools. Tests connect with the official SDK client over Streamable HTTP. Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-09-28 20:28:34 +08:00
1 parent 73ee3c6614
commit 653100ceb4
@@ -0,0 +1,192 @@
+import type { CapabilityRegistry } from "@umt/agent";
+import {
+  CAPABILITY_CATALOG,
+  type CapabilityAuthState,
+  type CapabilityStatsItem,
+  type CapabilityStatsResponse,
+  type CapabilityToolInfo,
+  type CapabilityToolsResponse,
+  callerOfSource,
+  type KnownCapabilityId,
+  type McpRecentCall,
+  mcpToolVisible,
+} from "@umt/contracts";
+import { type DB, type Kysely, sql } from "@umt/db";
+import { z } from "zod";
+import { GRANT_ACTIONS } from "../agent/authorize.ts";
+import type { SettingsReader } from "../types.ts";
+
+// 能力调用统计与能力清单。统计口径是 audit_event(registry.invoke 每次调用都写一条,含只读能力与拒绝);
+// capability_invocation 只记有副作用的调用,不适合做全量统计。
+
+const AUDIT_ACTIONS = ["capability.invoke", "capability.denied"];
+
+export async function capabilityStats(
+  db: Kysely<DB>,
+  days: number,
+  now = new Date(),
+): Promise<CapabilityStatsResponse> {
+  const since = new Date(now.getTime() - days * 86_400_000);
+  const agg = await sql<{
+    id: string;
+    calls: number;
+    failures: number;
+    denied: number;
+    avg_ms: string | null;
+    last_at: Date;
+    by_agent: number;
+    by_mcp: number;
+    by_rest: number;
+    by_other: number;
+  }>`
+    select target_id as id,
+           count(*)::int as calls,
+           count(*) filter (where payload->>'outcome' in ('error', 'invalid'))::int as failures,
+           count(*) filter (where payload->>'outcome' = 'denied')::int as denied,
+           avg((payload->>'durationMs')::numeric) filter (where payload->>'outcome' <> 'denied') as avg_ms,
+           max(at) as last_at,
+           count(*) filter (where payload->>'source' in ('agent', 'task'))::int as by_agent,
+           count(*) filter (where payload->>'source' = 'mcp')::int as by_mcp,
+           count(*) filter (where payload->>'source' in ('web', 'cli'))::int as by_rest,
+           count(*) filter (where coalesce(payload->>'source', '') not in ('agent', 'task', 'mcp', 'web', 'cli'))::int as by_other
+      from audit_event
+     where target_type = 'capability' and action in (${sql.join(AUDIT_ACTIONS)}) and at >= ${since}
+     group by target_id`.execute(db);
+  const last = await sql<{ id: string; outcome: string | null; source: string | null }>`
+    select distinct on (target_id) target_id as id, payload->>'outcome' as outcome, payload->>'source' as source
+      from audit_event
+     where target_type = 'capability' and action in (${sql.join(AUDIT_ACTIONS)}) and at >= ${since}
+     order by target_id, at desc`.execute(db);
+  const lastBy = new Map(last.rows.map((r) => [r.id, r]));
+  const items: CapabilityStatsItem[] = agg.rows
+    .map((r) => {
+      const l = lastBy.get(r.id);
+      return {
+        capabilityId: r.id,
+        calls: r.calls,
+        failures: r.failures,
+        denied: r.denied,
+        avgDurationMs: r.avg_ms === null ? null : Math.round(Number(r.avg_ms)),
+        lastCalledAt: new Date(r.last_at).toISOString(),
+        lastOutcome: l?.outcome ?? null,
+        lastCaller: l ? callerOfSource(l.source) : null,
+        byCaller: { agent: r.by_agent, mcp: r.by_mcp, rest: r.by_rest, other: r.by_other },
+      };
+    })
+    .sort((a, b) => b.calls - a.calls || a.capabilityId.localeCompare(b.capabilityId));
+  return { days, since: since.toISOString(), items };
+}
+
+/** 最近的 MCP 来源调用 */
+export async function recentMcpCalls(db: Kysely<DB>, limit = 20): Promise<McpRecentCall[]> {
+  const rows = await db
+    .selectFrom("audit_event")
+    .select(["at", "target_id", "actor_id", "payload"])
+    .where("target_type", "=", "capability")
+    .where("action", "in", AUDIT_ACTIONS)
+    .where(sql<boolean>`payload->>'source' = 'mcp'`)
+    .orderBy("at", "desc")
+    .limit(limit)
+    .execute();
+  return rows.map((r) => {
+    const p = (r.payload ?? {}) as { outcome?: string; durationMs?: number; errorCode?: string | null };
+    return {
+      at: r.at.toISOString(),
+      capabilityId: r.target_id ?? "",
+      outcome: p.outcome ?? "unknown",
+      actorId: r.actor_id,
+      durationMs: typeof p.durationMs === "number" ? p.durationMs : null,
+      errorCode: p.errorCode ?? null,
+    };
+  });
+}
+
+interface GrantRow {
+  action: string;
+  scope: unknown;
+  policy_revision_id: string | null;
+  policy_status: string | null;
+}
+
+async function activeGrants(db: Kysely<DB>, now: Date): Promise<GrantRow[]> {
+  return db
+    .selectFrom("automation_grant as g")
+    .leftJoin("policy_revision as p", "p.id", "g.policy_revision_id")
+    .select(["g.action", "g.scope", "g.policy_revision_id", "p.status as policy_status"])
+    .where("g.status", "=", "active")
+    .where((eb) => eb.or([eb("g.expires_at", "is", null), eb("g.expires_at", ">", now)]))
+    .execute();
+}
+
+/** 覆盖这个能力的有效持续授权数(与 authorize.ts 同口径:动作覆盖、策略版本仍发布、能力白名单包含) */
+export function grantsCovering(id: string, grants: readonly GrantRow[]): number {
+  const actions = new Set([...(GRANT_ACTIONS[id] ?? [id]), "*"]);
+  return grants.filter((g) => {
+    if (!actions.has(g.action)) return false;
+    if (g.policy_revision_id && g.policy_status !== "published") return false;
+    const caps = ((g.scope ?? {}) as { capabilities?: string[] }).capabilities;
+    return !caps?.length || caps.includes(id);
+  }).length;
+}
+
+function jsonSchema(schema: z.ZodType, io: "input" | "output"): Record<string, unknown> {
+  try {
+    const s = z.toJSONSchema(schema, { io, unrepresentable: "any" }) as Record<string, unknown>;
+    delete s.$schema;
+    return s;
+  } catch {
+    return {};
+  }
+}
+
+const pascal = (id: string) =>
+  id
+    .split(/[._]/)
+    .map((p) => p.charAt(0).toUpperCase() + p.slice(1))
+    .join("");
+
+export async function capabilityTools(
+  db: Kysely<DB>,
+  registry: CapabilityRegistry,
+  settings: SettingsReader,
+  now = new Date(),
+): Promise<CapabilityToolsResponse> {
+  const automation = await settings.get("automation");
+  const grants = await activeGrants(db, now);
+  const items: CapabilityToolInfo[] = registry
+    .list()
+    .map(({ descriptor: d, toolName }) => {
+      const cat = CAPABILITY_CATALOG[d.id as KnownCapabilityId];
+      const readish = d.kind === "read" || d.kind === "preview";
+      const n = readish ? 0 : grantsCovering(d.id, grants);
+      const authState: CapabilityAuthState = d.id.startsWith("agent.")
+        ? "agent_only"
+        : readish
+          ? "read_only"
+          : n > 0
+            ? "granted"
+            : "needs_approval";
+      return {
+        id: d.id,
+        version: d.version,
+        title: d.title,
+        description: d.description,
+        kind: d.kind,
+        inputSchema: cat?.inputSchema ?? `${pascal(d.id)}Input`,
+        outputSchema: cat?.outputSchema ?? `${pascal(d.id)}Output`,
+        scope: d.scope,
+        riskLevel: d.riskLevel,
+        idempotent: d.idempotent,
+        supportsPreview: d.supportsPreview,
+        replay: d.replay,
+        toolName,
+        inputJsonSchema: jsonSchema(d.inputSchema, "input"),
+        outputJsonSchema: jsonSchema(d.outputSchema, "output"),
+        authState,
+        activeGrants: n,
+        mcpExposed: mcpToolVisible(d, ["admin"]),
+      };
+    })
+    .sort((a, b) => a.id.localeCompare(b.id));
+  return { automationEnabled: automation.enabled, readOnlyReconcile: automation.readOnlyReconcile, items };
+}