“三小时搞定AI工具开发“：基于MCP的Node.js极简实践

基于Node.js的MCP协议完整实现指南：连接大模型与现实世界的桥梁

一、MCP协议：大模型的"瑞士军刀"接口

1.1 什么是MCP？

MCP（Model Context Protocol）是专为大型语言模型设计的开放协议，它如同AI世界的USB接口标准。想象一下，当ChatGPT需要查询实时股价或进行复杂计算时，MCP就是让它能"即插即用"调用外部工具的标准化方式。

核心优势：

突破文本限制：让LLM获得现实世界的数据访问能力
统一工具接口：不同模型（GPT/Claude等）可共享同一套工具
流式交互：支持实时进度反馈（如长时间任务处理）

1.2 Function Calling vs MCP 深度对比

协议基础

对比维度	OpenAI Function Calling	MCP (Model Context Protocol)
协议类型	厂商私有协议	开放行业标准
发起组织	OpenAI	Anthropic主导，多厂商参与
标准化程度	文档化但未开源	公开规范文档与参考实现

技术实现

对比维度	Function Calling	MCP
通信协议	HTTP/JSON	HTTP/JSON + SSE流式传输
工具发现机制	客户端硬编码声明	动态发现端点 (`GET /discovery`)
执行模式	同步阻塞调用	支持同步/异步流式响应
数据格式	自定义JSON结构	基于JSON-RPC 2.0规范
状态管理	无原生状态跟踪	内置任务ID和进度跟踪

功能特性

对比维度	Function Calling	MCP
多工具组合	需客户端手动编排	支持工具链式调用 (Workflow)
长任务支持	依赖轮询	原生SSE流式进度更新仅OpenAI模型
版本兼容	绑定模型版本	协议版本独立 (`MCP-1.0`)
工具热更新	需重新部署模型	服务端随时更新工具
权限控制	仅API Key基础认证	可扩展OAuth2.0/ACL

生态支持

对比维度	Function Calling	MCP
模型兼容性	OpenAI等模型	跨模型 (Claude/GPT/本地LLM等)
工具复用性	无法跨平台共享	工具可发布到公共市场
开发者生态	OpenAI专属	多厂商工具开发者社区
监控指标	基础用量统计	丰富的Prometheus指标

性能表现

对比维度	Function Calling	MCP
简单调用延迟	200-300ms	300-500ms (含协议开销)
流式响应速度	不支持	50-100ms/消息
并发能力	受限于模型配额	可水平扩展工具服务器
缓存机制	无内置缓存	支持响应缓存标注

适用场景建议

MCP 通信协议整理

MCP（Message Communication Protocol）是一种用于分布式系统和微服务架构的通信协议，旨在提供高效、可靠的消息传递和数据交换。MCP 协议支持多种通信方式，消息遵循 JSON-RPC 2.0 协议格式，适用于进程间通信、HTTP、SSE 和 WebSocket 等多种场景。

2. 通信方式

MCP 协议支持以下几种通信方式：

进程间通信（IPC）： 适用于同一主机上进程之间的消息传递，通常使用管道、共享内存或消息队列。
HTTP 协议： 基于请求-响应机制，适用于前后端交互和 RESTful API 调用。
SSE（Server-Sent Events）： 服务器单向推送协议，适用于实时更新和消息通知。
WebSocket 协议： 全双工通信协议，支持实时双向数据传输。

3. 消息协议

MCP 协议的消息格式遵循 JSON-RPC 2.0 标准，消息为 JSON 格式，具有以下特点：

基于 JSON 格式，轻量、易解析。
支持请求、响应和通知机制。
具备无状态和异步调用特性。

示例消息格式：

{"jsonrpc": "2.0","method": "subtract","params": [42, 23],"id": 1
}

响应格式：

{"jsonrpc": "2.0","result": 19,"id": 1
}

4. 相关协议详解

JSON-RPC 2.0 协议

轻量级远程过程调用协议，基于 JSON 数据格式。
适用于 HTTP、WebSocket 或 TCP 传输。
支持批处理和通知机制。

SSE（Server-Sent Events）协议

基于 HTTP 协议的单向推送机制。
服务器实时向客户端推送更新。
常用于实时数据流和通知场景。

WebSocket 协议

实时双向通信协议，基于 TCP 连接。
低延迟，适用于聊天应用和实时监控。

HTTP 协议

经典请求-响应机制，适合 REST API。
常用于 Web 应用前后端交互。

gRPC 协议

高性能 RPC 框架，基于 HTTP/2 和 Protocol Buffers。
适合微服务和实时数据传输。

MQTT 协议

轻量级发布/订阅协议，适用于物联网（IoT）。
支持低带宽和高效消息推送。

以上是 MCP 协议相关的通信方式和消息协议介绍。

1.3 MCP现状速览

厂商	支持情况	最新动态（2024）
OpenAI	通过Function Calling兼容	正在测试原生MCP支持
Anthropic	原生集成	Claude 3默认启用MCP工具市场
微软	Azure API网关支持	提供MCP工具安全认证服务

二、手把手遵循MCP实现一个需求

用户提问："对比特斯拉(TSLA)和英伟达(NVDA)过去三个月的股价波动，并给出投资组合建议"

实现思路

image

sequenceDiagramparticipant 用户participant ChatGPTparticipant MCP服务端participant 金融数据API用户->>ChatGPT: 输入查询请求ChatGPT->>MCP服务端: 调用stockHistory工具MCP服务端->>金融数据API: 获取TSLA历史数据MCP服务端->>金融数据API: 获取NVDA历史数据MCP服务端->>ChatGPT: 返回结构化数据ChatGPT->>MCP服务端: 调用analyzeTrend工具MCP服务端->>ChatGPT: 流式返回分析结果ChatGPT->>用户: 生成自然语言报告

实现拆解

server.js和 client.js 的代码在下面

工作流程

工具注册中心
像超市货架一样陈列所有可用工具，每个工具包含：

功能描述（供模型理解用途）
执行逻辑（具体的代码实现）
流式支持标记（是否支持实时进度反馈）

两大核心接口
- GET /discovery：
  模型通过此接口查看"货架清单"，返回示例：
```
{"tools": ["stockPrice", "analyzeSales"],"protocol": "MCP-1.0"
}
```
- POST /execute：
  模型实际调用工具时的入口，处理过程：
1. 检查请求工具是否存在
2. 验证参数格式
3. 执行工具逻辑
4. 返回JSON或流式结果
流式响应原理
对于耗时任务（如销售数据分析）：

创建PassThrough可读流
分阶段推送进度信息
保持连接直到任务完成

典型应用场景

当用户询问"特斯拉最近三个月销售趋势如何？"时：

模型发现需要调用analyzeSales工具
通过/execute接口触发分析

服务端持续返回：

{"progress": 30, "stage": "数据加载中..."}
{"progress": 80, "stage": "生成图表..."}
{"result": {...}}

2.1 server.js

import express from 'express';
import { EventEmitter } from 'events';const app = express();
app.use(express.json());// 工具注册（严格遵循MCP工具描述规范）
const tools = {stock_price: {description: "获取实时股票价格",parameters: {type: "object",properties: {symbol: { type: "string" }},required: ["symbol"]},handler: async ({ symbol }) => {return { price: (Math.random() * 100).toFixed(2), symbol };}}
};// MCP发现端点（必须实现）
app.post('/mcp/discover', (req, res) => {const response = {jsonrpc: "2.0",result: {protocol_version: "1.0",tools: Object.entries(tools).map(([name, def]) => ({name,description: def.description,parameters: def.parameters}))},id: req.body.id || null};res.json(response);
});// MCP调用端点（严格遵循JSON-RPC 2.0）
app.post('/mcp/invoke', async (req, res) => {const { jsonrpc, method, params, id } = req.body;// 校验JSON-RPC版本
if (jsonrpc !== "2.0") {return res.status(400).json({jsonrpc: "2.0",error: { code: -32600, message: "Invalid Request" },id});}// 解析工具名（method格式：tool.{name}）const [, toolName] = method.split('.');const tool = tools[toolName];if (!tool) {return res.json({jsonrpc: "2.0",error: { code: -32601, message: "Method not found" },id});}// 执行工具调用try {const result = await tool.handler(params);res.json({ jsonrpc: "2.0", result, id });} catch (err) {res.json({jsonrpc: "2.0",error: { code: -32603, message: err.message },id});}
});// SSE流式端点（MCP可选扩展）
app.get('/mcp/stream/:taskId', (req, res) => {res.setHeader('Content-Type', 'text/event-stream');const { taskId } = req.params;// 模拟流式进度
let progress = 0;const interval = setInterval(() => {progress += 10;res.write(`data: ${JSON.stringify({jsonrpc: "2.0",method: "progress_update",params: { taskId, progress }})}\n\n`);if (progress >= 100) {clearInterval(interval);res.write(`data: ${JSON.stringify({jsonrpc: "2.0",method: "task_complete",params: { taskId, result: "Done" }})}\n\n`);res.end();}}, 300);req.on('close', () => clearInterval(interval));
});app.listen(3000, () => {console.log('MCP Server running at http://localhost:3000');
});

2.2 client.js

import OpenAI from 'openai';
import { EventSource } from 'eventsource';const openai = new OpenAI({ apiKey: 'your-api-key' });// MCP工具描述转换器
function convertToOpenAITool(mcpTool) {
return {type: "function",function: {name: `tool.${mcpTool.name}`,description: mcpTool.description,parameters: mcpTool.parameters}};
}// 获取MCP工具列表
async functiongetMCPTools() {const response = await fetch('http://localhost:3000/mcp/discover', {method: 'POST',headers: { 'Content-Type': 'application/json' },body: JSON.stringify({ jsonrpc: "2.0", method: "discover", id: 1 })});const { result } = await response.json();
return result.tools.map(convertToOpenAITool);
}// 执行MCP调用
async function callMCPTool(method, params) {const response = await fetch('http://localhost:3000/mcp/invoke', {method: 'POST',headers: { 'Content-Type': 'application/json' },body: JSON.stringify({jsonrpc: "2.0",method,params,id: Math.floor(Math.random() * 1000)})});
return (await response.json()).result;
}// 主对话流程
async functionmain() {// 1. 动态获取工具列表const tools = await getMCPTools();// 2. 发起对话请求const chatCompletion = await openai.chat.completions.create({model: "gpt-4-turbo",messages: [{ role: "user", content: "AAPL当前股价是多少？" }],tools});// 3. 处理工具调用const toolCall = chatCompletion.choices[0].message.tool_calls?.[0];
if (toolCall) {const result = await callMCPTool(toolCall.function.name,JSON.parse(toolCall.function.arguments));// 4. 将结果返回给OpenAIconst finalCompletion = await openai.chat.completions.create({model: "gpt-4-turbo",messages: [{ role: "user", content: "AAPL当前股价是多少？" },chatCompletion.choices[0].message,{role: "tool",name: toolCall.function.name,content: JSON.stringify(result)}]});console.log(finalCompletion.choices[0].message.content);}
}main();

性能优化技巧

批处理工具调用：当多个工具需要并行调用时

const toolPromises = toolCalls.map(call => fetch(MCP_SERVER, {method: 'POST',body: JSON.stringify({tool: call.function.name,params: JSON.parse(call.function.arguments)})})
);const toolResults = await Promise.all(toolPromises);

缓存策略：对高频工具结果进行缓存

import { createClient } from 'redis';const redis = createClient();
await redis.connect();const cachedResult = await redis.get(`tool:${toolName}:${paramsHash}`);
if (cachedResult) return JSON.parse(cachedResult);// ...执行工具调用
await redis.setEx(`tool:${toolName}:${paramsHash}`, 60, JSON.stringify(result));

MCP的未来展望

技术演进路线

多模态扩展（2025）
支持图像生成/识别工具
视频处理工具规范
去中心化网络等

mermaid
graph TBA[模型] -->|MCP调用| B[工具市场]B --> C[区块链结算]B --> D[信誉评分]
安全增强（2024-2025）
硬件级隔离（Intel SGX）
联邦学习支持