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

// MCP 支持完善 · 阶段 C：外部服务向用户询问信息（Elicitation，docs/design/16-mcp-support.md §4）
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 { z } from 'zod'
import { Client } from '@modelcontextprotocol/sdk/client/index.js'
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'
import { InMemoryTaskMessageQueue, InMemoryTaskStore } from '@modelcontextprotocol/sdk/experimental/tasks/stores/in-memory.js'
import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'
import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js'
import type { ElicitResult } from '@modelcontextprotocol/sdk/types.js'
import type { InteractionFormRequest } from '@agent-server/shared'
import { createApp, type App } from '../src/app'
import { config } from '../src/config'
import { schemaToFields } from '../src/mcp-gateway/elicitation'
import { buildServer } from '../src/http/server'
import { taskTiming } from '../src/mcp-gateway/tasks'
import type { GatewayRunContext } from '../src/mcp-gateway/tokens'
import { resolveAccess } from '../src/services/access-policy'
import { validateFormContent, type InteractionOutcome, type NewForm } from '../src/services/interactions'

let app: App
let server: Awaited<ReturnType<typeof buildServer>>
let base: string
let upstream: http.Server
let serverId: string

const SCHEMA = {
  type: 'object' as const,
  properties: {
    name: { type: 'string' as const, title: '收件人', minLength: 1 },
    email: { type: 'string' as const, title: '邮箱', format: 'email' as const },
    count: { type: 'integer' as const, title: '数量', minimum: 1, maximum: 10, default: 1 },
    urgent: { type: 'boolean' as const, title: '加急' },
    color: { type: 'string' as const, title: '颜色', oneOf: [{ const: 'red', title: '红' }, { const: 'blue', title: '蓝' }] },
    tags: { type: 'array' as const, title: '标签', items: { type: 'string' as const, enum: ['a', 'b', 'c'] }, maxItems: 2 },
  },
  required: ['name', 'color'],
}

/** 有状态的测试服务：调用过程中询问用户；长任务中询问用户（任务按 MCP 会话归属） */
const taskStore = new InMemoryTaskStore()
function startUpstream(): Promise<string> {
  const transports = new Map<string, StreamableHTTPServerTransport>()
  const build = () => {
    const mcp = new McpServer({ name: 'ask', version: '1.0.0' }, { capabilities: { tasks: { requests: { tools: { call: {} } } } }, taskStore, taskMessageQueue: new InMemoryTaskMessageQueue() })
    mcp.registerTool('order', { description: '下单（会询问收件信息）', inputSchema: {}, annotations: { readOnlyHint: true } }, async () => {
      const r = await mcp.server.elicitInput({ mode: 'form', message: '请填写收件信息', requestedSchema: SCHEMA })
      return { content: [{ type: 'text', text: `${r.action}:${JSON.stringify(r.content ?? {})}` }] }
    })
    mcp.registerTool('login', { description: '需要授权', inputSchema: {}, annotations: { readOnlyHint: true } }, async () => {
      const r = await mcp.server.elicitInput({ mode: 'url', message: '请先授权访问日历', url: 'https://auth.example.com/authorize?x=1', elicitationId: 'el1' })
      return { content: [{ type: 'text', text: r.action }] }
    })
    mcp.experimental.tasks.registerToolTask('approve_job', { description: '长任务中途需要确认', inputSchema: { key: z.string() }, execution: { taskSupport: 'required' } }, {
      async createTask({ key }, { taskStore: store, taskRequestedTtl }) {
        const task = await store.createTask({ ttl: taskRequestedTtl, pollInterval: 50 })
        void (async () => {
          await store.updateTaskStatus(task.taskId, 'input_required')
          let result: ElicitResult | undefined
          for await (const m of mcp.server.experimental.tasks.elicitInputStream({ mode: 'form', message: `继续执行 ${key} 吗？`, requestedSchema: { type: 'object', properties: { ok: { type: 'boolean', title: '继续' } }, required: ['ok'] } }, { relatedTask: { taskId: task.taskId } })) {
            if (m.type === 'result') result = m.result as ElicitResult
          }
          await store.storeTaskResult(task.taskId, 'completed', { content: [{ type: 'text', text: `任务结束：${result?.action}:${JSON.stringify(result?.content ?? {})}` }] })
        })()
        return { task }
      },
      getTask: async (_a, { taskId, taskStore: store }) => (await store.getTask(taskId))!,
      getTaskResult: async (_a, { taskId, taskStore: store }) => await store.getTaskResult(taskId) as never,
    })
    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 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, body ? JSON.parse(body) : undefined)
  })
  return new Promise(r => upstream.listen(0, '127.0.0.1', () => r(`http://127.0.0.1:${(upstream.address() as AddressInfo).port}/mcp`)))
}

const forms: NewForm[] = []
let reply: (f: NewForm) => InteractionOutcome = () => ({ resolution: 'cancelled' })
function runCtx(): GatewayRunContext {
  const ws = path.join(config.dataDir, 'workspaces', `ses_ask_${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 => {
      if (req.kind !== 'form') return { resolution: 'allowed', scope: 'once' }
      forms.push(req)
      return reply(req)
    },
  }
}
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 text = (r: unknown) => (r as { content: { text: string }[] }).content[0]!.text

Object.assign(taskTiming, { minPollMs: 50, defaultPollMs: 100, foregroundPollMs: 50, tickMs: 50 })

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' })
  const url = await startUpstream()
  serverId = app.mcpServers.create({ name: 'ask', title: '下单服务', description: '', transport: 'http', enabled: true, url, headers: [], command: '', args: [], env: [], allowNetwork: false, readPaths: [], writePaths: [] }).id
  await app.mcpServers.refresh(serverId)
  app.mcpServers.setToolPolicy(serverId, 'approve_job', 'auto')
})
afterAll(async () => {
  app.mcpTasks.stop()
  app.backgroundCalls.stop()
  await app.mcpUpstream.closeAll()
  await server.close()
  upstream.closeAllConnections()
  upstream.close()
})

describe('表单字段', () => {
  it('requestedSchema 转换为表单字段；提交时按字段校验', () => {
    const fields = schemaToFields(SCHEMA)
    expect(fields).toEqual([
      { name: 'name', title: '收件人', required: true, type: 'string', minLength: 1 },
      { name: 'email', title: '邮箱', required: false, type: 'string', format: 'email' },
      { name: 'count', title: '数量', required: false, type: 'integer', minimum: 1, maximum: 10, default: 1 },
      { name: 'urgent', title: '加急', required: false, type: 'boolean' },
      { name: 'color', title: '颜色', required: true, type: 'choice', options: [{ value: 'red', label: '红' }, { value: 'blue', label: '蓝' }], multiple: false },
      { name: 'tags', title: '标签', required: false, type: 'choice', options: ['a', 'b', 'c'].map(v => ({ value: v, label: v })), multiple: true, maxItems: 2 },
    ])
    expect(schemaToFields({ type: 'object', properties: { x: { type: 'object' } } })).toMatch(/不支持/)
    if (typeof fields === 'string') throw new Error(fields)
    expect(validateFormContent(fields, { name: '张三', color: 'red', count: '3', extra: 'x' })).toEqual({ name: '张三', color: 'red', count: 3 })
    expect(() => validateFormContent(fields, { color: 'red' })).toThrow(/请填写：收件人/)
    expect(() => validateFormContent(fields, { name: 'a', color: 'green' })).toThrow(/颜色 的选项不正确/)
    expect(() => validateFormContent(fields, { name: 'a', color: 'red', count: 11 })).toThrow(/不能大于 10/)
    expect(() => validateFormContent(fields, { name: 'a', color: 'red', email: 'nope' })).toThrow(/邮箱地址/)
    expect(() => validateFormContent(fields, { name: 'a', color: 'red', tags: ['a', 'b', 'c'] })).toThrow(/最多选 2 项/)
  })
})

describe('调用过程中询问', () => {
  it('表单询问显示在这一轮，提交的内容回给服务；拒绝、关闭原样回给服务', async () => {
    const c = await gateway(runCtx())
    reply = () => ({ resolution: 'submitted', content: { name: '张三', color: 'blue' } })
    expect(text(await c.callTool({ name: 'ask__order', arguments: {} }))).toBe('accept:{"name":"张三","color":"blue"}')
    expect(forms[0]).toMatchObject({ kind: 'form', mode: 'form', source: '下单服务', message: '请填写收件信息' })
    expect(forms[0]!.fields).toHaveLength(6)

    reply = () => ({ resolution: 'denied' })
    expect(text(await c.callTool({ name: 'ask__order', arguments: {} }))).toBe('decline:{}')
    reply = () => ({ resolution: 'timeout' })
    expect(text(await c.callTool({ name: 'ask__order', arguments: {} }))).toBe('cancel:{}')
    await c.close()
  })

  it('网址询问：显示网址，用户完成后回复 accept', async () => {
    const c = await gateway(runCtx())
    reply = () => ({ resolution: 'submitted', content: {} })
    expect(text(await c.callTool({ name: 'ask__login', arguments: {} }))).toBe('accept')
    expect(forms.at(-1)).toMatchObject({ mode: 'url', url: 'https://auth.example.com/authorize?x=1', message: '请先授权访问日历' })
    await c.close()
  })

  it('无法确定属于哪个会话的询问一律拒绝', async () => {
    const row = app.mcpServers.rows([serverId])[0]!
    const r = await app.elicitation.handle({ row, params: { mode: 'form', message: 'x', requestedSchema: { type: 'object', properties: {} } }, taskId: null, scope: null, owners: [{ sessionId: 'a' }, { sessionId: 'b' }] })
    expect(r).toEqual({ action: 'decline' })
  })
})

describe('长任务中询问', () => {
  it('任务进入 input_required 时发出询问卡片（不属于某一轮），回答后任务继续并完成', async () => {
    const session = await app.sessions.create('admin', {})
    const ctx = { ...runCtx(), sessionId: session.id }
    const c = await gateway(ctx)
    const call = c.callTool({ name: 'ask__approve_job', arguments: { key: 'J1' } }, undefined, { timeout: 30_000 })
    const pending = await expect.poll(() => app.interactions.pendingFor(ctx.sessionId)[0], { timeout: 8000 }).toBeTruthy().then(() => app.interactions.pendingFor(ctx.sessionId)[0] as InteractionFormRequest)
    expect(pending).toMatchObject({ kind: 'form', message: '继续执行 J1 吗？', fields: [{ name: 'ok', type: 'boolean', required: true }] })
    expect(() => app.interactions.respond(pending.interactionId, { action: 'accept', content: {} })).toThrow(/请填写/)
    app.interactions.respond(pending.interactionId, { action: 'accept', content: { ok: true } })
    expect(text(await call)).toBe('任务结束：accept:{"ok":true}')
    await c.close()
  }, 20_000)
})