AgentServer/apps/server/test/mcp-test-policy.test.ts

// MCP 工具策略"仅后台测试"：只在后台的测试会话中提供（每次审批、不能在脚本中调用）；外部 API、Bot API、定时任务的会话中看不到也调不了
import fs from 'node:fs'
import http from 'node:http'
import type { AddressInfo } from 'node:net'
import path from 'node:path'
import pino from 'pino'
import { afterAll, beforeAll, describe, expect, it } from 'vitest'
import { Client } from '@modelcontextprotocol/sdk/client/index.js'
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'
import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'
import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js'
import { z } from 'zod'
import { createApp, type App } from '../src/app'
import { config } from '../src/config'
import { buildServer } from '../src/http/server'
import type { GatewayRunContext } from '../src/mcp-gateway/tokens'
import type { AgentProvider, RunContext } from '../src/providers/types'
import { resolveAccess } from '../src/services/access-policy'
import type { NewApproval } from '../src/services/interactions'

let app: App
let server: Awaited<ReturnType<typeof buildServer>>
let base: string
let upstream: http.Server
let serverId: string
const approvals: NewApproval[] = []

/** 假引擎：记录每次运行时网关上下文中的 adminTest */
const seen: (boolean | undefined)[] = []
const recorder: AgentProvider = {
  id: 'claude',
  async run(c: RunContext) {
    seen.push(app.gatewayTokens.get(c.gateway.token)?.adminTest)
    return { usage: { inputTokens: 1, outputTokens: 1, cacheReadTokens: 0, cacheWriteTokens: 0, costUsd: 0 } }
  },
  async status() {
    return { id: 'claude', enabled: true, available: true, loggedIn: true, account: 'test', cliPath: '', version: 'x', models: ['m'], defaultModel: 'm', message: null }
  },
}

beforeAll(async () => {
  const log = pino({ level: 'silent' })
  app = createApp(log, { providers: { claude: recorder } })
  server = await buildServer(app, { logger: log })
  base = await server.listen({ port: 0, host: '127.0.0.1' })
  upstream = http.createServer(async (req, res) => {
    const mcp = new McpServer({ name: 'fixture', version: '1.0.0' })
    mcp.registerTool('lookup', { description: '查询（已测完）', inputSchema: { q: z.string() }, annotations: { readOnlyHint: true } }, async ({ q }) => ({ content: [{ type: 'text', text: `found:${q}` }] }))
    mcp.registerTool('new_report', { description: '新报表（测试中）', inputSchema: {}, annotations: { readOnlyHint: true } }, async () => ({ content: [{ type: 'text', text: 'report' }] }))
    const t = new StreamableHTTPServerTransport({ sessionIdGenerator: undefined })
    res.on('close', () => { void t.close(); void mcp.close() })
    await mcp.connect(t)
    let body = ''
    req.on('data', c => (body += c))
    req.on('end', () => void t.handleRequest(req, res, body ? JSON.parse(body) : undefined))
  })
  const url = await new Promise<string>(r => upstream.listen(0, '127.0.0.1', () => r(`http://127.0.0.1:${(upstream.address() as AddressInfo).port}/mcp`)))
  const s = app.mcpServers.create({
    name: 'biz', title: '业务', description: '', transport: 'http', enabled: true, url, headers: [], command: '', args: [], env: [], allowNetwork: false, readPaths: [], writePaths: [],
  })
  serverId = s.id
  await app.mcpServers.refresh(s.id)
})
afterAll(async () => {
  await app.mcpUpstream.closeAll()
  await server.close()
  upstream.close()
})

function ctx(over: Partial<GatewayRunContext>): GatewayRunContext {
  const ws = path.join(config.dataDir, 'workspaces', `ses_tp_${Math.random().toString(36).slice(2)}`)
  fs.mkdirSync(ws, { recursive: true })
  const settings = app.settings.get()
  return {
    sessionId: `ses_${Math.random().toString(36).slice(2)}`, runId: 'r', provider: 'claude', workspaceDir: ws, permissionMode: 'auto', settings,
    access: resolveAccess({ policy: { mode: 'restricted', read: [], write: [], deny: [] }, knowledgeRoots: [], selectedKnowledgeRootIds: [], workspaceDir: ws, settings }),
    knowledgeRoots: [], mcpServerIds: [serverId], audit: () => {},
    interact: async req => {
      approvals.push(req as NewApproval)
      return { resolution: 'allowed', scope: 'session' }
    },
    ...over,
  }
}
async function gateway(c: GatewayRunContext) {
  const client = new Client({ name: 'test', version: '1' })
  await client.connect(new StreamableHTTPClientTransport(new URL(`${base}/mcp`), { requestInit: { headers: { Authorization: `Bearer ${app.gatewayTokens.issue(c)}` } } }))
  return client
}
const text = (r: unknown) => (r as { content: { text: string }[] }).content[0]!.text

describe('仅后台测试', () => {
  it('外部会话看不到也调不了；后台测试会话中提供、标注测试中，每次都要审批；不能在脚本中调用', async () => {
    const s = app.mcpServers.setToolPolicy(serverId, 'new_report', 'test')
    expect(s.tools.find(t => t.name === 'new_report')!.policy).toBe('test')

    // 外部 API / Bot API / 定时任务：adminTest 为 false
    const ext = await gateway(ctx({ adminTest: false }))
    const extNames = (await ext.listTools()).tools.map(t => t.name)
    expect(extNames).toContain('biz__lookup')
    expect(extNames).not.toContain('biz__new_report')
    const denied = await ext.callTool({ name: 'biz__new_report', arguments: {} })
    expect(denied.isError).toBe(true)
    expect(text(denied)).toMatch(/不存在或当前不可用/)
    expect(text(await ext.callTool({ name: 'biz__lookup', arguments: { q: 'x' } }))).toBe('found:x')
    await ext.close()

    // 后台测试会话：提供，并标注测试中；即使在自动模式、即使选择"本会话允许"，每次都要审批
    const admin = await gateway(ctx({ adminTest: true }))
    const tool = (await admin.listTools()).tools.find(t => t.name === 'biz__new_report')!
    expect(tool.description).toContain('[测试中]')
    approvals.length = 0
    expect(text(await admin.callTool({ name: 'biz__new_report', arguments: {} }))).toBe('report')
    expect(text(await admin.callTool({ name: 'biz__new_report', arguments: {} }))).toBe('report')
    expect(approvals).toHaveLength(2)
    expect(approvals[0]!.options).toEqual(['allow_once', 'deny'])
    expect(String(approvals[0]!.detail?.reason)).toContain('测试中')
    // 已测完的工具不受影响（只读、自动模式直接调用）
    expect(text(await admin.callTool({ name: 'biz__lookup', arguments: { q: 'y' } }))).toBe('found:y')
    expect(approvals).toHaveLength(2)

    // 脚本中不能调用测试中的工具（在执行脚本前拒绝）
    const script = await admin.callTool({ name: 'run_script', arguments: { code: 'print(1)', tools: ['biz__new_report'], description: '试' } })
    expect(script.isError).toBe(true)
    expect(text(script)).toMatch(/还在测试中/)
    await admin.close()

    // 测完后改回其他策略：外部会话可用
    app.mcpServers.setToolPolicy(serverId, 'new_report', 'auto')
    const ext2 = await gateway(ctx({ adminTest: false }))
    expect((await ext2.listTools()).tools.map(t => t.name)).toContain('biz__new_report')
    await ext2.close()
  })
})

describe('哪些会话是后台测试会话', () => {
  it('后台发起的会话是；外部 API Key 的会话、定时任务触发的运行不是', async () => {
    const builtin = app.profiles.getBuiltin().id
    const key = app.apiKeys.create('平台A', { allowedProfileIds: [builtin], defaultProfileId: builtin }).apiKey
    const run = async (source: string, unattended = false) => {
      const s = await app.sessions.create(source, {})
      const n = seen.length
      app.runs.start(app.sessions.getRow(s.id), '你好', [], { unattended })
      for (let i = 0; i < 1000 && seen.length === n; i++) await new Promise(r => setTimeout(r, 10))
      return seen[n]
    }
    expect(await run('admin')).toBe(true)
    expect(await run(key.id)).toBe(false)
    expect(await run('admin', true)).toBe(false)
  })
})