从JS云函数到MCP：打造跨平台AI Agent工具的工程实践

GET /room/info?roomid=123

exportclassQueryLiveRoomToolextendsStructuredTool {  name = 'QueryLiveRoomTool'  description = '根据 uid 查询主播信息'  schema = z.object({    roomid: z.string().describe('直播间id')  })  async _call({ roomid }) {    const res = await axios.post({...})    return formatForAI(res)  }}

/** * 核心思路抽象版 */import { NodeVM } from 'vm2'import { StructuredTool } from '@langchain/core/tools'//...classDynamicToolextendsStructuredTool {  constructor({ name, description, schema, code }) {    super({ name, description, schema })    this.code = code // 用户在在线编辑器里写的 JS 代码  }  async _call(args, runManager, flowContext) {    // 1. 构建沙箱（所有注入能力都放在这里）    const sandbox = {      // 用户传入的参数转成 $xxx      ...Object.fromEntries(Object.entries(args).map(([k, v]) => [`$${k}`, v])),      // 内部上下文（cookie/env/session）      $flow: flowContext,      $cookie: flowContext.cookie,      // 内网能力封装（gRPC/HTTP）      $yuumi: yuumi,      // 结果转换工具（用于把 JSON 转成更 AI 友好的 Markdown 表格）      $json2MarkdownTable: json2MarkdownTable,      // 内部 AI 模型      $biliLLM: biliLLMClient    }    // 2. 构建 NodeVM 沙箱    const vm = new NodeVM({      sandbox,      console: "inherit",      require: {        builtin: allowedBuiltinDeps,        external: allowedExternalDeps      }    })    // 3. 执行开发者写的云函数代码    return await vm.run(      `module.exports = async () => { ${this.code} }()`,      __dirname    )  }}

const res = await $yuumi.grpc({  appId: 'live.service',  path: '/room/info',  params: { roomid: $roomid },  cookie: $cookie,})return $json2MarkdownTable(res.data.list)

roomid: string

export function createMCPServer({ app, AppDataSource }: App){  // 单个云函数 → MCP-StreamableHTTP  app.post('/api/mcp/function-tool/:toolId', singleToolCreateStreamableHTTPServer)  // 多个云函数组合 → MCP-StreamableHTTP  app.post('/api/mcp/:mcpId', multipleToolCreateStreamableHTTPServer)  // 会话后续请求复用  app.get('/api/mcp/function-tool/:id', handleSessionRequest)  app.delete('/api/mcp/function-tool/:id', handleSessionRequest)  app.get('/api/mcp/:mcpId', handleSessionRequest)  app.delete('/api/mcp/:mcpId', handleSessionRequest)  const transports = {    streamable: {} as Record<string, StreamableHTTPServerTransport>  }  // ... 省略 SSE 相关 ...}

async function createStreamableHTTP(config: CreateMcpServerConfig){  const { req, res, username } = config  const sessionId = req.headers['mcp-session-id'] as string | undefined  let transport: StreamableHTTPServerTransport  if (sessionId && transports.streamable[sessionId]) {    // ① 有 sessionId，复用之前的 transport    transport = transports.streamable[sessionId]  } elseif (!sessionId && isInitializeRequest(req.body)) {    // ② 首次初始化，请求体符合 MCP 初始化格式    transport = new StreamableHTTPServerTransport({      sessionIdGenerator: () => randomUUID(),      onsessioninitialized: (sessionId) => {        transports.streamable[sessionId] = transport      }    })    // 连接关闭时清理掉这个 session    transport.onclose = () => {      if (transport.sessionId) {        delete transports.streamable[transport.sessionId]        opsLogger.info(`[Streamable] 删除sessionId: username=${username} sessionId=${transport.sessionId}`)      }    }    const server = buildMcpServer({      ...config,      transport: 'streamable-http'    })    await server.connect(transport)  } else {    // 既不是初始化，又没有合法 sessionId：直接返回错误    res.status(400).json({      jsonrpc: '2.0',      error: {        code: -32000,        message: 'Bad Request: No valid session ID provided'      },      id: null    })    return  }  await transport.handleRequest(req, res, req.body)}

async function handleSessionRequest(req: Request, res: Response){  const sessionId = req.headers['mcp-session-id'] as string | undefined  if (!sessionId || !transports.streamable[sessionId]) {    res.status(400).send('Invalid or missing session ID')    return  }  const transport = transports.streamable[sessionId]  await transport.handleRequest(req, res)}

function buildMcpServer(config: BuildMcpServerConfig): McpServer {  const { name, tools, cookie, env, id, type, username, transport } = config  const server = new McpServer({    name,    version: '1.0.0'  })  for (const tool of tools) {    // 1）把数据库里的元信息，转成 DynamicStructuredTool 入参    const obj = {      name: tool.name,      description: tool.description,      schema: z.object(convertSchemaToZod(tool.schema as string | object)),      code: tool.func as string    }    // DynamicStructuredTool 内部就是一个“动态 StructuredTool” + NodeVM 沙箱    const dynamicStructuredTool = new DynamicStructuredTool(obj)    // Flow 信息：cookie + env（来自 MCP URL 上的 sign / 其它 query）    dynamicStructuredTool.setFlowObject({      cookie,      env: typeof env === 'object' && Object.keys(env).length ? env : {}    })    // 2）把我们在 UI 里配置的 Tool Arguments schema，转成 MCP 需要的 zod 参数定义    const schemaParser = (tool.schema ? JSON.parse(tool.schema) : []) as Schema[]    const paramsSchema = schemaParser.reduce((res, cur) => {      if (cur.required) {        res[cur.property] = z[cur.type]({ required_error: `${cur.property} required` })          .describe(cur.description) as z.ZodTypeAny      } else {        res[cur.property] = z[cur.type]()          .describe(cur.description)          .optional() as z.ZodTypeAny      }      return res    }, {} as any)    // 3）用官方 server.tool() 方法注册 MCP Tool    server.tool(tool.name, tool.description, paramsSchema, async (args, _extra) => {      // 注意：这里的 call() 对应的就是 LangChain StructuredTool._call()      const runnerResult = await dynamicStructuredTool.call({        ...args      })      // 线上环境做一层使用上报（方便统计谁在用哪个 MCP）      if (process.env.DEPLOY_ENV === 'prod') {        reportMcpUse({          id,          type,          name,          username,          transport        })      }      // MCP 协议统一的返回格式：content[] 数组      return {        content: [{ type: 'text', text: runnerResult }]      }    })  }  return server}