AgentServer/examples/mcp-reference-server/server.mjs

// 参考 MCP 服务：按《MCP 接入指南》（docs/guides/mcp-guide.md）编写，演示 AgentServer 支持的各项 MCP 能力。
// 业务是一个虚构的"订单系统"，数据在内存中，重启后恢复初始状态。
//
//   PORT=8790 REF_TOKEN=secret node server.mjs
//
// 在 AgentServer 后台"MCP 服务"中添加：远程 HTTP，地址 http://<主机>:8790/mcp，请求头 Authorization: Bearer secret
import crypto from 'node:crypto'
import http from 'node:http'
import { z } from 'zod'
import { McpServer, ResourceTemplate } from '@modelcontextprotocol/sdk/server/mcp.js'
import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js'
import { InMemoryTaskMessageQueue, InMemoryTaskStore } from '@modelcontextprotocol/sdk/experimental/tasks/stores/in-memory.js'

const PORT = Number(process.env.PORT ?? 8790)
const HOST = process.env.HOST ?? '127.0.0.1'
const TOKEN = process.env.REF_TOKEN ?? 'ref-secret'

// ---- 业务数据 ----
const orders = new Map([
  ['A1001', { id: 'A1001', customer: '张三', amount: 1299, status: 'paid', items: ['储能电池 5kWh'] }],
  ['A1002', { id: 'A1002', customer: '李四', amount: 399, status: 'shipped', items: ['智能网关'] }],
])
const notes = []

/**
 * 长任务的存储：所有连接共用。
 * 指南 §5.3：任务要按认证身份归属，不要绑定在 MCP 会话上，否则平台重连（或重启）后查不到。
 * 这里只有一个令牌，所以直接共用；有多个调用方时，按令牌区分。
 */
class SharedTaskStore extends InMemoryTaskStore {
  createTask(params, requestId, request) { return super.createTask(params, requestId, request, undefined) }
  getTask(taskId) { return super.getTask(taskId, undefined) }
  storeTaskResult(taskId, status, result) { return super.storeTaskResult(taskId, status, result, undefined) }
  getTaskResult(taskId) { return super.getTaskResult(taskId, undefined) }
  updateTaskStatus(taskId, status, message) { return super.updateTaskStatus(taskId, status, message, undefined) }
  listTasks(cursor) { return super.listTasks(cursor, undefined) }
}
const taskStore = new SharedTaskStore()
const taskQueue = new InMemoryTaskMessageQueue()

let betaEnabled = false

function buildServer() {
  const server = new McpServer(
    { name: 'order-reference', version: '1.0.0' },
    {
      capabilities: { tasks: { requests: { tools: { call: {} } }, cancel: {}, list: {} } },
      taskStore,
      taskMessageQueue: taskQueue,
      // 写给模型看的服务说明：什么时候用哪个工具
      instructions: '订单系统。查订单用 get_order；导出报表很慢，用 export_report（长任务）；删除类操作会要求确认。',
    },
  )

  // ---- 1. 只读工具 + 结构化输出（指南 §3、§4） ----
  server.registerTool('get_order', {
    title: '查询订单',
    description: '按订单号查询订单的客户、金额、状态和商品。订单号形如 A1001。',
    inputSchema: { orderId: z.string().regex(/^A\d{4}$/).describe('订单号，例如 A1001') },
    outputSchema: { id: z.string(), customer: z.string(), amount: z.number(), status: z.enum(['paid', 'shipped', 'cancelled']), items: z.array(z.string()) },
    annotations: { readOnlyHint: true, openWorldHint: false },
  }, async ({ orderId }) => {
    const o = orders.get(orderId)
    // 错误写给模型看：说清楚原因和下一步怎么做（指南 §3.4）
    if (!o) return { isError: true, content: [{ type: 'text', text: `订单 ${orderId} 不存在。可以先用 list_orders 查看现有订单号。` }] }
    // 结构化结果同时序列化成一段文本，兼容只看文本的客户端
    return { content: [{ type: 'text', text: JSON.stringify(o) }], structuredContent: o }
  })

  server.registerTool('list_orders', {
    title: '订单列表',
    description: '分页列出订单（每页最多 20 条），可以按状态筛选。',
    inputSchema: { status: z.enum(['paid', 'shipped', 'cancelled']).optional(), cursor: z.string().optional().describe('上一页返回的 nextCursor') },
    annotations: { readOnlyHint: true, openWorldHint: false },
  }, async ({ status, cursor }) => {
    const all = [...orders.values()].filter(o => !status || o.status === status)
    const start = cursor ? Number(cursor) : 0
    const page = all.slice(start, start + 20)
    const next = start + 20 < all.length ? String(start + 20) : null
    const structuredContent = { orders: page.map(o => ({ id: o.id, customer: o.customer, status: o.status })), nextCursor: next }
    return { content: [{ type: 'text', text: JSON.stringify(structuredContent) }], structuredContent }
  })

  // ---- 2. 写操作与提示（指南 §3.3） ----
  server.registerTool('add_note', {
    title: '添加备注',
    description: '给订单添加一条备注。每次调用都会新增一条（不幂等）。',
    inputSchema: { orderId: z.string(), text: z.string().max(500) },
    annotations: { readOnlyHint: false, destructiveHint: false, idempotentHint: false, openWorldHint: false },
  }, async ({ orderId, text }) => {
    notes.push({ orderId, text, at: new Date().toISOString() })
    return { content: [{ type: 'text', text: `已为 ${orderId} 添加备注（共 ${notes.filter(n => n.orderId === orderId).length} 条）` }] }
  })

  server.registerTool('cancel_order', {
    title: '取消订单',
    description: '取消订单。已取消的订单再次取消不会有变化（幂等）。这是破坏性操作，平台会要求用户确认。',
    inputSchema: { orderId: z.string(), reason: z.string().describe('取消原因') },
    annotations: { readOnlyHint: false, destructiveHint: true, idempotentHint: true, openWorldHint: false },
  }, async ({ orderId, reason }) => {
    const o = orders.get(orderId)
    if (!o) return { isError: true, content: [{ type: 'text', text: `订单 ${orderId} 不存在` }] }
    o.status = 'cancelled'
    return { content: [{ type: 'text', text: `订单 ${orderId} 已取消，原因：${reason}` }] }
  })

  // ---- 3. 调用过程中询问用户（指南 §6） ----
  server.registerTool('ship_order', {
    title: '安排发货',
    description: '为订单安排发货。会请用户填写收件信息。',
    inputSchema: { orderId: z.string() },
    annotations: { readOnlyHint: false, destructiveHint: false, idempotentHint: false, openWorldHint: true },
  }, async ({ orderId }) => {
    const r = await server.server.elicitInput({
      mode: 'form',
      message: `请填写订单 ${orderId} 的收件信息`,
      requestedSchema: {
        type: 'object',
        properties: {
          name: { type: 'string', title: '收件人', minLength: 1 },
          phone: { type: 'string', title: '手机号', minLength: 11, maxLength: 11 },
          address: { type: 'string', title: '地址' },
          speed: { type: 'string', title: '配送方式', oneOf: [{ const: 'standard', title: '普通' }, { const: 'express', title: '加急' }], default: 'standard' },
          insured: { type: 'boolean', title: '保价', default: false },
        },
        required: ['name', 'phone', 'address'],
      },
    })
    // 三种结果都要处理：提交、拒绝、关闭
    if (r.action === 'accept') return { content: [{ type: 'text', text: `已安排发货：${JSON.stringify(r.content)}` }] }
    if (r.action === 'decline') return { content: [{ type: 'text', text: '用户拒绝提供收件信息，未发货' }] }
    return { content: [{ type: 'text', text: '用户暂未处理，未发货。可以稍后再试' }] }
  })

  server.registerTool('connect_crm', {
    title: '连接 CRM',
    description: '连接外部 CRM。需要用户在浏览器中授权（网址模式），不在对话中索要密码。',
    inputSchema: {},
    annotations: { readOnlyHint: false, destructiveHint: false, openWorldHint: true },
  }, async () => {
    const r = await server.server.elicitInput({ mode: 'url', message: '请在打开的页面中登录 CRM 并授权订单系统访问', url: 'https://crm.example.com/oauth/authorize?client=orders', elicitationId: crypto.randomUUID() })
    return { content: [{ type: 'text', text: r.action === 'accept' ? '用户已完成授权' : '用户没有授权' }] }
  })

  // ---- 4. 长任务（指南 §5） ----
  server.experimental.tasks.registerToolTask('export_report', {
    title: '导出销售报表',
    description: '导出指定月份的销售报表。很慢（演示中按 seconds 参数计时），以长任务方式执行，完成后平台会把结果发给你。',
    inputSchema: { month: z.string().regex(/^\d{4}-\d{2}$/).describe('月份，例如 2026-10'), seconds: z.number().int().min(1).max(7200).default(30).describe('演示用：执行多少秒') },
    execution: { taskSupport: 'optional' },
    annotations: { readOnlyHint: true },
  }, {
    async createTask({ month, seconds }, { taskStore: store, taskRequestedTtl }) {
      const task = await store.createTask({ ttl: taskRequestedTtl ?? 24 * 3600_000, pollInterval: 5000 })
      const started = Date.now()
      const timer = setInterval(() => {
        const pct = Math.min(99, Math.round(((Date.now() - started) / (seconds * 1000)) * 100))
        void taskStore.updateTaskStatus(task.taskId, 'working', `已完成 ${pct}%`).catch(() => clearInterval(timer))
      }, 5000)
      setTimeout(() => {
        clearInterval(timer)
        void taskStore.storeTaskResult(task.taskId, 'completed', {
          content: [
            { type: 'text', text: `${month} 销售报表已生成：订单 ${orders.size} 笔` },
            // 文件以嵌入资源返回：平台保存到会话工作区，不放进模型上下文（指南 §4.3）
            { type: 'resource', resource: { uri: `ref://reports/sales-${month}.csv`, mimeType: 'text/csv', text: ['id,customer,amount,status', ...[...orders.values()].map(o => `${o.id},${o.customer},${o.amount},${o.status}`)].join('\n') } },
          ],
        }).catch(() => {})
      }, seconds * 1000)
      // 给模型的即时说明：提交后马上返回
      return { task, _meta: { 'io.modelcontextprotocol/model-immediate-response': `${month} 报表已开始导出，预计 ${seconds} 秒` } }
    },
    getTask: async (_args, { taskId }) => taskStore.getTask(taskId),
    getTaskResult: async (_args, { taskId }) => taskStore.getTaskResult(taskId),
  })

  server.experimental.tasks.registerToolTask('purge_cancelled', {
    title: '清理已取消订单',
    description: '永久删除所有已取消的订单。执行中会请用户确认（长任务中的询问）。',
    inputSchema: {},
    execution: { taskSupport: 'required' },
    annotations: { readOnlyHint: false, destructiveHint: true },
  }, {
    async createTask(_args, { taskStore: store, taskRequestedTtl }) {
      const task = await store.createTask({ ttl: taskRequestedTtl ?? 3600_000, pollInterval: 2000 })
      void (async () => {
        const victims = [...orders.values()].filter(o => o.status === 'cancelled').map(o => o.id)
        await taskStore.updateTaskStatus(task.taskId, 'input_required', '等待用户确认')
        let answer
        for await (const m of server.server.experimental.tasks.elicitInputStream(
          { mode: 'form', message: `将永久删除 ${victims.length} 个已取消订单：${victims.join('、') || '（无）'}。确认吗？`, requestedSchema: { type: 'object', properties: { confirm: { type: 'boolean', title: '确认删除' } }, required: ['confirm'] } },
          { relatedTask: { taskId: task.taskId } },
        )) if (m.type === 'result') answer = m.result
        const ok = answer?.action === 'accept' && answer.content?.confirm === true
        if (ok) for (const id of victims) orders.delete(id)
        await taskStore.storeTaskResult(task.taskId, 'completed', { content: [{ type: 'text', text: ok ? `已删除 ${victims.length} 个订单` : '用户没有确认，未删除' }] })
      })().catch(err => taskStore.storeTaskResult(task.taskId, 'failed', { isError: true, content: [{ type: 'text', text: String(err) }] }).catch(() => {}))
      return { task }
    },
    getTask: async (_args, { taskId }) => taskStore.getTask(taskId),
    getTaskResult: async (_args, { taskId }) => taskStore.getTaskResult(taskId),
  })

  // ---- 5. 工具列表变化（指南 §7） ----
  server.registerTool('enable_beta', {
    title: '开启测试功能',
    description: '开启测试功能后会多出一个 beta_forecast 工具（演示工具列表变化通知）。',
    inputSchema: {},
    annotations: { readOnlyHint: false, idempotentHint: true, openWorldHint: false },
  }, async () => {
    betaEnabled = true
    for (const s of sessions.values()) s.addBeta()
    return { content: [{ type: 'text', text: '已开启，工具列表已更新' }] }
  })
  const addBeta = () => {
    if (server._registeredTools?.beta_forecast) return
    server.registerTool('beta_forecast', { title: '销量预测（测试）', description: '预测下个月的销量（测试功能）', inputSchema: {}, annotations: { readOnlyHint: true } },
      async () => ({ content: [{ type: 'text', text: '预计下月订单 120 笔' }] }))
  }
  if (betaEnabled) addBeta()

  // ---- 6. 资源与资源模板（指南 §8） ----
  server.registerResource('api-guide', 'ref://docs/order-api.md', {
    title: '订单系统接口说明', description: '订单状态的含义、退款规则', mimeType: 'text/markdown',
  }, async uri => ({
    contents: [{ uri: uri.href, mimeType: 'text/markdown', text: '# 订单状态\n- paid：已支付，未发货\n- shipped：已发货\n- cancelled：已取消\n\n# 退款规则\n发货前全额退款；发货后 7 天内可退。' }],
  }))
  server.registerResource('order', new ResourceTemplate('ref://orders/{orderId}', {
    list: async () => ({ resources: [...orders.values()].map(o => ({ uri: `ref://orders/${o.id}`, name: o.id, title: `订单 ${o.id}`, mimeType: 'application/json' })) }),
  }), { title: '订单详情', description: '单个订单的完整数据（JSON）', mimeType: 'application/json' }, async (uri, { orderId }) => {
    const o = orders.get(String(orderId))
    if (!o) throw new Error(`订单 ${orderId} 不存在`)
    return { contents: [{ uri: uri.href, mimeType: 'application/json', text: JSON.stringify({ ...o, notes: notes.filter(n => n.orderId === o.id) }, null, 2) }] }
  })

  // ---- 7. 提示模板（指南 §8） ----
  server.registerPrompt('order_review', {
    title: '订单复盘',
    description: '针对一个订单写复盘：时间线、问题、改进建议',
    argsSchema: { orderId: z.string().describe('订单号'), focus: z.string().optional().describe('关注点（可选）') },
  }, async ({ orderId, focus }) => ({
    messages: [
      { role: 'user', content: { type: 'text', text: `请对订单 ${orderId} 做一次复盘，列出时间线、发现的问题和改进建议${focus ? `，重点关注：${focus}` : ''}。` } },
      { role: 'user', content: { type: 'resource', resource: { uri: `ref://orders/${orderId}`, mimeType: 'application/json', text: JSON.stringify(orders.get(orderId) ?? {}, null, 2) } } },
    ],
  }))

  return { server, addBeta }
}

// ---- HTTP：Streamable HTTP，有状态（询问用户需要在同一个 MCP 会话中收发） ----
const sessions = new Map()
const httpServer = http.createServer(async (req, res) => {
  const url = new URL(req.url ?? '/', `http://${req.headers.host}`)
  if (url.pathname !== '/mcp') return res.writeHead(404).end()
  // 认证（指南 §9）：每个请求都检查令牌
  if (req.headers.authorization !== `Bearer ${TOKEN}`) return res.writeHead(401, { 'content-type': 'application/json' }).end(JSON.stringify({ error: 'unauthorized' }))
  let body = ''
  for await (const chunk of req) body += chunk
  const sid = req.headers['mcp-session-id']
  let s = sid ? sessions.get(sid) : undefined
  if (!s) {
    if (sid) return res.writeHead(404).end() // 会话已失效：客户端会重新初始化
    const { server, addBeta } = buildServer()
    const transport = new StreamableHTTPServerTransport({
      sessionIdGenerator: () => crypto.randomUUID(),
      onsessioninitialized: id => sessions.set(id, s),
    })
    transport.onclose = () => { if (transport.sessionId) sessions.delete(transport.sessionId) }
    s = { server, transport, addBeta }
    await server.connect(transport)
  }
  await s.transport.handleRequest(req, res, body ? JSON.parse(body) : undefined)
})

httpServer.listen(PORT, HOST, () => console.log(`MCP 参考服务：http://${HOST}:${PORT}/mcp（Authorization: Bearer ${TOKEN === 'ref-secret' ? 'ref-secret' : '***'}）`))