Node.js 在 AI Agent 中的应用：从工具开发到 MCP 协议

Python:
  await asyncio.gather(tool1(), tool2(), tool3())  # 需要显式管理

Node.js:
  await Promise.all([tool1(), tool2(), tool3()])   # 天然并行

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new Server({
  name: "weather-server",
  version: "1.0.0",
}, {
  capabilities: { tools: {} }
});

// 注册工具
server.setRequestHandler("tools/list", async () => ({
  tools: [{
    name: "get_weather",
    description: "获取指定城市的实时天气",
    inputSchema: {
      type: "object",
      properties: {
        city: { type: "string", description: "城市名称" }
      },
      required: ["city"]
    }
  }]
}));

// 处理工具调用
server.setRequestHandler("tools/call", async (request) => {
  const { name, arguments: args } = request.params;
  
  if (name === "get_weather") {
    const response = await fetch(
      `https://api.weather.com/v1/current?city=${args.city}`
    );
    const data = await response.json();
    return {
      content: [{
        type: "text",
        text: `${args.city}当前温度${data.temp}°C，${data.condition}`
      }]
    };
  }
});

// 启动
const transport = new StdioServerTransport();
await server.connect(transport);

// Node.js 流式返回
server.setRequestHandler("tools/call", async (request) => {
  const stream = db.query("SELECT * FROM logs").stream();
  
  return {
    content: [{
      type: "stream",
      stream: new ReadableStream({
        start(controller) {
          stream.on("data", (row) => {
            controller.enqueue(JSON.stringify(row) + "\n");
          });
          stream.on("end", () => controller.close());
        }
      })
    }]
  };
});

import Fastify from "fastify";

const app = Fastify();

app.get("/chat/stream", async (request, reply) => {
  reply.raw.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache",
    "Connection": "keep-alive",
  });

  const stream = await openai.chat.completions.create({
    model: "gpt-4o",
    messages: [{ role: "user", content: request.query.q }],
    stream: true,
  });

  for await (const chunk of stream) {
    const content = chunk.choices[0]?.delta?.content || "";
    if (content) {
      reply.raw.write(`data: ${JSON.stringify({ chunk: content })}\n\n`);
    }
  }

  reply.raw.write("data: [DONE]\n\n");
  reply.raw.end();
});

class ToolExecutor {
  constructor(maxConcurrent = 5) {
    this.semaphore = new Semaphore(maxConcurrent);
  }

  async executeAll(toolCalls) {
    const results = await Promise.allSettled(
      toolCalls.map(async (call) => {
        await this.semaphore.acquire();
        try {
          return await this.executeOne(call);
        } finally {
          this.semaphore.release();
        }
      })
    );

    return results.map((r, i) => ({
      tool: toolCalls[i].function.name,
      result: r.status === "fulfilled" ? r.value : { error: r.reason.message }
    }));
  }

  async executeOne(call) {
    const { name, arguments: args } = call.function;
    const parsed = JSON.parse(args);
    const handler = this.handlers[name];
    if (!handler) throw new Error(`Unknown tool: ${name}`);
    return handler(parsed);
  }
}

async function withRetry(fn, { maxRetries = 3, timeout = 10000 } = {}) {
  for (let i = 0; i <= maxRetries; i++) {
    try {
      return await Promise.race([
        fn(),
        new Promise((_, reject) =>
          setTimeout(() => reject(new Error("Timeout")), timeout)
        )
      ]);
    } catch (err) {
      if (i === maxRetries) throw err;
      await new Promise(r => setTimeout(r, Math.pow(2, i) * 1000));
    }
  }
}

class MemoryManager {
  constructor(maxTokens = 4000) {
    this.messages = [];
    this.maxTokens = maxTokens;
  }

  add(role, content) {
    this.messages.push({ role, content, timestamp: Date.now() });
    this._trim();
  }

  _trim() {
    let totalTokens = this._countTokens();
    
    // 超过上限时，摘最早的对话
    while (totalTokens > this.maxTokens && this.messages.length > 4) {
      const removed = this.messages.splice(0, 2); // 删一对对话
      this.summary = this._summarize(removed);
      totalTokens = this._countTokens();
    }
  }

  _countTokens() {
    return this.messages.reduce(
      (sum, m) => sum + Math.ceil(m.content.length / 4), 0
    );
  }

  getContext() {
    const context = [];
    if (this.summary) {
      context.push({ role: "system", content: `历史摘要: ${this.summary}` });
    }
    return [...context, ...this.messages];
  }
}

import { z } from "zod";
import { zodToJsonSchema } from "zod-to-json-schema";

// 定义工具的输入 schema
const WeatherInput = z.object({
  city: z.string().describe("城市名称"),
  days: z.number().min(1).max(7).default(3).describe("预报天数")
});

// 自动生成 OpenAI Function Calling 格式
const toolSchema = {
  type: "function",
  function: {
    name: "get_weather",
    description: "获取城市天气预报",
    parameters: zodToJsonSchema(WeatherInput)
  }
};

class Agent {
  constructor({ llm, tools, memory }) {
    this.llm = llm;
    this.tools = tools;
    this.memory = memory;
    this.executor = new ToolExecutor(5);
  }

  async run(userInput) {
    this.memory.add("user", userInput);

    let loopCount = 0;
    const MAX_LOOPS = 10;

    while (loopCount < MAX_LOOPS) {
      loopCount++;

      const response = await this.llm.chat({
        messages: this.memory.getContext(),
        tools: this.tools.getSchemas(),
      });

      const choice = response.choices[0];
      
      // LLM 想直接回复
      if (choice.finish_reason === "stop") {
        const content = choice.message.content;
        this.memory.add("assistant", content);
        return content;
      }

      // LLM 想调用工具
      const toolCalls = choice.message.tool_calls;
      if (toolCalls) {
        const results = await this.executor.executeAll(toolCalls);
        
        // 把工具结果加入上下文
        this.memory.add("assistant", null, toolCalls);
        for (const r of results) {
          this.memory.add("tool", JSON.stringify(r.result), null, r.tool);
        }
      }
    }

    throw new Error("Agent loop limit exceeded");
  }
}