From 653100ceb47c00e7a5001844537efd26ef98b843 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 20:28:34 +0800 Subject: [PATCH] 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 --- apps/server/src/modules/index.ts | 2 + apps/server/src/modules/mcp/catalog.ts | 186 ++++++++++++++ apps/server/src/modules/mcp/config-store.ts | 56 +++++ apps/server/src/modules/mcp/index.ts | 31 +++ apps/server/src/modules/mcp/routes.ts | 179 ++++++++++++++ apps/server/src/modules/mcp/server.ts | 188 ++++++++++++++ apps/server/src/modules/mcp/sessions.ts | 133 ++++++++++ apps/server/src/modules/mcp/stats.ts | 192 +++++++++++++++ apps/server/test/mcp/mcp.test.ts | 257 ++++++++++++++++++++ apps/server/test/modules.test.ts | 3 +- 10 files changed, 1226 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/modules/mcp/catalog.ts create mode 100644 apps/server/src/modules/mcp/config-store.ts create mode 100644 apps/server/src/modules/mcp/index.ts create mode 100644 apps/server/src/modules/mcp/routes.ts create mode 100644 apps/server/src/modules/mcp/server.ts create mode 100644 apps/server/src/modules/mcp/sessions.ts create mode 100644 apps/server/src/modules/mcp/stats.ts create mode 100644 apps/server/test/mcp/mcp.test.ts diff --git a/apps/server/src/modules/index.ts b/apps/server/src/modules/index.ts index 6c34d1bb..7654a110 100644 --- a/apps/server/src/modules/index.ts +++ b/apps/server/src/modules/index.ts @@ -3,6 +3,7 @@ import { agentModule } from "./agent/index.ts"; import { authSettingsModule } from "./auth-settings/index.ts"; import { browserModule } from "./browser/index.ts"; import { fileManagerModule } from "./file-manager/index.ts"; +import { mcpModule } from "./mcp/index.ts"; import { piModule } from "./pi/index.ts"; import { schedulesModule } from "./schedules/index.ts"; import { searchModule } from "./search/index.ts"; @@ -26,5 +27,6 @@ export const SERVER_MODULES: readonly ServerModule[] = [ schedulesModule, searchModule, browserModule, + mcpModule, systemModule, ]; diff --git a/apps/server/src/modules/mcp/catalog.ts b/apps/server/src/modules/mcp/catalog.ts new file mode 100644 index 00000000..3c514cae --- /dev/null +++ b/apps/server/src/modules/mcp/catalog.ts @@ -0,0 +1,186 @@ +// @ts-types="@modelcontextprotocol/sdk/types" +import type { GetPromptResult } from "@modelcontextprotocol/sdk/types.js"; +import { sql } from "@umt/db"; +import type { ServerContext } from "../types.ts"; + +// MCP 资源(只读摘要)与提示模板。资源内容出服务端前再过一遍 redactForModel(见 server.ts)。 + +export interface McpResourceDef { + uri: string; + name: string; + title: string; + description: string; + mimeType: "application/json"; + load(ctx: ServerContext): Promise; +} + +const OPEN_DECISION = ["open", "investigating", "awaiting_user"] as const; + +export const MCP_RESOURCES: readonly McpResourceDef[] = [ + { + uri: "umt://settings/summary", + name: "settings-summary", + title: "设置概要", + description: "自动化总开关、只读对账、预算与 Agent 运行上限。不含凭证。", + mimeType: "application/json", + async load(ctx) { + const s = await ctx.settings.all(); + return { + automation: s.automation, + budget: s.budget, + agent: { + maxToolCalls: s.agent.maxToolCalls, + maxDurationSec: s.agent.maxDurationSec, + autoInvestigate: s.agent.autoInvestigate, + autoInvestigateMode: s.agent.autoInvestigateMode, + }, + organize: { defaultMode: s.organize.defaultMode, allowCrossVolumeMove: s.organize.allowCrossVolumeMove }, + retention: s.retention, + }; + }, + }, + { + uri: "umt://decisions/pending", + name: "decisions-pending", + title: "待处理", + description: "未结束的待处理事项（最多 50 条），按严重程度与时间排序。", + mimeType: "application/json", + async load(ctx) { + const rows = await ctx.db + .selectFrom("decision") + .select(["id", "kind", "status", "severity", "title", "summary", "subject_type", "subject_id", "created_at"]) + .where("status", "in", OPEN_DECISION) + .orderBy(sql`case severity when 'critical' then 0 when 'warning' then 1 else 2 end`) + .orderBy("created_at", "desc") + .limit(50) + .execute(); + const total = await ctx.db + .selectFrom("decision") + .select((eb) => eb.fn.countAll().as("n")) + .where("status", "in", OPEN_DECISION) + .executeTakeFirst(); + return { + total: Number(total?.n ?? 0), + items: rows.map((r) => ({ + id: r.id, + kind: r.kind, + status: r.status, + severity: r.severity, + title: r.title, + summary: r.summary, + subject: r.subject_type ? { type: r.subject_type, id: r.subject_id } : null, + createdAt: r.created_at.toISOString(), + })), + }; + }, + }, + { + uri: "umt://subscriptions/overview", + name: "subscriptions-overview", + title: "订阅总览", + description: "各状态的订阅数与最近更新的 20 个订阅。", + mimeType: "application/json", + async load(ctx) { + const counts = await ctx.db + .selectFrom("subscription") + .select(["status", (eb) => eb.fn.countAll().as("n")]) + .groupBy("status") + .execute(); + const recent = await ctx.db + .selectFrom("subscription as s") + .leftJoin("media_entity as e", "e.id", "s.media_entity_id") + .select([ + "s.id", + "s.status", + "s.media_kind", + "s.title_query", + "e.title", + "e.year", + "s.last_searched_at", + "s.next_search_at", + "s.updated_at", + ]) + .orderBy("s.updated_at", "desc") + .limit(20) + .execute(); + return { + byStatus: Object.fromEntries(counts.map((c) => [c.status, Number(c.n)])), + recent: recent.map((r) => ({ + id: r.id, + title: r.title ?? r.title_query, + year: r.year, + mediaKind: r.media_kind, + status: r.status, + lastSearchedAt: r.last_searched_at?.toISOString() ?? null, + nextSearchAt: r.next_search_at?.toISOString() ?? null, + })), + }; + }, + }, +]; + +export interface McpPromptDef { + name: string; + title: string; + description: string; + arguments: { name: string; description: string; required: boolean }[]; + render(args: Record): string; +} + +export const MCP_PROMPTS: readonly McpPromptDef[] = [ + { + name: "triage_pending", + title: "梳理待处理", + description: "读待处理事项，逐条给出结论和下一步。", + arguments: [], + render: () => + [ + "读资源 umt://decisions/pending，逐条处理待处理事项：", + "1. 用 task__inspect、catalog__lookup、release__search 等只读工具查清原因；", + "2. 每条给出结论（能自动恢复 / 需要人工决定 / 可以忽略）和依据；", + "3. 有副作用的操作（重试任务、提交下载、整理）只在已有持续授权时调用，返回 unauthorized 就停下，列出需要的审批。", + ].join("\n"), + }, + { + name: "acquire_title", + title: "收录一部作品", + description: "从查作品到生成下载计划；提交下载要有授权。", + arguments: [ + { name: "title", description: "作品标题", required: true }, + { name: "year", description: "年份（可选）", required: false }, + ], + render: (a) => + [ + `目标：收录「${a.title ?? ""}」${a.year ? `（${a.year}）` : ""}。`, + "1. catalog__lookup 确认作品，拿不准时列出候选，不要猜；", + "2. release__search 找资源，release__inspect 看细节；", + "3. download__plan 生成下载计划并说明选择理由；", + "4. 有持续授权时再调用 download__submit；返回 unauthorized 就停下，给出计划 ID 等审批。", + ].join("\n"), + }, + { + name: "diagnose_task", + title: "诊断任务", + description: "查看业务任务的状态和操作记录，说明失败原因与恢复办法。", + arguments: [{ name: "taskId", description: "业务任务 ID", required: true }], + render: (a) => + [ + `用 task__inspect 查看任务 ${a.taskId ?? ""}：`, + "1. 说明当前状态、失败在哪一步、错误原文；", + "2. 判断重试能否解决；能解决且有持续授权时调用 task__retry，否则说明需要的操作；", + "3. 不要编造任务里没有的信息。", + ].join("\n"), + }, +]; + +/** 渲染提示模板;模板不存在或缺必填参数时返回错误说明 */ +export function renderPrompt(name: string, args: Record): GetPromptResult | { error: string } { + const p = MCP_PROMPTS.find((x) => x.name === name); + if (!p) return { error: `没有这个提示模板：${name}` }; + const missing = p.arguments.filter((a) => a.required && !args[a.name]?.trim()).map((a) => a.name); + if (missing.length) return { error: `缺少参数：${missing.join("、")}` }; + return { + description: p.description, + messages: [{ role: "user", content: { type: "text", text: p.render(args) } }], + }; +} diff --git a/apps/server/src/modules/mcp/config-store.ts b/apps/server/src/modules/mcp/config-store.ts new file mode 100644 index 00000000..eb731e8b --- /dev/null +++ b/apps/server/src/modules/mcp/config-store.ts @@ -0,0 +1,56 @@ +import type { McpServerConfig, McpServerConfigUpdate } from "@umt/contracts"; +import type { DB, Kysely } from "@umt/db"; + +// mcp_server_config 单行读写。读有 5s 缓存(每个 MCP 请求都要看启用开关与允许的 Key),写后立即失效。 + +const TTL_MS = 5_000; + +export interface McpConfigStore { + get(): Promise; + update(patch: McpServerConfigUpdate, actorId: string): Promise; +} + +const DEFAULT: McpServerConfig = { enabled: true, allowedApiKeyIds: null, updatedAt: null }; + +export function createMcpConfigStore(db: () => Kysely): McpConfigStore { + let cache: { at: number; value: McpServerConfig } | null = null; + const read = async (): Promise => { + const r = await db().selectFrom("mcp_server_config").selectAll().where("id", "=", 1).executeTakeFirst(); + if (!r) return DEFAULT; + return { enabled: r.enabled, allowedApiKeyIds: r.allowed_api_key_ids, updatedAt: r.updated_at.toISOString() }; + }; + return { + async get() { + if (cache && Date.now() - cache.at < TTL_MS) return cache.value; + const value = await read(); + cache = { at: Date.now(), value }; + return value; + }, + async update(patch, actorId) { + const set = { + ...(patch.enabled !== undefined ? { enabled: patch.enabled } : {}), + ...(patch.allowedApiKeyIds !== undefined + ? { allowed_api_key_ids: patch.allowedApiKeyIds ? [...new Set(patch.allowedApiKeyIds)] : null } + : {}), + updated_by: actorId, + updated_at: new Date(), + }; + await db() + .insertInto("mcp_server_config") + .values({ id: 1, ...set }) + .onConflict((oc) => oc.column("id").doUpdateSet(set)) + .execute(); + cache = null; + return read(); + }, + }; +} + +/** 这个 Key 能不能连 MCP:总开关开着,且在允许列表里(列表为 null 表示不限) */ +export function keyAllowed(cfg: McpServerConfig, principal: { kind: string; id: string }): string | null { + if (!cfg.enabled) return "MCP 服务端已停用。在设置 → Agent → 工具与 MCP 里启用"; + if (principal.kind === "api_key" && cfg.allowedApiKeyIds && !cfg.allowedApiKeyIds.includes(principal.id)) { + return "这个 API Key 不在 MCP 允许列表里"; + } + return null; +} diff --git a/apps/server/src/modules/mcp/index.ts b/apps/server/src/modules/mcp/index.ts new file mode 100644 index 00000000..0b81571d --- /dev/null +++ b/apps/server/src/modules/mcp/index.ts @@ -0,0 +1,31 @@ +import denoJson from "../../../deno.json" with { type: "json" }; +import type { ServerModule } from "../types.ts"; +import { createMcpConfigStore } from "./config-store.ts"; +import { registerMcpRoutes } from "./routes.ts"; +import { McpSessionTable } from "./sessions.ts"; + +/** + * AG-1 对外 MCP 服务端(REQ R13、§12.3):/mcp 走 Streamable HTTP,API Key 鉴权, + * 每个工具对应一个能力,调用走同一个 registry.invoke(不绕过授权);另提供能力统计、能力清单与 MCP 状态/配置。 + * 会话表在进程内(api / all 角色),停机时全部关闭。 + */ +export function createMcpModule(opts: { version?: string } = {}): ServerModule & { sessions: McpSessionTable } { + const table = new McpSessionTable(); + return { + name: "mcp", + sessions: table, + registerRoutes(app, ctx) { + registerMcpRoutes(app, ctx, { + table, + config: createMcpConfigStore(() => ctx.db), + version: opts.version ?? denoJson.version, + }); + }, + registerTasks() {}, + async stop() { + await table.closeAll(); + }, + }; +} + +export const mcpModule = createMcpModule(); diff --git a/apps/server/src/modules/mcp/routes.ts b/apps/server/src/modules/mcp/routes.ts new file mode 100644 index 00000000..4c126707 --- /dev/null +++ b/apps/server/src/modules/mcp/routes.ts @@ -0,0 +1,179 @@ +import { randomUUID } from "node:crypto"; +// @ts-types="@modelcontextprotocol/sdk/server/auth/types" +import type { AuthInfo } from "@modelcontextprotocol/sdk/server/auth/types.js"; +// @ts-types="@modelcontextprotocol/sdk/server/webStandardStreamableHttp" +import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; +// @ts-types="@modelcontextprotocol/sdk/types" +import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"; +import { API_PREFIX, Api, type ApiScope, MCP_PATH, MCP_SERVER_NAME } from "@umt/contracts"; +import type { Context } from "hono"; +import type { AuthEnv } from "../../http/auth.ts"; +import { route } from "../agent/http.ts"; +import type { ModuleApp, ServerContext } from "../types.ts"; +import { keyAllowed, type McpConfigStore } from "./config-store.ts"; +import { createMcpServer, visibleTools } from "./server.ts"; +import { MCP_LIMITS, type McpSessionTable } from "./sessions.ts"; +import { capabilityStats, capabilityTools, recentMcpCalls } from "./stats.ts"; + +// POST/GET/DELETE /mcp:Streamable HTTP(不进 OpenAPI);其余为能力统计、能力清单与 MCP 状态/配置。 + +/** JSON-RPC 错误信封(HTTP 层拒绝时用;MCP 客户端按 JSON-RPC 解析) */ +function rpcError(c: Context, status: 400 | 401 | 403 | 404 | 429 | 503, message: string) { + return c.json({ jsonrpc: "2.0", error: { code: -32000, message }, id: null }, status); +} + +export interface McpRouteDeps { + table: McpSessionTable; + config: McpConfigStore; + version: string; +} + +/** 常见 Key 组合 → 可见工具数(键名给界面用) */ +const SCOPE_SAMPLES: Record<"read" | "readExecute" | "admin", ApiScope[]> = { + read: ["read"], + readExecute: ["read", "organize.execute"], + admin: ["admin"], +}; + +export function registerMcpRoutes(app: ModuleApp, ctx: ServerContext, deps: McpRouteDeps) { + const { table, config, version } = deps; + + app.all(MCP_PATH, async (c) => { + const p = c.get("principal"); + if (!p) return rpcError(c, 401, "需要 Bearer API Key"); + if (p.kind === "admin_session") return rpcError(c, 403, "MCP 只接受 API Key（Authorization: Bearer）"); + const denied = keyAllowed(await config.get(), p); + if (denied) return rpcError(c, 403, denied); + if (!table.requests.take(p.id)) return rpcError(c, 429, "请求太频繁，超过每分钟上限。稍后重试"); + const authInfo: AuthInfo = { token: "", clientId: p.id, scopes: p.scopes, extra: { kind: p.kind, name: p.name } }; + + const sid = c.req.header("mcp-session-id"); + if (sid) { + const s = table.get(sid); + if (!s) return rpcError(c, 404, "会话不存在或已过期。重新初始化"); + if (s.principalId !== p.id) return rpcError(c, 403, "这个会话属于另一个 API Key"); + s.lastSeenAt = new Date(); + const res = await s.transport.handleRequest(c.req.raw, { authInfo }); + if (c.req.method === "DELETE") await table.remove(sid, "client"); + return res; + } + + if (c.req.method !== "POST") return rpcError(c, 400, "缺少 Mcp-Session-Id。先发 initialize"); + let body: unknown; + try { + body = await c.req.json(); + } catch { + return rpcError(c, 400, "请求体不是合法的 JSON"); + } + const init = Array.isArray(body) ? body.find(isInitializeRequest) : isInitializeRequest(body) ? body : null; + if (!init) return rpcError(c, 400, "缺少 Mcp-Session-Id。先发 initialize"); + if (table.countFor(p.id) >= MCP_LIMITS.maxSessionsPerKey) { + return rpcError(c, 429, `这个 API Key 的会话数已达上限（${MCP_LIMITS.maxSessionsPerKey}）。先关闭不用的会话`); + } + if (table.sessions.size >= MCP_LIMITS.maxSessions) return rpcError(c, 503, "MCP 会话数已满。稍后重试"); + + const clientName = (init.params.clientInfo?.name ?? null) as string | null; + let sessionId: string | undefined; + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + onsessioninitialized: (id) => { + sessionId = id; + const now = new Date(); + table.add({ + id, + transport, + server, + principalKind: p.kind, + principalId: p.id, + principalName: p.name, + clientName, + createdAt: now, + lastSeenAt: now, + toolCalls: 0, + }); + void audit(ctx, p, "mcp.session.open", id, { clientName }); + }, + onsessionclosed: (id) => void table.remove(id, "client"), + }); + const server = createMcpServer(ctx, table, version, () => sessionId); + await server.connect(transport); + return transport.handleRequest(c.req.raw, { parsedBody: body, authInfo }); + }); + + route( + app, + Api.capabilities.stats, + async ({ query }) => capabilityStats(ctx.db, (query as { days: number }).days ?? 7), + { tag: "mcp" }, + ); + route(app, Api.capabilities.tools, async () => capabilityTools(ctx.db, ctx.capabilities, ctx.settings), { + tag: "mcp", + }); + route( + app, + Api.mcp.status, + async () => { + const cfg = await config.get(); + return { + ...cfg, + path: `${API_PREFIX}${MCP_PATH}`, + transport: "streamable-http" as const, + serverName: MCP_SERVER_NAME, + serverVersion: version, + rateLimitPerMinute: MCP_LIMITS.toolCallsPerMinute, + maxSessionsPerKey: MCP_LIMITS.maxSessionsPerKey, + sessions: table.list(), + recentCalls: await recentMcpCalls(ctx.db), + toolsByScope: { + read: visibleTools(ctx.capabilities, SCOPE_SAMPLES.read).length, + readExecute: visibleTools(ctx.capabilities, SCOPE_SAMPLES.readExecute).length, + admin: visibleTools(ctx.capabilities, SCOPE_SAMPLES.admin).length, + }, + }; + }, + { tag: "mcp" }, + ); + route( + app, + Api.mcp.update, + async ({ body, actorId, c }) => { + const before = await config.get(); + const after = await config.update(body, actorId); + const p = c.get("principal"); + await audit(ctx, p, "mcp.config.update", "mcp", { + before: { enabled: before.enabled, allowedApiKeyIds: before.allowedApiKeyIds }, + after: { enabled: after.enabled, allowedApiKeyIds: after.allowedApiKeyIds }, + }); + // 停用或缩小允许列表时,立即断开不再允许的会话 + for (const s of [...table.sessions.values()]) { + if (keyAllowed(after, { kind: s.principalKind, id: s.principalId })) await table.remove(s.id, "config"); + } + return after; + }, + { tag: "mcp" }, + ); +} + +async function audit( + ctx: ServerContext, + p: { kind: string; id: string } | undefined, + action: string, + targetId: string, + payload: object, +) { + try { + await ctx.db + .insertInto("audit_event") + .values({ + actor_type: p?.kind === "api_key" ? "api_key" : "user", + actor_id: p ? `${p.kind === "api_key" ? "api_key" : "user"}:${p.id}` : null, + action, + target_type: "mcp", + target_id: targetId, + payload: JSON.stringify(payload), + }) + .execute(); + } catch (err) { + ctx.logger.warn({ err, action }, "mcp audit failed"); + } +} diff --git a/apps/server/src/modules/mcp/server.ts b/apps/server/src/modules/mcp/server.ts new file mode 100644 index 00000000..c607b415 --- /dev/null +++ b/apps/server/src/modules/mcp/server.ts @@ -0,0 +1,188 @@ +// @ts-types="@modelcontextprotocol/sdk/server/auth/types" +import type { AuthInfo } from "@modelcontextprotocol/sdk/server/auth/types.js"; +// @ts-types="@modelcontextprotocol/sdk/server/index" +import { Server } from "@modelcontextprotocol/sdk/server/index.js"; +// @ts-types="@modelcontextprotocol/sdk/types" +import { + CallToolRequestSchema, + type CallToolResult, + ErrorCode, + GetPromptRequestSchema, + ListPromptsRequestSchema, + ListResourcesRequestSchema, + ListToolsRequestSchema, + McpError, + ReadResourceRequestSchema, + type Tool, +} from "@modelcontextprotocol/sdk/types.js"; +import { type CapabilityRegistry, isCapabilityError, redactForModel, toolNameToCapabilityId } from "@umt/agent"; +import { type CapabilityKind, MCP_SERVER_NAME, mcpToolVisible } from "@umt/contracts"; +import { sharedInvokeLimiter, UpstreamBusyError, UpstreamTimeoutError } from "../agent/invoke-limits.ts"; +import type { ServerContext } from "../types.ts"; +import { MCP_PROMPTS, MCP_RESOURCES, renderPrompt } from "./catalog.ts"; +import type { McpSessionTable } from "./sessions.ts"; + +// 一个 MCP 会话对应一个 Server 实例。tools 就是能力注册中心里对这个 Key 可见的能力, +// 调用一律走 ctx.capabilities.invoke(source = "mcp"):授权回调、审计、服务端幂等键、操作账本与 Web / Agent 同一条路径。 +// 模型或客户端参数里的"已获同意"不起作用,授权只看 automation_grant / approval 记录。 + +const KIND_LABEL: Record = { + read: "只读", + preview: "预览，不落变更", + propose: "提案，需要授权", + approve: "审批，需要授权", + execute: "执行，有副作用，需要持续授权或单次审批", +}; + +/** 请求时的调用者(每个请求都重新鉴权,scope 以当前请求为准) */ +export interface McpCaller { + kind: string; + id: string; + name: string; + scopes: string[]; +} + +export function callerOf(auth: AuthInfo | undefined): McpCaller { + const extra = (auth?.extra ?? {}) as { kind?: string; name?: string }; + return { + kind: extra.kind ?? "api_key", + id: auth?.clientId ?? "?", + name: extra.name ?? "?", + scopes: auth?.scopes ?? [], + }; +} + +export const actorIdOfCaller = (c: McpCaller) => (c.kind === "api_key" ? `api_key:${c.id}` : `user:${c.id}`); + +/** 对这组 scope 可见的 MCP 工具 */ +export function visibleTools(registry: CapabilityRegistry, scopes: readonly string[]): Tool[] { + return registry + .list() + .filter((c) => mcpToolVisible(c.descriptor, scopes)) + .map(({ descriptor: d, toolName }) => { + const schema = registry.inputJsonSchema(d.id); + return { + name: toolName, + title: d.title, + description: [ + d.description, + "", + `能力 ${d.id} v${d.version}；${KIND_LABEL[d.kind]}；风险 ${d.riskLevel}；scope ${d.scope}。`, + ].join("\n"), + inputSchema: { ...schema, type: "object" } as Tool["inputSchema"], + annotations: { + title: d.title, + readOnlyHint: d.kind === "read" || d.kind === "preview", + destructiveHint: d.kind === "execute" && d.riskLevel === "high", + idempotentHint: d.idempotent, + openWorldHint: false, + }, + }; + }) + .sort((a, b) => a.name.localeCompare(b.name)); +} + +const errorResult = (code: string, message: string): CallToolResult => ({ + isError: true, + content: [{ type: "text", text: JSON.stringify({ error: { code, message } }) }], +}); + +export function createMcpServer( + ctx: ServerContext, + table: McpSessionTable, + version: string, + getSessionId: () => string | undefined, +): Server { + const server = new Server( + { name: MCP_SERVER_NAME, title: "UHD 媒体自动化", version }, + { + capabilities: { + tools: { listChanged: false }, + resources: { listChanged: false }, + prompts: { listChanged: false }, + }, + instructions: + "工具与控制台、内置 Agent 共用同一套能力、授权与审计。只读工具可以直接调用；有副作用的工具需要持续授权或单次审批，缺授权时返回 unauthorized，不要重复尝试。资源里有设置概要、待处理事项与订阅总览。", + }, + ); + const limiter = sharedInvokeLimiter(); + + server.setRequestHandler(ListToolsRequestSchema, (_req, extra) => ({ + tools: visibleTools(ctx.capabilities, callerOf(extra.authInfo).scopes), + })); + + server.setRequestHandler(CallToolRequestSchema, async (req, extra): Promise => { + const caller = callerOf(extra.authInfo); + const id = toolNameToCapabilityId(req.params.name); + const entry = id ? ctx.capabilities.get(id) : undefined; + if (!id || !entry) return errorResult("not_found", `没有这个工具：${req.params.name}`); + if (!mcpToolVisible(entry.descriptor, caller.scopes)) { + return errorResult("forbidden", `当前 API Key 的 scope 不能调用 ${id}（需要 ${entry.descriptor.scope}）`); + } + if (!table.toolCalls.take(caller.id)) { + return errorResult("rate_limited", "工具调用太频繁，超过每分钟上限。稍后重试"); + } + const sessionId = extra.sessionId ?? getSessionId() ?? "nosession"; + const session = table.get(sessionId); + if (session) session.toolCalls++; + const allowedKinds: CapabilityKind[] = ["read", "preview", "propose", "approve", "execute"].filter((k) => + mcpToolVisible({ ...entry.descriptor, kind: k as CapabilityKind }, caller.scopes), + ) as CapabilityKind[]; + try { + const input = req.params.arguments ?? {}; + const output = await limiter.run(id, input, extra.signal, (signal) => + ctx.capabilities.invoke(id, input, { + actorId: actorIdOfCaller(caller), + // 幂等键按会话归属:同一会话同参数的有副作用调用只执行一次(账本直接返回上次结果) + taskId: `mcp:${sessionId}`, + scopeIds: [], + signal: signal ?? extra.signal, + source: "mcp", + invocationId: `mcp:${sessionId}:${String(extra.requestId)}`, + allowedKinds, + }), + ); + const safe = redactForModel(output); + const structured = + safe && typeof safe === "object" && !Array.isArray(safe) ? (safe as Record) : undefined; + return { + content: [{ type: "text", text: JSON.stringify(safe ?? null) }], + ...(structured ? { structuredContent: structured } : {}), + }; + } catch (err) { + if (err instanceof UpstreamBusyError) return errorResult("rate_limited", err.message); + if (err instanceof UpstreamTimeoutError) return errorResult("upstream_unavailable", err.message); + if (isCapabilityError(err)) return errorResult(err.code, err.message); + ctx.logger.error({ err, capability: id }, "mcp tool call failed"); + return errorResult("internal", "服务器内部错误"); + } + }); + + server.setRequestHandler(ListResourcesRequestSchema, (_req, extra) => { + const caller = callerOf(extra.authInfo); + return { resources: hasReadScope(caller.scopes) ? MCP_RESOURCES.map(({ load: _, ...r }) => r) : [] }; + }); + + server.setRequestHandler(ReadResourceRequestSchema, async (req, extra) => { + const caller = callerOf(extra.authInfo); + const r = MCP_RESOURCES.find((x) => x.uri === req.params.uri); + if (!r) throw new McpError(ErrorCode.InvalidParams, `没有这个资源：${req.params.uri}`); + if (!hasReadScope(caller.scopes)) throw new McpError(ErrorCode.InvalidRequest, "读取资源需要 read scope"); + const data = redactForModel(await r.load(ctx)); + return { contents: [{ uri: r.uri, mimeType: r.mimeType, text: JSON.stringify(data, null, 2) }] }; + }); + + server.setRequestHandler(ListPromptsRequestSchema, () => ({ + prompts: MCP_PROMPTS.map(({ render: _, ...p }) => p), + })); + + server.setRequestHandler(GetPromptRequestSchema, (req) => { + const out = renderPrompt(req.params.name, req.params.arguments ?? {}); + if (!("messages" in out)) throw new McpError(ErrorCode.InvalidParams, out.error); + return out; + }); + + return server; +} + +const hasReadScope = (scopes: readonly string[]) => scopes.includes("read") || scopes.includes("admin"); diff --git a/apps/server/src/modules/mcp/sessions.ts b/apps/server/src/modules/mcp/sessions.ts new file mode 100644 index 00000000..9229e583 --- /dev/null +++ b/apps/server/src/modules/mcp/sessions.ts @@ -0,0 +1,133 @@ +// @ts-types="@modelcontextprotocol/sdk/server/index" +import type { Server } from "@modelcontextprotocol/sdk/server/index.js"; +// @ts-types="@modelcontextprotocol/sdk/server/webStandardStreamableHttp" +import type { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; +import type { McpSessionInfo } from "@umt/contracts"; + +// MCP 会话表(进程内):会话与建立它的 API Key 绑定,换 Key 复用会话一律拒绝。 +// 空闲超时回收;每个 Key 的会话数与每分钟工具调用数有上限。 + +export const MCP_LIMITS = { + /** 每个 API Key 每分钟的工具调用上限 */ + toolCallsPerMinute: 60, + /** 每个 API Key 每分钟的 HTTP 请求上限(含 list / resources / 推送流) */ + requestsPerMinute: 600, + /** 每个 API Key 同时保持的会话上限 */ + maxSessionsPerKey: 8, + /** 全进程会话上限 */ + maxSessions: 256, + /** 空闲多久回收(ms) */ + idleMs: 30 * 60_000, +} as const; + +export interface McpSession { + id: string; + transport: WebStandardStreamableHTTPServerTransport; + server: Server; + principalKind: string; + principalId: string; + principalName: string; + clientName: string | null; + createdAt: Date; + lastSeenAt: Date; + toolCalls: number; +} + +/** 滑动窗口计数器:key → 最近一分钟的时间戳 */ +export class RateWindow { + private readonly hits = new Map(); + constructor( + readonly limit: number, + readonly windowMs = 60_000, + private readonly now: () => number = Date.now, + ) {} + + /** 记一次;超限返回 false(不计入) */ + take(key: string): boolean { + const t = this.now(); + const arr = (this.hits.get(key) ?? []).filter((x) => t - x < this.windowMs); + if (arr.length >= this.limit) { + this.hits.set(key, arr); + return false; + } + arr.push(t); + this.hits.set(key, arr); + return true; + } + + /** 清掉整窗口都没有记录的 key */ + sweep(): void { + const t = this.now(); + for (const [k, arr] of this.hits) if (arr.every((x) => t - x >= this.windowMs)) this.hits.delete(k); + } +} + +export class McpSessionTable { + readonly sessions = new Map(); + readonly toolCalls = new RateWindow(MCP_LIMITS.toolCallsPerMinute); + readonly requests = new RateWindow(MCP_LIMITS.requestsPerMinute); + private timer: ReturnType | null = null; + + constructor(private readonly onClose?: (s: McpSession, reason: string) => void) {} + + countFor(principalId: string): number { + let n = 0; + for (const s of this.sessions.values()) if (s.principalId === principalId) n++; + return n; + } + + add(s: McpSession): void { + this.sessions.set(s.id, s); + this.ensureSweeper(); + } + + get(id: string): McpSession | undefined { + return this.sessions.get(id); + } + + /** 从表里摘掉并关闭传输(transport.close 会触发 onsessionclosed,重复调用无害) */ + async remove(id: string, reason: string): Promise { + const s = this.sessions.get(id); + if (!s) return; + this.sessions.delete(id); + this.onClose?.(s, reason); + await s.server.close().catch(() => {}); + } + + /** 回收空闲会话 */ + async sweep(now = Date.now()): Promise { + for (const s of [...this.sessions.values()]) { + if (now - s.lastSeenAt.getTime() > MCP_LIMITS.idleMs) await this.remove(s.id, "idle"); + } + this.toolCalls.sweep(); + this.requests.sweep(); + } + + async closeAll(): Promise { + if (this.timer) clearInterval(this.timer); + this.timer = null; + for (const id of [...this.sessions.keys()]) await this.remove(id, "shutdown"); + } + + list(): McpSessionInfo[] { + return [...this.sessions.values()] + .sort((a, b) => b.lastSeenAt.getTime() - a.lastSeenAt.getTime()) + .map((s) => ({ + idPrefix: s.id.slice(0, 8), + apiKeyId: s.principalId, + apiKeyName: s.principalName, + clientName: s.clientName, + createdAt: s.createdAt.toISOString(), + lastSeenAt: s.lastSeenAt.toISOString(), + toolCalls: s.toolCalls, + })); + } + + private ensureSweeper() { + if (this.timer) return; + this.timer = setInterval(() => void this.sweep(), 60_000); + // 不因为回收定时器挂住进程退出 + (this.timer as unknown as { unref?: () => void }).unref?.(); + if (typeof Deno !== "undefined") Deno.unrefTimer(this.timer as unknown as number); + } +} diff --git a/apps/server/src/modules/mcp/stats.ts b/apps/server/src/modules/mcp/stats.ts new file mode 100644 index 00000000..f59646cd --- /dev/null +++ b/apps/server/src/modules/mcp/stats.ts @@ -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, + days: number, + now = new Date(), +): Promise { + 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, limit = 20): Promise { + 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`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, now: Date): Promise { + 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 { + try { + const s = z.toJSONSchema(schema, { io, unrepresentable: "any" }) as Record; + 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, + registry: CapabilityRegistry, + settings: SettingsReader, + now = new Date(), +): Promise { + 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 }; +} diff --git a/apps/server/test/mcp/mcp.test.ts b/apps/server/test/mcp/mcp.test.ts new file mode 100644 index 00000000..2a2897db --- /dev/null +++ b/apps/server/test/mcp/mcp.test.ts @@ -0,0 +1,257 @@ +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; +import type { CapabilitySpec } from "@umt/contracts"; +import { sql } from "@umt/db"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { z } from "zod"; +import { createCapabilityRegistry } from "../../src/modules/agent/capability-registry.ts"; +import { ensureSetupToken } from "../../src/modules/auth-settings/auth-service.ts"; +import { createMcpModule } from "../../src/modules/mcp/index.ts"; +import { RateWindow } from "../../src/modules/mcp/sessions.ts"; +import { Browser, bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts"; +import { json } from "../helpers.ts"; + +// AG-1:对外 MCP 服务端。用官方 SDK client 走 Streamable HTTP 实际连一次: +// 工具列表按 API Key scope 过滤、调用 read 能力、execute 能力无授权被拒(同一 registry.invoke → 审计 / 账本)、 +// 会话与 Key 绑定、启用开关 / 允许列表、资源与提示模板、能力统计与能力清单端点。 + +const hasPg = await pgAvailable(); + +const spec = (over: Partial) => + ({ + version: 1, + title: "t", + description: "测试用", + idempotent: true, + supportsPreview: false, + riskLevel: "low", + inputSchema: z.object({ text: z.string() }), + outputSchema: z.object({ echo: z.string(), token: z.string().optional() }), + ...over, + }) as unknown as CapabilitySpec; + +const ECHO = spec({ id: "demo.echo", kind: "read", scope: "read", replay: "safe" }); +const APPLY = spec({ + id: "demo.apply", + kind: "execute", + scope: "organize.execute", + riskLevel: "high", + replay: "never", +}); +const SPAWN = spec({ id: "agent.spawn", kind: "propose", scope: "organize.execute", replay: "never" }); + +describe.skipIf(!hasPg)("MCP 服务端(真实 PG + SDK client)", () => { + let h: Harness; + let readKey: string; + let execKey: string; + let applied = 0; + const mcp = createMcpModule({ version: "0.0.0-test" }); + + beforeAll(async () => { + h = await createHarness({ extraModules: [mcp] }); + // 换成带库的注册中心:真实授权器 + audit_event + 操作账本 + const reg = createCapabilityRegistry({ logger: h.ctx.logger, db: h.db, settings: h.ctx.settings }); + reg.register(ECHO, (i) => ({ echo: (i as { text: string }).text, token: "secret-token-value" })); + reg.register(APPLY, (i) => { + applied++; + return { echo: `applied:${(i as { text: string }).text}` }; + }); + reg.register(SPAWN, () => ({ echo: "x" })); + h.ctx.capabilities = reg; + + const admin = new Browser(h.app); + const token = await ensureSetupToken(h.ctx); + const r = await admin.req("POST", "/auth/setup", { + setupToken: token, + username: "admin", + password: "correct horse battery", + }); + expect(r.status).toBe(200); + const mk = async (name: string, scopes: string[]) => + (await json(await admin.req("POST", "/api-keys", { name, scopes }))).data.plaintextKey as string; + readKey = await mk("reader", ["read"]); + execKey = await mk("executor", ["read", "organize.execute"]); + }, 120_000); + + afterAll(async () => { + await mcp.stop?.(); + await h?.close(); + }); + + async function connect(key: string) { + const client = new Client({ name: "vitest-client", version: "1.0.0" }); + const transport = new StreamableHTTPClientTransport(new URL("http://umt.test/api/v1/mcp"), { + requestInit: { headers: { authorization: `Bearer ${key}` } }, + fetch: async (url, init) => await h.app.fetch(new Request(url, init)), + }); + await client.connect(transport); + return { client, transport }; + } + + it("read Key:只看到 read / preview 工具,调用只读能力拿到脱敏结果,execute 工具被拒", async () => { + const { client } = await connect(readKey); + const tools = (await client.listTools()).tools.map((t) => t.name); + expect(tools).toContain("demo__echo"); + expect(tools).not.toContain("demo__apply"); + expect(tools).not.toContain("agent__spawn"); + const echo = (await client.listTools()).tools.find((t) => t.name === "demo__echo"); + expect(echo?.inputSchema).toMatchObject({ type: "object", properties: { text: { type: "string" } } }); + expect(echo?.annotations?.readOnlyHint).toBe(true); + + const ok = await client.callTool({ name: "demo__echo", arguments: { text: "hi" } }); + expect(ok.isError).toBeFalsy(); + expect(ok.structuredContent).toMatchObject({ echo: "hi" }); + expect(JSON.stringify(ok.content)).not.toContain("secret-token-value"); + + const bad = await client.callTool({ name: "demo__echo", arguments: { text: 1 } }); + expect(bad.isError).toBe(true); + expect(JSON.stringify(bad.content)).toContain("invalid_input"); + + const denied = await client.callTool({ name: "demo__apply", arguments: { text: "x" } }); + expect(denied.isError).toBe(true); + expect(JSON.stringify(denied.content)).toContain("forbidden"); + expect(applied).toBe(0); + await client.close(); + }); + + it("execute Key:能看到 execute 工具,但没有持续授权时授权器拒绝;有授权后执行并记账本", async () => { + const { client } = await connect(execKey); + expect((await client.listTools()).tools.map((t) => t.name)).toContain("demo__apply"); + + const denied = await client.callTool({ name: "demo__apply", arguments: { text: "a" } }); + expect(denied.isError).toBe(true); + expect(JSON.stringify(denied.content)).toContain("unauthorized"); + expect(applied).toBe(0); + const audits = await sql<{ n: number }>` + select count(*)::int as n from audit_event + where action = 'capability.denied' and target_id = 'demo.apply' and payload->>'source' = 'mcp'`.execute(h.db); + expect(audits.rows[0]!.n).toBeGreaterThanOrEqual(1); + + await h.db.insertInto("automation_grant").values({ action: "demo.apply", granted_by: "user:admin" }).execute(); + const ok = await client.callTool({ name: "demo__apply", arguments: { text: "a" } }); + expect(ok.isError).toBeFalsy(); + expect(ok.structuredContent).toMatchObject({ echo: "applied:a" }); + // 同一会话同参数:账本返回上次结果,不重复执行 + await client.callTool({ name: "demo__apply", arguments: { text: "a" } }); + expect(applied).toBe(1); + const ledger = await h.db + .selectFrom("capability_invocation") + .select(["source", "status", "actor_id"]) + .where("capability_id", "=", "demo.apply") + .execute(); + expect(ledger).toEqual([{ source: "mcp", status: "succeeded", actor_id: expect.stringMatching(/^api_key:/) }]); + await client.close(); + }); + + it("资源与提示模板", async () => { + const { client } = await connect(readKey); + const res = (await client.listResources()).resources.map((r) => r.uri); + expect(res).toEqual(["umt://settings/summary", "umt://decisions/pending", "umt://subscriptions/overview"]); + for (const uri of res) { + const c = await client.readResource({ uri }); + const first = c.contents[0] as { text: string }; + expect(() => JSON.parse(first.text)).not.toThrow(); + } + const settings = JSON.parse( + ((await client.readResource({ uri: "umt://settings/summary" })).contents[0] as { text: string }).text, + ); + expect(settings.automation).toMatchObject({ enabled: true }); + const prompts = (await client.listPrompts()).prompts.map((p) => p.name); + expect(prompts).toEqual(["triage_pending", "acquire_title", "diagnose_task"]); + const p = await client.getPrompt({ name: "acquire_title", arguments: { title: "沙丘" } }); + expect(JSON.stringify(p.messages)).toContain("沙丘"); + await expect(client.getPrompt({ name: "diagnose_task", arguments: {} })).rejects.toThrow(/taskId/); + await client.close(); + }); + + it("会话与 API Key 绑定;cookie 会话、无凭证、停用与允许列表都被拒", async () => { + const { client, transport } = await connect(readKey); + const sid = transport.sessionId!; + expect(sid).toBeTruthy(); + const hijack = await h.app.request("/api/v1/mcp", { + method: "POST", + headers: { + authorization: `Bearer ${execKey}`, + "mcp-session-id": sid, + "content-type": "application/json", + accept: "application/json, text/event-stream", + }, + body: JSON.stringify({ jsonrpc: "2.0", id: 9, method: "tools/list" }), + }); + expect(hijack.status).toBe(403); + expect((await h.app.request("/api/v1/mcp", { method: "POST", body: "{}" })).status).toBe(401); + + const status = await json(await bearer(h.app, execKey)("GET", "/mcp/status")); + expect(status.error?.code).toBe("forbidden"); // admin scope + + const admin = new Browser(h.app); + await admin.req("POST", "/auth/login", { username: "admin", password: "correct horse battery" }); + const init = { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-06-18", capabilities: {}, clientInfo: { name: "x", version: "1" } }, + }; + const viaCookie = await admin.req("POST", "/mcp", init, { + headers: { accept: "application/json, text/event-stream" }, + }); + expect(viaCookie.status).toBe(403); + + const st = (await json(await admin.req("GET", "/mcp/status"))).data; + expect(st).toMatchObject({ enabled: true, path: "/api/v1/mcp", transport: "streamable-http" }); + expect(st.sessions.length).toBeGreaterThanOrEqual(1); + expect(st.recentCalls.length).toBeGreaterThan(0); + expect(st.toolsByScope.read).toBe(1); + + // 允许列表只放 execKey 之外的 Key → readKey 现有会话立即断开,新连接 403 + const keys = (await json(await admin.req("GET", "/api-keys?limit=50"))).data.items as { + id: string; + name: string; + }[]; + const execId = keys.find((k) => k.name === "executor")!.id; + const upd = await admin.req("PATCH", "/mcp/config", { allowedApiKeyIds: [execId] }); + expect(upd.status).toBe(200); + await expect(client.listTools()).rejects.toThrow(); + await expect(connect(readKey)).rejects.toThrow(); + const exec = await connect(execKey); + await exec.client.close(); + + await admin.req("PATCH", "/mcp/config", { enabled: false }); + await expect(connect(execKey)).rejects.toThrow(); + await admin.req("PATCH", "/mcp/config", { enabled: true, allowedApiKeyIds: null }); + const again = await connect(readKey); + await again.client.close(); + }); + + it("能力统计与能力清单端点", async () => { + const get = bearer(h.app, readKey); + const stats = (await json(await get("GET", "/capabilities/stats?days=7"))).data; + const echo = stats.items.find((i: { capabilityId: string }) => i.capabilityId === "demo.echo"); + expect(echo).toMatchObject({ lastCaller: "mcp" }); + expect(echo.calls).toBeGreaterThanOrEqual(2); + expect(echo.failures).toBeGreaterThanOrEqual(1); + expect(echo.byCaller.mcp).toBe(echo.calls); + const apply = stats.items.find((i: { capabilityId: string }) => i.capabilityId === "demo.apply"); + expect(apply.denied).toBeGreaterThanOrEqual(1); + + const tools = (await json(await get("GET", "/capabilities/tools"))).data; + const by = (id: string) => tools.items.find((i: { id: string }) => i.id === id); + expect(by("demo.echo")).toMatchObject({ authState: "read_only", toolName: "demo__echo", mcpExposed: true }); + expect(by("demo.apply")).toMatchObject({ authState: "granted", activeGrants: 1 }); + expect(by("agent.spawn")).toMatchObject({ authState: "agent_only", mcpExposed: false }); + expect(by("demo.echo").inputJsonSchema).toMatchObject({ type: "object" }); + }); +}); + +describe("RateWindow", () => { + it("窗口内超限拒绝,窗口滑过后恢复", () => { + let t = 0; + const w = new RateWindow(2, 1000, () => t); + expect(w.take("k")).toBe(true); + expect(w.take("k")).toBe(true); + expect(w.take("k")).toBe(false); + expect(w.take("other")).toBe(true); + t = 1001; + expect(w.take("k")).toBe(true); + }); +}); diff --git a/apps/server/test/modules.test.ts b/apps/server/test/modules.test.ts index 820bbaff..efb4bb85 100644 --- a/apps/server/test/modules.test.ts +++ b/apps/server/test/modules.test.ts @@ -43,7 +43,7 @@ const pingModule: ServerModule = { }; describe("ServerModule 骨架", () => { - it("清单里是四个 W2 模块 + 文件管理 + Pi 上游 + 定时任务 + 全局搜索 + 浏览器 + 进程与健康(末尾),名字唯一", () => { + it("清单里是四个 W2 模块 + 文件管理 + Pi 上游 + 定时任务 + 全局搜索 + 浏览器 + MCP + 进程与健康(末尾),名字唯一", () => { expect(SERVER_MODULES.map((m) => m.name)).toEqual([ "auth-settings", "sources-identity", @@ -54,6 +54,7 @@ describe("ServerModule 骨架", () => { "schedules", "search", "browser", + "mcp", "system", ]); });