|
|
@@ -1,98 +1,173 @@
|
|
|
-// MCP Gateway:服务进程内的 Streamable HTTP MCP 服务(/mcp),两个引擎共用同一套工具(ADR-0006)
|
|
|
+// MCP Gateway:服务进程内的 Streamable HTTP MCP 服务(/mcp),两个引擎共用同一套工具(ADR-0006)。
|
|
|
+// 提供内置工具(builtin.ts),并转接 Profile 勾选的第三方 MCP 服务(ADR-0021):工具名为 <服务名>__<工具名>,
|
|
|
+// 调用前按工具策略与权限模式审批,结果过长时截断并把完整内容存入工作区
|
|
|
+import crypto from 'node:crypto'
|
|
|
+import fs from 'node:fs'
|
|
|
import path from 'node:path'
|
|
|
import type { FastifyInstance } from 'fastify'
|
|
|
-import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'
|
|
|
+import { Server } from '@modelcontextprotocol/sdk/server/index.js'
|
|
|
import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js'
|
|
|
+import { CallToolRequestSchema, ListToolsRequestSchema, type Tool } from '@modelcontextprotocol/sdk/types.js'
|
|
|
import { z } from 'zod'
|
|
|
-import { CONVERT_FORMATS, convertDocument, recalcWorkbook } from '../services/office'
|
|
|
-import { FetchBlockedError, webFetch } from '../services/web-fetch'
|
|
|
+import type { AppSettings } from '@agent-server/shared'
|
|
|
+import type { McpCachedTool } from '../db/schema'
|
|
|
+import { defaultPolicy, type McpServerService } from '../services/mcp-servers'
|
|
|
+import { BUILTIN_TOOLS, fail, type ToolContent, type ToolResult } from './builtin'
|
|
|
import type { GatewayRunContext, RunTokenRegistry } from './tokens'
|
|
|
+import { isTransient, type McpServerRow, type McpUpstream } from './upstream'
|
|
|
|
|
|
export const GATEWAY_SERVER = 'gateway'
|
|
|
-export const GATEWAY_TOOLS = ['convert_document', 'recalc_workbook', 'web_fetch', 'kb_list_roots'] as const
|
|
|
+export const GATEWAY_TOOLS = BUILTIN_TOOLS.map(t => t.name)
|
|
|
+/** 第三方工具名中服务名与工具名的分隔符 */
|
|
|
+export const MCP_SEP = '__'
|
|
|
|
|
|
-type ToolResult = { content: { type: 'text'; text: string }[]; isError?: boolean }
|
|
|
-const ok = (text: string): ToolResult => ({ content: [{ type: 'text', text }] })
|
|
|
-const fail = (text: string): ToolResult => ({ content: [{ type: 'text', text }], isError: true })
|
|
|
+const MAX_TEXT = 100_000
|
|
|
+const MAX_INLINE_BINARY = 5 * 1048576
|
|
|
+const CALL_TIMEOUT_MS = 10 * 60_000
|
|
|
+const LIST_TIMEOUT_MS = 20_000
|
|
|
|
|
|
-function buildServer(ctx: GatewayRunContext): McpServer {
|
|
|
- const server = new McpServer({ name: 'agent-server-gateway', version: '0.1.0' })
|
|
|
- const readonlyBlock = () => (ctx.permissionMode === 'readonly' ? fail('当前为只读模式,不允许生成或修改文件') : null)
|
|
|
- const run = async (tool: string, args: unknown, fn: () => Promise<ToolResult>) => {
|
|
|
- const started = Date.now()
|
|
|
- try {
|
|
|
- const r = await fn()
|
|
|
- ctx.audit('gateway_call', `${tool}${r.isError ? '(失败)' : ''}`, { args, ms: Date.now() - started, engine: ctx.provider })
|
|
|
- return r
|
|
|
- } catch (err) {
|
|
|
- ctx.audit('gateway_call', `${tool}(异常)`, { args, error: (err as Error).message })
|
|
|
- return fail((err as Error).message)
|
|
|
+/** 引擎等待网关工具的时间(秒):交互超时 + 第三方工具最长执行时间 + 余量 */
|
|
|
+export const gatewayToolTimeoutSec = (s: AppSettings) => s.limits.interactionTimeoutSec + CALL_TIMEOUT_MS / 1000 + 60
|
|
|
+
|
|
|
+export interface GatewayDeps {
|
|
|
+ mcp: McpServerService
|
|
|
+ upstream: McpUpstream
|
|
|
+}
|
|
|
+
|
|
|
+/** 会话内"本会话允许"过的第三方工具 */
|
|
|
+const sessionAllowed = new Map<string, Set<string>>()
|
|
|
+
|
|
|
+function builtinTools(): Tool[] {
|
|
|
+ return BUILTIN_TOOLS.map(t => ({
|
|
|
+ name: t.name, description: t.description,
|
|
|
+ inputSchema: z.toJSONSchema(t.schema) as Tool['inputSchema'],
|
|
|
+ ...(t.readOnly ? { annotations: { readOnlyHint: true } } : {}),
|
|
|
+ }))
|
|
|
+}
|
|
|
+
|
|
|
+/** 这次对话可用的第三方工具(只读模式下只提供只读工具) */
|
|
|
+async function externalTools(ctx: GatewayRunContext, deps: GatewayDeps): Promise<{ row: McpServerRow; tool: McpCachedTool; exposed: string }[]> {
|
|
|
+ const out: { row: McpServerRow; tool: McpCachedTool; exposed: string }[] = []
|
|
|
+ for (const row of deps.mcp.rows(ctx.mcpServerIds)) {
|
|
|
+ let tools = row.toolsCache
|
|
|
+ if (!tools.length) {
|
|
|
+ tools = await Promise.race([
|
|
|
+ deps.mcp.listToolsFor(row).catch(() => [] as McpCachedTool[]),
|
|
|
+ new Promise<McpCachedTool[]>(r => setTimeout(() => r([]), LIST_TIMEOUT_MS)),
|
|
|
+ ])
|
|
|
+ }
|
|
|
+ for (const tool of tools) {
|
|
|
+ const policy = row.toolPolicies[tool.name] ?? defaultPolicy(tool)
|
|
|
+ if (policy === 'off' || (ctx.permissionMode === 'readonly' && !tool.readOnly)) continue
|
|
|
+ out.push({ row, tool, exposed: `${row.name}${MCP_SEP}${tool.name}`.slice(0, 64) })
|
|
|
}
|
|
|
}
|
|
|
+ return out
|
|
|
+}
|
|
|
|
|
|
- server.registerTool('convert_document', {
|
|
|
- description: '用 LibreOffice 转换文档格式(例如 docx→pdf、xlsx→pdf、pptx→pdf、doc→docx、html→pdf)。'
|
|
|
- + '输入和输出都必须在工作区内;输出文件名与输入相同、扩展名换成目标格式。不要在命令行里调用 soffice,它在沙箱中无法运行。',
|
|
|
- inputSchema: {
|
|
|
- input: z.string().describe('输入文件路径(相对工作区或绝对路径)'),
|
|
|
- format: z.enum(CONVERT_FORMATS).describe('目标格式'),
|
|
|
- out_dir: z.string().optional().describe('输出目录,默认 outputs'),
|
|
|
- },
|
|
|
- }, args => run('convert_document', args, async () => {
|
|
|
- const blocked = readonlyBlock()
|
|
|
- if (blocked) return blocked
|
|
|
- const out = await convertDocument({ workspaceDir: ctx.workspaceDir, input: args.input, format: args.format, outDir: args.out_dir })
|
|
|
- return ok(`已生成:${path.relative(ctx.workspaceDir, out)}`)
|
|
|
- }))
|
|
|
+function buildServer(ctx: GatewayRunContext, deps: GatewayDeps): Server {
|
|
|
+ const server = new Server({ name: 'agent-server-gateway', version: '0.1.0' }, { capabilities: { tools: {} } })
|
|
|
|
|
|
- server.registerTool('recalc_workbook', {
|
|
|
- description: '用 LibreOffice 重新计算 Excel 文件中的所有公式并写回原文件,然后报告公式数量和错误单元格(#REF!、#DIV/0! 等)。'
|
|
|
- + '用 openpyxl 写入公式后必须调用它,否则文件里没有计算结果。不要在命令行里调用 soffice 或 recalc.py,它们在沙箱中无法运行。',
|
|
|
- inputSchema: { input: z.string().describe('xlsx/xlsm 文件路径(相对工作区或绝对路径)') },
|
|
|
- }, args => run('recalc_workbook', args, async () => {
|
|
|
- const blocked = readonlyBlock()
|
|
|
- if (blocked) return blocked
|
|
|
- const r = await recalcWorkbook({ workspaceDir: ctx.workspaceDir, input: args.input, pythonRuntime: ctx.settings.paths.pythonRuntime })
|
|
|
- const lines = [`已重算:${path.relative(ctx.workspaceDir, r.path)},公式 ${r.formulas} 个,错误 ${r.errorCount} 个`]
|
|
|
- for (const e of r.errors) lines.push(`- ${e.sheet}!${e.cell}: ${e.error}`)
|
|
|
- return r.errorCount > 0 ? fail(lines.join('\n')) : ok(lines.join('\n'))
|
|
|
+ server.setRequestHandler(ListToolsRequestSchema, async () => ({
|
|
|
+ tools: [
|
|
|
+ ...builtinTools(),
|
|
|
+ ...(await externalTools(ctx, deps)).map(({ row, tool, exposed }) => ({
|
|
|
+ name: exposed,
|
|
|
+ description: `[${row.title}] ${tool.description}`.slice(0, 2000),
|
|
|
+ inputSchema: (tool.inputSchema?.type === 'object' ? tool.inputSchema : { type: 'object', properties: {} }) as Tool['inputSchema'],
|
|
|
+ ...(tool.readOnly ? { annotations: { readOnlyHint: true } } : {}),
|
|
|
+ })),
|
|
|
+ ],
|
|
|
}))
|
|
|
|
|
|
- server.registerTool('web_fetch', {
|
|
|
- description: '抓取网页并转换为 Markdown(HTML),或原样返回文本/JSON;PDF、图片、表格等二进制文件会下载到工作区并返回路径。'
|
|
|
- + '内容过长时只返回前一部分,完整内容保存在工作区。不能访问内网地址。',
|
|
|
- inputSchema: { url: z.string().describe('http/https 网址') },
|
|
|
- annotations: { readOnlyHint: true },
|
|
|
- }, args => run('web_fetch', args, async () => {
|
|
|
+ server.setRequestHandler(CallToolRequestSchema, async (req, extra) => {
|
|
|
+ const name = req.params.name
|
|
|
+ const args = (req.params.arguments ?? {}) as Record<string, unknown>
|
|
|
+ const started = Date.now()
|
|
|
+ const audit = (summary: string, detail: Record<string, unknown> = {}) =>
|
|
|
+ ctx.audit('gateway_call', summary, { args, ms: Date.now() - started, engine: ctx.provider, ...detail })
|
|
|
+
|
|
|
+ const builtin = BUILTIN_TOOLS.find(t => t.name === name)
|
|
|
+ if (builtin) {
|
|
|
+ const parsed = builtin.schema.safeParse(args)
|
|
|
+ if (!parsed.success) return fail(`参数不正确:${parsed.error.issues.map(i => `${i.path.join('.')} ${i.message}`).join(';')}`)
|
|
|
+ try {
|
|
|
+ const r = await builtin.handler(parsed.data as Record<string, unknown>, ctx)
|
|
|
+ audit(`${name}${r.isError ? '(失败)' : ''}`)
|
|
|
+ return r
|
|
|
+ } catch (err) {
|
|
|
+ audit(`${name}(异常)`, { error: (err as Error).message })
|
|
|
+ return fail((err as Error).message)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ const found = (await externalTools(ctx, deps)).find(t => t.exposed === name)
|
|
|
+ if (!found) return fail(`工具不存在或当前不可用:${name}`)
|
|
|
+ const { row, tool } = found
|
|
|
+ const policy = row.toolPolicies[tool.name] ?? defaultPolicy(tool)
|
|
|
+ // 需要审批:跟随权限模式(需审批时询问;自动模式直接调用;只读模式已在列表中排除非只读工具)
|
|
|
+ if (policy === 'ask' && ctx.permissionMode === 'ask' && !sessionAllowed.get(ctx.sessionId)?.has(name)) {
|
|
|
+ const outcome = await ctx.interact({
|
|
|
+ kind: 'approval', tool: `mcp__${GATEWAY_SERVER}__${name}`, title: `${row.title} · ${tool.name}`,
|
|
|
+ input: args, detail: { reason: tool.description.slice(0, 300) || undefined }, options: ['allow_once', 'allow_session', 'deny'],
|
|
|
+ })
|
|
|
+ ctx.audit('approval', `MCP 工具 ${name}:${outcome.resolution}`, { args })
|
|
|
+ if (outcome.resolution !== 'allowed') return fail(`用户拒绝了这次调用${outcome.resolution === 'denied' && outcome.reason ? `:${outcome.reason}` : ''}`)
|
|
|
+ if (outcome.scope === 'session') sessionAllowed.set(ctx.sessionId, new Set([...(sessionAllowed.get(ctx.sessionId) ?? []), name]))
|
|
|
+ }
|
|
|
try {
|
|
|
- const r = await webFetch({ url: args.url, workspaceDir: ctx.workspaceDir, blockPrivate: ctx.settings.network.blockPrivateFetch })
|
|
|
- const head = `URL: ${r.url}\n状态: ${r.status}\n类型: ${r.contentType}${r.savedPath ? `\n${r.truncated ? '完整内容' : '文件'}已保存到: ${r.savedPath}` : ''}${r.truncated ? '\n(内容过长,以下为前一部分)' : ''}\n\n`
|
|
|
- return r.status >= 400 ? fail(head + r.text) : ok(head + r.text)
|
|
|
+ const call = () => deps.upstream.callTool(row, tool.name, args, extra.signal, CALL_TIMEOUT_MS) as Promise<{ content?: ToolContent[]; structuredContent?: unknown; isError?: boolean }>
|
|
|
+ // 偶发的网络错误:只读工具重新连接后重试一次;其他工具不重试,避免重复执行
|
|
|
+ const r = await call().catch(async err => {
|
|
|
+ if (!tool.readOnly || !isTransient(err)) throw err
|
|
|
+ await deps.upstream.close(row.id)
|
|
|
+ return call()
|
|
|
+ })
|
|
|
+ const result = postProcess(r, ctx.workspaceDir, name)
|
|
|
+ audit(`${name}${result.isError ? '(失败)' : ''}`, { server: row.name })
|
|
|
+ return result
|
|
|
} catch (err) {
|
|
|
- if (err instanceof FetchBlockedError) ctx.audit('network_denied', err.message, { url: args.url, engine: ctx.provider })
|
|
|
- throw err
|
|
|
+ audit(`${name}(异常)`, { server: row.name, error: (err as Error).message })
|
|
|
+ return fail(`调用 ${row.title} 失败:${(err as Error).message}`)
|
|
|
}
|
|
|
- }))
|
|
|
-
|
|
|
- server.registerTool('kb_list_roots', {
|
|
|
- description: '列出当前可用的本地资料库(名称、路径、说明、读写属性)。查找本地资料前先调用它。',
|
|
|
- inputSchema: {},
|
|
|
- annotations: { readOnlyHint: true },
|
|
|
- }, () => run('kb_list_roots', {}, async () => {
|
|
|
- if (!ctx.knowledgeRoots.length) return ok('当前没有可用的资料库。')
|
|
|
- return ok(ctx.knowledgeRoots.map(k => `- ${k.name}:${k.path}(${k.access === 'write' ? '可读写' : '只读'})${k.description ? `\n ${k.description}` : ''}`).join('\n'))
|
|
|
- }))
|
|
|
+ })
|
|
|
|
|
|
return server
|
|
|
}
|
|
|
|
|
|
-export function registerGateway(f: FastifyInstance, tokens: RunTokenRegistry) {
|
|
|
+/** 只保留引擎能处理的内容;过长的文本和过大的二进制存入工作区 .agent/mcp/ 并返回路径 */
|
|
|
+function postProcess(r: { content?: ToolContent[]; structuredContent?: unknown; isError?: boolean }, workspaceDir: string, tool: string): ToolResult {
|
|
|
+ const save = (ext: string, data: string | Buffer) => {
|
|
|
+ const dir = path.join(workspaceDir, '.agent', 'mcp')
|
|
|
+ fs.mkdirSync(dir, { recursive: true })
|
|
|
+ const file = path.join(dir, `${tool.replace(/[^\w.-]/g, '_')}-${Date.now()}-${crypto.randomBytes(3).toString('hex')}.${ext}`)
|
|
|
+ fs.writeFileSync(file, data)
|
|
|
+ return path.relative(workspaceDir, file)
|
|
|
+ }
|
|
|
+ const content: ToolContent[] = []
|
|
|
+ for (const c of r.content ?? []) {
|
|
|
+ if (c.type === 'text' && typeof c.text === 'string' && c.text.length > MAX_TEXT) {
|
|
|
+ const saved = save('txt', c.text)
|
|
|
+ content.push({ type: 'text', text: `${c.text.slice(0, MAX_TEXT)}\n\n(内容过长,以上为前一部分;完整内容已保存到 ${saved})` })
|
|
|
+ } else if ((c.type === 'image' || c.type === 'audio') && typeof c.data === 'string' && c.data.length > MAX_INLINE_BINARY) {
|
|
|
+ const ext = String(c.mimeType ?? '').split('/')[1]?.replace(/[^\w]/g, '') || 'bin'
|
|
|
+ content.push({ type: 'text', text: `(${c.type === 'image' ? '图片' : '音频'}过大,已保存到 ${save(ext, Buffer.from(c.data, 'base64'))})` })
|
|
|
+ } else {
|
|
|
+ content.push(c)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (!content.length && r.structuredContent !== undefined) content.push({ type: 'text', text: JSON.stringify(r.structuredContent, null, 2).slice(0, MAX_TEXT) })
|
|
|
+ if (!content.length) content.push({ type: 'text', text: '(无返回内容)' })
|
|
|
+ return { content, ...(r.isError ? { isError: true } : {}) }
|
|
|
+}
|
|
|
+
|
|
|
+export function registerGateway(f: FastifyInstance, tokens: RunTokenRegistry, deps: GatewayDeps) {
|
|
|
f.post('/mcp', async (req, reply) => {
|
|
|
const auth = req.headers.authorization
|
|
|
const ctx = tokens.get(auth?.startsWith('Bearer ') ? auth.slice(7).trim() : undefined)
|
|
|
if (!ctx) return reply.code(401).send({ jsonrpc: '2.0', error: { code: -32001, message: '无效或已过期的运行令牌' }, id: null })
|
|
|
// 无状态模式:每个请求一个 server/transport 实例,绑定到该 Run 的上下文
|
|
|
- const server = buildServer(ctx)
|
|
|
+ const server = buildServer(ctx, deps)
|
|
|
const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: undefined })
|
|
|
reply.hijack()
|
|
|
reply.raw.on('close', () => {
|
|
|
@@ -106,3 +181,8 @@ export function registerGateway(f: FastifyInstance, tokens: RunTokenRegistry) {
|
|
|
f.get('/mcp', async (_req, reply) => reply.code(405).send(notAllowed))
|
|
|
f.delete('/mcp', async (_req, reply) => reply.code(405).send(notAllowed))
|
|
|
}
|
|
|
+
|
|
|
+/** 会话删除时清除"本会话允许" */
|
|
|
+export function forgetGatewaySession(sessionId: string) {
|
|
|
+ sessionAllowed.delete(sessionId)
|
|
|
+}
|