AgentServer/apps/server/src/services/mcp-servers.ts

// 第三方 MCP 服务的登记与管理（ADR-0021）。连接由 mcp-gateway/upstream.ts 管理，运行时经 MCP Gateway 转接
import fs from 'node:fs'
import path from 'node:path'
import { eq } from 'drizzle-orm'
import { z } from 'zod'
import type { AppSettings, McpSecretEntry, McpServerDto, McpServerInput, McpStateDto, McpTestResultDto, McpToolDto, McpToolHints, McpToolPolicy } from '@agent-server/shared'
import { config } from '../config'
import type { DB } from '../db'
import { mcpServers, profiles, type McpCachedTool, type McpFeatures, type McpOAuthState } from '../db/schema'
import { badRequest, notFound } from '../lib/errors'
import { newId, nowIso } from '../lib/ids'
import { isWithin, realpathLoose } from '../lib/paths'
import { BUILTIN_TOOL_INFO } from '../mcp-gateway/builtin'
import { BWRAP, mcpHomeDir, McpUpstream, type McpServerRow } from '../mcp-gateway/upstream'
import { globalDenyList } from './access-policy'

const NAME_RE = /^[a-z0-9][a-z0-9_-]{0,31}$/
const ENV_RE = /^[A-Za-z_][A-Za-z0-9_]*$/
const HEADER_RE = /^[A-Za-z0-9!#$%&'*+.^_`|~-]+$/

const secretSchema = z.object({ name: z.string().min(1).max(200), value: z.string().max(20_000).nullable() })
const inputSchema = z.object({
  name: z.string().regex(NAME_RE, '名称只能用小写字母、数字、下划线、连字符（1-32 位，以字母或数字开头）'),
  title: z.string().trim().min(1).max(60),
  description: z.string().max(1000).default(''),
  transport: z.enum(['http', 'stdio']),
  enabled: z.boolean().default(true),
  url: z.string().max(2000).default(''),
  headers: z.array(secretSchema).max(50).default([]),
  oauth: z.object({
    enabled: z.boolean(), clientId: z.string().trim().max(500).default(''), clientSecret: z.string().max(2000).nullable().default(null), scope: z.string().trim().max(1000).default(''),
  }).optional(),
  command: z.string().max(500).default(''),
  args: z.array(z.string().max(2000)).max(100).default([]),
  env: z.array(secretSchema).max(100).default([]),
  allowNetwork: z.boolean().default(false),
  readPaths: z.array(z.string()).max(50).default([]),
  writePaths: z.array(z.string()).max(50).default([]),
})

export interface EffectiveHints {
  readOnly: boolean
  /** 只认明确声明（或管理员设置）的 true：规范的默认值 true 不能用来强制审批 */
  destructive: boolean
  idempotent: boolean
  openWorld: boolean
}

/** 生效的工具提示：管理员的覆盖优先，其次是服务的声明（docs/design/16-mcp-support.md §2.1） */
export function effectiveHints(t: McpCachedTool, o: McpToolHints = {}): EffectiveHints {
  const readOnly = o.readOnly ?? t.readOnly
  return {
    readOnly,
    destructive: !readOnly && (o.destructive ?? t.annotations?.destructiveHint === true),
    idempotent: readOnly || (o.idempotent ?? t.annotations?.idempotentHint === true),
    openWorld: t.annotations?.openWorldHint !== false,
  }
}

/** 没有单独设置时：只读工具直接调用；明确声明破坏性的总是审批；其余跟随权限模式 */
export const defaultPolicy = (h: Pick<EffectiveHints, 'readOnly' | 'destructive'>): McpToolPolicy => (h.readOnly ? 'auto' : h.destructive ? 'confirm' : 'ask')

/** 工具的生效提示与策略 */
export function toolState(r: Pick<McpServerRow, 'toolHints' | 'toolPolicies'>, t: McpCachedTool): { hints: EffectiveHints; policy: McpToolPolicy } {
  const hints = effectiveHints(t, r.toolHints[t.name])
  return { hints, policy: r.toolPolicies[t.name] ?? defaultPolicy(hints) }
}

export class McpServerService {
  private refreshing = new Map<string, NodeJS.Timeout>()

  constructor(private db: DB, private upstream: McpUpstream, private settings: () => AppSettings, private audit: (summary: string) => void) {
    // 服务通知工具列表变化，或新建了连接：重新列出工具（合并 2 秒内的多次通知，docs/design/16-mcp-support.md §2.3）
    upstream.onToolsChanged = id => {
      clearTimeout(this.refreshing.get(id))
      this.refreshing.set(id, setTimeout(() => {
        this.refreshing.delete(id)
        const r = this.db.select().from(mcpServers).where(eq(mcpServers.id, id)).get()
        if (r?.enabled) void this.listToolsFor(r).catch(() => {})
      }, 2000))
    }
  }

  state(): McpStateDto {
    return { builtin: BUILTIN_TOOL_INFO, servers: this.list(), sandboxAvailable: !!BWRAP }
  }

  list(): McpServerDto[] {
    const usedBy = this.usage()
    return this.db.select().from(mcpServers).orderBy(mcpServers.name).all().map(r => this.toDto(r, usedBy.get(r.id) ?? []))
  }

  get(id: string): McpServerDto {
    return this.toDto(this.row(id), this.usage().get(id) ?? [])
  }

  /** 运行时：取 Profile 勾选且已启用的服务 */
  rows(ids: string[]): McpServerRow[] {
    if (!ids.length) return []
    return this.db.select().from(mcpServers).all().filter(r => r.enabled && ids.includes(r.id))
  }

  create(input: McpServerInput): McpServerDto {
    const data = this.validate(input, null)
    if (this.db.select().from(mcpServers).where(eq(mcpServers.name, data.name)).get()) throw badRequest(`名称 ${data.name} 已被使用`)
    const id = newId('mcp')
    this.db.insert(mcpServers).values({ id, ...data }).run()
    this.audit(`添加 MCP 服务 ${data.name}（${data.transport === 'http' ? data.url : data.command}）`)
    return this.get(id)
  }

  update(id: string, input: McpServerInput): McpServerDto {
    const prev = this.row(id)
    const data = this.validate(input, prev)
    const dup = this.db.select().from(mcpServers).where(eq(mcpServers.name, data.name)).get()
    if (dup && dup.id !== id) throw badRequest(`名称 ${data.name} 已被使用`)
    this.db.update(mcpServers).set({ ...data, updatedAt: nowIso() }).where(eq(mcpServers.id, id)).run()
    this.audit(`修改 MCP 服务 ${data.name}`)
    return this.get(id)
  }

  async delete(id: string) {
    const r = this.row(id)
    // 从引用它的 Profile 中移除
    for (const p of this.db.select().from(profiles).all()) {
      if (p.mcpServerIds.includes(id)) this.db.update(profiles).set({ mcpServerIds: p.mcpServerIds.filter(x => x !== id) }).where(eq(profiles.id, p.id)).run()
    }
    this.db.delete(mcpServers).where(eq(mcpServers.id, id)).run()
    await this.upstream.close(id)
    fs.rmSync(mcpHomeDir(id), { recursive: true, force: true })
    this.audit(`删除 MCP 服务 ${r.name}`)
  }

  /** 重新连接并更新工具列表 */
  async refresh(id: string): Promise<McpServerDto> {
    const r = this.row(id)
    await this.upstream.close(id)
    try {
      await this.listToolsFor(r)
    } catch {
      // 错误记录在连接状态中，随 DTO 返回
    }
    return this.get(id)
  }

  async listToolsFor(r: McpServerRow): Promise<McpCachedTool[]> {
    const tools = await this.upstream.listTools(r)
    const features = await this.upstream.features(r).catch(() => r.features)
    this.updateTools(r, tools, features)
    return tools
  }

  /** 总是在后台执行（docs/design/15-background-calls.md §3） */
  setToolBackground(id: string, tool: string, background: boolean): McpServerDto {
    const r = this.row(id)
    const list = r.backgroundTools.filter(t => t !== tool)
    this.db.update(mcpServers).set({ backgroundTools: background ? [...list, tool] : list, updatedAt: nowIso() }).where(eq(mcpServers.id, id)).run()
    this.audit(`MCP 工具 ${r.name}__${tool} ${background ? '设为总是后台执行' : '取消总是后台执行'}`)
    return this.get(id)
  }

  setToolPolicy(id: string, tool: string, policy: McpToolPolicy): McpServerDto {
    const r = this.row(id)
    this.db.update(mcpServers).set({ toolPolicies: { ...r.toolPolicies, [tool]: policy }, updatedAt: nowIso() }).where(eq(mcpServers.id, id)).run()
    this.audit(`MCP 工具 ${r.name}__${tool} 设为 ${({ auto: '直接调用', ask: '跟随权限模式', confirm: '总是审批', off: '关闭' } as const)[policy]}`)
    return this.get(id)
  }

  /** 覆盖工具提示；某一项为 undefined 表示沿用服务的声明 */
  setToolHints(id: string, tool: string, hints: McpToolHints): McpServerDto {
    const r = this.row(id)
    const clean: McpToolHints = {}
    for (const k of ['readOnly', 'destructive', 'idempotent'] as const) if (typeof hints[k] === 'boolean') clean[k] = hints[k]
    const next = { ...r.toolHints }
    if (Object.keys(clean).length) next[tool] = clean
    else delete next[tool]
    this.db.update(mcpServers).set({ toolHints: next, updatedAt: nowIso() }).where(eq(mcpServers.id, id)).run()
    const label = { readOnly: '只读', destructive: '破坏性', idempotent: '幂等' } as const
    const desc = Object.entries(clean).map(([k, v]) => `${label[k as keyof McpToolHints]}=${v ? '是' : '否'}`).join('、') || '沿用服务的声明'
    this.audit(`MCP 工具 ${r.name}__${tool} 的提示：${desc}`)
    return this.get(id)
  }

  /** 测试一份配置（新建前或修改前）；已登记的服务可以只传新值，未修改的密钥沿用原值 */
  async test(input: McpServerInput, id: string | null): Promise<McpTestResultDto> {
    const prev = id ? this.row(id) : null
    const data = this.validate(input, prev, true)
    const t0 = Date.now()
    const tempId = id ?? `test_${Date.now()}`
    try {
      const { tools, stderr } = await this.upstream.test({ ...(prev ?? emptyRow()), ...data, id: tempId })
      return { ok: true, tools: tools.map(t => toToolDto(t, prev ?? { toolPolicies: {}, toolHints: {}, backgroundTools: [] })), error: null, stderrTail: stderr.length ? stderr.join('\n') : null, ms: Date.now() - t0 }
    } catch (err) {
      return { ok: false, tools: [], error: (err as Error).message, stderrTail: null, ms: Date.now() - t0 }
    } finally {
      // 未登记的服务测试时使用的临时数据目录
      if (!id) fs.rmSync(mcpHomeDir(tempId), { recursive: true, force: true })
    }
  }

  /**
   * 工具列表或能力有变化时才更新（toolsUpdatedAt 是 Codex 会话的工具版本之一，变化会触发分叉，ADR-0020）；
   * 第一次列出时也记录时间
   */
  private updateTools(r: McpServerRow, tools: McpCachedTool[], features: McpFeatures) {
    const cur = this.db.select({ tools: mcpServers.toolsCache, features: mcpServers.features, at: mcpServers.toolsUpdatedAt }).from(mcpServers).where(eq(mcpServers.id, r.id)).get()
    if (cur?.at && JSON.stringify(cur.tools) === JSON.stringify(tools) && JSON.stringify(cur.features) === JSON.stringify(features)) return
    this.db.update(mcpServers).set({ toolsCache: tools, features, toolsUpdatedAt: nowIso() }).where(eq(mcpServers.id, r.id)).run()
  }

  private row(id: string): McpServerRow {
    const r = this.db.select().from(mcpServers).where(eq(mcpServers.id, id)).get()
    if (!r) throw notFound('MCP 服务')
    return r
  }

  private usage(): Map<string, string[]> {
    const out = new Map<string, string[]>()
    for (const p of this.db.select({ name: profiles.name, ids: profiles.mcpServerIds }).from(profiles).all()) {
      for (const id of p.ids) out.set(id, [...(out.get(id) ?? []), p.name])
    }
    return out
  }

  private toDto(r: McpServerRow, profileNames: string[]): McpServerDto {
    const masked = (rec: Record<string, string>): McpSecretEntry[] => Object.keys(rec).map(name => ({ name, value: null, isSet: rec[name] !== '' }))
    return {
      id: r.id, name: r.name, title: r.title, description: r.description, transport: r.transport, enabled: r.enabled,
      url: r.url, headers: masked(r.headers),
      oauth: r.oauth ? { enabled: r.oauth.enabled, clientId: r.oauth.clientId, clientSecret: null, scope: r.oauth.scope } : { enabled: false, clientId: '', clientSecret: null, scope: '' },
      oauthStatus: r.oauth?.enabled ? {
        authorized: !!r.oauth.tokens, authorizedAt: r.oauth.authorizedAt ?? null, clientSecretSet: !!r.oauth.clientSecret,
        registered: !!(r.oauth.clientId || r.oauth.clientInfo), lastError: r.oauth.lastError ?? null,
      } : null,
      command: r.command, args: r.args, env: masked(r.env),
      allowNetwork: r.allowNetwork, readPaths: r.readPaths, writePaths: r.writePaths,
      tools: r.toolsCache.map(t => toToolDto(t, r)), toolsUpdatedAt: r.toolsUpdatedAt, features: { resources: !!r.features.resources, prompts: !!r.features.prompts },
      ...this.upstream.status(r.id), profileNames, createdAt: r.createdAt, updatedAt: r.updatedAt,
    }
  }

  private validate(input: McpServerInput, prev: McpServerRow | null, forTest = false) {
    const r = inputSchema.safeParse(input)
    if (!r.success) {
      const i = r.error.issues[0]!
      throw badRequest(`${i.path.length ? `${i.path.join('.')}：` : ''}${i.message}`)
    }
    const d = r.data
    if (d.transport === 'http') {
      let u: URL
      try {
        u = new URL(d.url)
      } catch {
        throw badRequest('服务地址不是有效的 URL')
      }
      if (!['http:', 'https:'].includes(u.protocol)) throw badRequest('服务地址只支持 http / https')
    } else {
      if (!d.command.trim()) throw badRequest('请填写启动命令')
      if (!BWRAP && !forTest) throw badRequest('服务器上没有 bubblewrap（bwrap），无法运行本地 MCP 服务')
    }
    for (const h of d.headers) if (!HEADER_RE.test(h.name)) throw badRequest(`请求头名称不正确：${h.name}`)
    for (const e of d.env) if (!ENV_RE.test(e.name)) throw badRequest(`环境变量名称不正确：${e.name}`)
    return {
      name: d.name, title: d.title, description: d.description, transport: d.transport, enabled: d.enabled,
      url: d.transport === 'http' ? d.url.trim() : '',
      headers: d.transport === 'http' ? mergeSecrets(d.headers, prev?.headers ?? {}) : {},
      oauth: d.transport === 'http' ? mergeOAuth(d.oauth, prev?.oauth ?? null) : null,
      command: d.transport === 'stdio' ? d.command.trim() : '',
      args: d.transport === 'stdio' ? d.args : [],
      env: d.transport === 'stdio' ? mergeSecrets(d.env, prev?.env ?? {}) : {},
      allowNetwork: d.transport === 'stdio' && d.allowNetwork,
      readPaths: d.transport === 'stdio' ? this.checkPaths(d.readPaths, 'read') : [],
      writePaths: d.transport === 'stdio' ? this.checkPaths(d.writePaths, 'write') : [],
    }
  }

  /** 沙箱中额外开放的路径：必须是存在的绝对路径；不能是全局禁止项、服务数据目录或会话工作区 */
  private checkPaths(paths: string[], kind: 'read' | 'write'): string[] {
    const s = this.settings()
    const deny = globalDenyList(s)
    return paths.map(p => p.trim()).filter(Boolean).map(p => {
      if (!path.isAbsolute(p)) throw badRequest(`路径必须是绝对路径：${p}`)
      const real = realpathLoose(p)
      if (!fs.existsSync(real)) throw badRequest(`路径不存在：${p}`)
      if (real === '/') throw badRequest('不能开放根目录')
      for (const protectedDir of [config.dataDir, s.paths.workspaceRoot]) {
        if (isWithin(real, realpathLoose(protectedDir)) || isWithin(realpathLoose(protectedDir), real)) throw badRequest(`不能开放服务数据目录或会话工作区：${p}`)
      }
      const hit = deny.find(d => !d.includes('*') && (isWithin(real, d) || isWithin(d, real)))
      if (hit) throw badRequest(`${kind === 'write' ? '可写' : '只读'}路径与全局禁止项冲突：${hit}`)
      return real
    })
  }
}

/** OAuth 配置：密钥为 null 时沿用原值；Client ID 变化时清除注册信息与令牌 */
function mergeOAuth(input: { enabled: boolean; clientId: string; clientSecret: string | null; scope: string } | undefined, prev: McpOAuthState | null): McpOAuthState | null {
  if (!input) return prev
  const clientSecret = input.clientSecret ?? (prev && prev.clientId === input.clientId ? prev.clientSecret : '')
  const same = prev && prev.clientId === input.clientId && prev.scope === input.scope
  return { ...(same ? prev : {}), enabled: input.enabled, clientId: input.clientId, clientSecret, scope: input.scope }
}

/** 密钥：value 为 null 表示保留原值 */
function mergeSecrets(entries: { name: string; value: string | null }[], prev: Record<string, string>): Record<string, string> {
  const out: Record<string, string> = {}
  for (const e of entries) {
    const v = e.value ?? prev[e.name]
    if (v === undefined) throw badRequest(`请填写 ${e.name} 的值`)
    out[e.name] = v
  }
  return out
}

function toToolDto(t: McpCachedTool, r: Pick<McpServerRow, 'toolPolicies' | 'toolHints' | 'backgroundTools'>): McpToolDto {
  const { hints, policy } = toolState(r, t)
  const a = t.annotations ?? {}
  return {
    name: t.name, title: t.title ?? null, description: t.description, ...hints,
    declared: { readOnly: a.readOnlyHint ?? (t.readOnly || null), destructive: a.destructiveHint ?? null, idempotent: a.idempotentHint ?? null, openWorld: a.openWorldHint ?? null },
    overrides: r.toolHints[t.name] ?? {}, policy, background: r.backgroundTools.includes(t.name),
    hasOutputSchema: !!t.outputSchema, taskSupport: t.taskSupport ?? null,
  }
}

function emptyRow(): McpServerRow {
  const now = nowIso()
  return {
    id: '', name: '', title: '', description: '', transport: 'http', enabled: true, url: '', headers: {}, command: '', args: [], env: {},
    allowNetwork: false, readPaths: [], writePaths: [], toolPolicies: {}, toolHints: {}, backgroundTools: [], oauth: null, features: {}, toolsCache: [], toolsUpdatedAt: null, createdAt: now, updatedAt: now,
  }
}