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

// MCP 支持完善 · 阶段 A：工具提示、结构化输出、工具列表变化（docs/design/16-mcp-support.md §2）
import crypto from 'node:crypto'
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 type { McpServerInput } from '@agent-server/shared'
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 { resolveAccess } from '../src/services/access-policy'
import type { InteractionOutcome, NewApproval } from '../src/services/interactions'

let app: App
let server: Awaited<ReturnType<typeof buildServer>>
let base: string
let upstream: http.Server
let upstreamUrl: string
let flakyCalls = 0
const servers = new Set<McpServer>()

/** 有状态的测试服务（会话 id），可以发出工具列表变化通知 */
function startUpstream(): Promise<string> {
  const transports = new Map<string, StreamableHTTPServerTransport>()
  const build = () => {
    const mcp = new McpServer({ name: 'fixture-hints', version: '1.0.0' })
    mcp.registerTool('delete_all', { title: '删除全部', description: '删除全部数据', inputSchema: {}, annotations: { destructiveHint: true } }, async () => ({ content: [{ type: 'text', text: 'deleted' }] }))
    mcp.registerTool('set_value', { description: '设置值', inputSchema: { v: z.string() }, annotations: { idempotentHint: true, destructiveHint: false, openWorldHint: false } }, async ({ v }) => ({ content: [{ type: 'text', text: `set:${v}` }] }))
    mcp.registerTool('send_once', { description: '发送（不幂等）', inputSchema: {} }, async () => ({ content: [{ type: 'text', text: 'sent' }] }))
    mcp.registerTool('stats', {
      description: '统计', inputSchema: {}, annotations: { readOnlyHint: true },
      outputSchema: { total: z.number(), items: z.array(z.string()) },
    }, async () => {
      const structuredContent = { total: 2, items: ['a', 'b'] }
      return { content: [{ type: 'text', text: JSON.stringify(structuredContent) }], structuredContent }
    })
    mcp.registerTool('stats_summary', { description: '统计（文本与结构化不同）', inputSchema: {}, annotations: { readOnlyHint: true } }, async () => ({
      content: [{ type: 'text', text: '共 2 条' }], structuredContent: { total: 2 },
    }))
    // 服务端自己的校验接受任何值，对外声明的 JSON Schema 却是 number：模拟返回结构与声明不符的服务
    mcp.registerTool('stats_bad', { description: '结构不符', inputSchema: {}, annotations: { readOnlyHint: true }, outputSchema: { total: z.any().meta({ type: 'number' }) } }, async () => ({
      content: [{ type: 'text', text: '坏结果' }], structuredContent: { total: 'two' },
    }))
    servers.add(mcp)
    return mcp
  }
  upstream = http.createServer(async (req, res) => {
    let body = ''
    req.on('data', c => (body += c))
    await new Promise(r => req.on('end', r))
    const parsed = body ? JSON.parse(body) : undefined
    // 偶发的网络错误：调用 set_value / send_once 的第一次直接断开连接
    if (parsed?.method === 'tools/call' && ['set_value', 'send_once'].includes(parsed.params.name) && flakyCalls++ % 2 === 0) return req.socket.destroy()
    const sid = req.headers['mcp-session-id'] as string | undefined
    let t = sid ? transports.get(sid) : undefined
    if (!t) {
      t = new StreamableHTTPServerTransport({ sessionIdGenerator: () => crypto.randomUUID(), onsessioninitialized: id => void transports.set(id, t!) })
      await build().connect(t)
    }
    await t.handleRequest(req, res, parsed)
  })
  return new Promise(r => upstream.listen(0, '127.0.0.1', () => r(`http://127.0.0.1:${(upstream.address() as AddressInfo).port}/mcp`)))
}

const input = (): McpServerInput => ({
  name: 'hints', title: '提示服务', description: '', transport: 'http', enabled: true, url: upstreamUrl,
  headers: [], command: '', args: [], env: [], allowNetwork: false, readPaths: [], writePaths: [],
})

const approvals: NewApproval[] = []
const audits: { summary: string; detail: Record<string, unknown> }[] = []
const answer: InteractionOutcome = { resolution: 'allowed', scope: 'once' }
function runCtx(over: Partial<GatewayRunContext> = {}): GatewayRunContext {
  const ws = path.join(config.dataDir, 'workspaces', `ses_hints_${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: [], audit: (_k, summary, detail) => void audits.push({ summary, detail: detail as Record<string, unknown> }),
    interact: async req => { approvals.push(req as NewApproval); return answer },
    ...over,
  }
}
async function gateway(ctx: 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(ctx)}` } } }))
  return client
}
const texts = (r: unknown) => (r as { content: { text: string }[] }).content.map(c => c.text)

beforeAll(async () => {
  const log = pino({ level: 'silent' })
  app = createApp(log, { providers: {} })
  server = await buildServer(app, { logger: log })
  base = await server.listen({ port: 0, host: '127.0.0.1' })
  upstreamUrl = await startUpstream()
})
afterAll(async () => {
  await app.mcpUpstream.closeAll()
  await server.close()
  upstream.closeAllConnections()
  upstream.close()
})

describe('工具提示', () => {
  it('明确声明破坏性的工具默认总是审批（自动模式也问、每次都问）；管理员可以覆盖提示', async () => {
    const s = app.mcpServers.create(input())
    const tools = (await app.mcpServers.refresh(s.id)).tools
    const del = tools.find(t => t.name === 'delete_all')!
    expect(del).toMatchObject({ title: '删除全部', destructive: true, policy: 'confirm', declared: { destructive: true, readOnly: null } })
    // 没写提示的工具：规范默认值不用来强制审批
    expect(tools.find(t => t.name === 'send_once')).toMatchObject({ destructive: false, idempotent: false, openWorld: true, policy: 'ask' })
    expect(tools.find(t => t.name === 'set_value')).toMatchObject({ idempotent: true, openWorld: false, policy: 'ask' })
    expect(tools.find(t => t.name === 'stats')).toMatchObject({ readOnly: true, policy: 'auto', hasOutputSchema: true })

    const ctx = runCtx({ mcpServerIds: [s.id] })
    const c = await gateway(ctx)
    const listed = (await c.listTools()).tools.find(t => t.name === 'hints__delete_all')!
    expect(listed.annotations).toMatchObject({ destructiveHint: true, readOnlyHint: false })
    expect(listed.title).toBe('删除全部')
    expect(texts(await c.callTool({ name: 'hints__delete_all', arguments: {} }))).toEqual(['deleted'])
    expect(texts(await c.callTool({ name: 'hints__delete_all', arguments: {} }))).toEqual(['deleted'])
    expect(approvals).toHaveLength(2)
    expect(approvals[0]).toMatchObject({ options: ['allow_once', 'deny'], title: '提示服务 · 删除全部' })
    expect(approvals[0]!.detail?.reason).toMatch(/破坏性/)

    // 覆盖为非破坏性：回到跟随权限模式，自动模式下不再询问
    const after = app.mcpServers.setToolHints(s.id, 'delete_all', { destructive: false })
    expect(after.tools.find(t => t.name === 'delete_all')).toMatchObject({ destructive: false, policy: 'ask', overrides: { destructive: false } })
    await c.callTool({ name: 'hints__delete_all', arguments: {} })
    expect(approvals).toHaveLength(2)
    // 覆盖为只读：只读模式下也提供
    app.mcpServers.setToolHints(s.id, 'send_once', { readOnly: true })
    const ro = await gateway(runCtx({ mcpServerIds: [s.id], permissionMode: 'readonly' }))
    expect((await ro.listTools()).tools.map(t => t.name)).toEqual(expect.arrayContaining(['hints__send_once', 'hints__stats']))
    await ro.close()
    // 清除覆盖
    expect(app.mcpServers.setToolHints(s.id, 'send_once', {}).tools.find(t => t.name === 'send_once')!.overrides).toEqual({})
    await c.close()
  })

  it('网络错误：幂等工具重新连接后重试，不幂等的不重试', async () => {
    const s = app.mcpServers.list().find(x => x.name === 'hints')!
    const c = await gateway(runCtx({ mcpServerIds: [s.id] }))
    flakyCalls = 0
    expect(texts(await c.callTool({ name: 'hints__set_value', arguments: { v: '1' } }))).toEqual(['set:1'])
    flakyCalls = 0
    const r = await c.callTool({ name: 'hints__send_once', arguments: {} })
    expect(r.isError).toBe(true)
    expect(flakyCalls).toBe(1)
    await c.close()
  })
})

describe('结构化输出', () => {
  it('文本已含同样的 JSON 时不重复；不同时附在后面；不符合 outputSchema 时照常返回并记录', async () => {
    const s = app.mcpServers.list().find(x => x.name === 'hints')!
    const ctx = runCtx({ mcpServerIds: [s.id] })
    const c = await gateway(ctx)
    const listed = (await c.listTools()).tools.find(t => t.name === 'hints__stats')!
    expect(listed.description).toMatch(/返回结构（JSON Schema）：.*"total"/)
    expect(listed.outputSchema).toBeUndefined()

    expect(texts(await c.callTool({ name: 'hints__stats', arguments: {} }))).toEqual(['{"total":2,"items":["a","b"]}'])
    expect(texts(await c.callTool({ name: 'hints__stats_summary', arguments: {} }))).toEqual(['共 2 条', '结构化结果：\n{\n  "total": 2\n}'])

    audits.length = 0
    const bad = await c.callTool({ name: 'hints__stats_bad', arguments: {} })
    expect(bad.isError).toBeFalsy()
    expect(texts(bad)[0]).toBe('坏结果')
    expect(audits.at(-1)!.detail).toMatchObject({ structured: { total: 'two' }, structuredError: expect.stringMatching(/outputSchema/) })
    await c.close()
  })
})

describe('工具列表变化', () => {
  it('服务通知工具列表变化后自动刷新缓存', async () => {
    const s = app.mcpServers.list().find(x => x.name === 'hints')!
    const c = await gateway(runCtx({ mcpServerIds: [s.id] }))
    await c.callTool({ name: 'hints__stats', arguments: {} }) // 建立上游连接
    await new Promise(r => setTimeout(r, 2500))
    for (const mcp of servers) {
      if (mcp.isConnected()) mcp.registerTool('added_later', { description: '后来加的', inputSchema: {} }, async () => ({ content: [{ type: 'text', text: 'new' }] }))
    }
    await expect.poll(() => app.mcpServers.get(s.id).tools.map(t => t.name), { timeout: 8000 }).toContain('added_later')
    expect((await c.listTools()).tools.map(t => t.name)).toContain('hints__added_later')
    await c.close()
  }, 15_000)
})