夜雨聆风MCP 协议实战：让 Agent 插上 100+ 工具的翅膀
📌 本文是《从零吃透企业级 AI 平台：元景万悟源码学习手记》系列第一季·源码学习篇的第 6 篇🎯 读完本文你将：① 理解 MCP 协议的设计哲学与通信机制 ② 掌握万悟 MCP 广场的架构与工具接入流程 ③ 能独立开发一个 MCP Server 并接入万悟⏱️ 预计阅读时间：35 分钟 | 动手实践：80 分钟💻 前置要求：已完成第 5 篇，理解 Agent 与 Function Calling 机制
📌 系列修订说明
本系列前 5 篇发布后，我已逐篇对照万悟源码（GitHub go.mod / docker-compose / 源码目录）核实修正。与前 5 篇可能不一致的关键点：
① 万悟是 11 个 Go 微服务 + 4 个 Python 服务，网关是 bff-service（早期文章写"5 个微服务"为简化）
② 向量检索走 Elasticsearch（不是 Milvus）
③ GraphRAG 是 Python 引擎（不是 Go）
④ Agent 的 LLM 执行在 agent-wanwu（Python），Go 侧 pkg/wga 负责编排
⑤ 工作流引擎在独立仓库 wanwu-workflow
⑥ 本文（第 6 篇）为修订后内容，可放心阅读
一、这篇文章要解决什么问题？
在第 5 篇中，我们给 Agent 挂了 3 个内置工具（知识库、计算器、搜索）。但企业真实场景中，Agent 需要对接的系统远不止这些：
查 CRM 里的客户信息 调 ERP 下的采购订单 读 Jira 里的工单状态 操作数据库执行 SQL 调用内部微服务 API
如果每个系统都写一个"内置工具"，代码会膨胀到不可维护。而且每接一个新系统，都要改 Agent 服务代码、重新部署。
MCP（Model Context Protocol）就是为了解决这个问题而生的。它的核心思想：
把"工具"从 Agent 中解耦出来，变成独立的服务。Agent 通过标准协议发现和调用工具，就像浏览器通过 HTTP 访问网站一样。
没有 MCP：Agent 代码里硬编码 50 个工具 → 改一个工具要重新部署 Agent有了 MCP：Agent 通过协议动态发现工具 → 新工具上线，Agent 自动可用
万悟内置了MCP 广场，已接入 100+ 工具。今天我们把协议本身和万悟的实现都拆开看。
二、核心概念：用大白话讲清楚
2.1 MCP 是什么？一个类比
💡 类比：MCP 之于 AI 工具，就像 USB 之于外设。
没有 USB 之前：键盘用 PS/2 口，打印机用并口，鼠标用串口。每加一个设备，主板就要加一种接口。 有了 USB 之后：所有设备统一接口，即插即用。 MCP 就是 AI 工具的"USB 接口"。任何工具只要实现了 MCP 协议，就能被任何支持 MCP 的 Agent 调用，无需修改 Agent 代码。
2.2 MCP 的三个角色
internal/mcp-service/client/ | ||
┌─────────────────────────────────────────────────────────┐│ MCP Host (万悟 Agent) ││ ││ ┌───────────┐ ┌───────────┐ ┌───────────┐ ││ │MCP Client │ │MCP Client │ │MCP Client │ ... ││ └─────┬─────┘ └─────┬─────┘ └─────┬─────┘ │└────────┼───────────────┼───────────────┼────────────────┘│ │ │JSON-RPC 2.0 JSON-RPC 2.0 JSON-RPC 2.0│ │ │┌────▼────┐ ┌────▼────┐ ┌────▼────┐│MCP Server│ │MCP Server│ │MCP Server││ CRM工具 │ │ ERP工具 │ │自研工具 │└─────────┘ └─────────┘ └─────────┘
2.3 MCP 通信协议：JSON-RPC 2.0
MCP 底层使用JSON-RPC 2.0通信，支持两种传输方式：
一次典型的 MCP 交互：
// 1. Client → Server: 初始化握手{”jsonrpc”:”2.0”,”id”:1,”method”:”initialize”,”params”:{”protocolVersion”:”2024-11-05”,”capabilities”:{}}}// 2. Server → Client: 返回能力声明{”jsonrpc”:”2.0”,”id”:1,”result”:{”protocolVersion”:”2024-11-05”,”capabilities”:{”tools”:{}},”serverInfo”:{”name”:”crm-tool”,”version”:”1.0.0”}}}// 3. Client → Server: 列出可用工具{”jsonrpc”:”2.0”,”id”:2,”method”:”tools/list”}// 4. Server → Client: 返回工具列表（JSON Schema 格式）{”jsonrpc”:”2.0”,”id”:2,”result”:{”tools”:[{”name”:”query_customer”,”description”:”根据客户ID查询CRM信息”,”inputSchema”:{...}}]}}// 5. Client → Server: 调用工具{”jsonrpc”:”2.0”,”id”:3,”method”:”tools/call”,”params”:{”name”:”query_customer”,”arguments”:{”customer_id”:”C-10086”}}}// 6. Server → Client: 返回结果{”jsonrpc”:”2.0”,”id”:3,”result”:{”content”:[{”type”:”text”,”text”:”客户：张三，等级：VIP，最近订单：...”}]}}
📌 关键洞察：MCP 的工具描述格式和 OpenAI Function Calling 的 JSON Schema 几乎一致。这意味着 LLM 不需要"学习"新格式——它看到的工具描述和之前完全一样。MCP 只是把工具的"注册、发现、调用"从 Agent 代码中抽离到了独立服务。
2.4 MCP vs 直接 Function Calling
三、万悟是怎么实现的？（源码篇）
3.1 定位源码
⚠️ 诚实说明：以下 3.2–3.4 的代码是教学重建版——用最小代码讲清 MCP 协议代理与工具注册的核心逻辑。万悟真实的 MCP 相关代码分布在以下路径，结构更复杂（gRPC 微服务 + proto 定义 + DB 层），但设计思想完全一致。
# 万悟真实源码结构（Go 单体仓库）internal/mcp-service/# MCP 服务（gRPC 微服务）├── client/# MCP 客户端├── config/# MCP Server 配置└── server/grpc/# gRPC 服务入口pkg/mcp2skill/# MCP → Skill 自动转换（万悟特色！）pkg/openapi2skill/# OpenAPI → Skill 自动转换internal/bff-service/# BFF 网关（HTTP 入口，MCP 广场前端 API）# 万悟使用 ThinkInAIXYZ/go-mcp + mark3labs/mcp-go 两个 MCP 库
3.2 MCP Client 代理
// 教学重建版（对应真实路径：internal/mcp-service/client/）// MCPProxy 是 Agent 侧的 MCP 客户端// 它将 Agent 的工具调用请求转发给对应的 MCP Servertype MCPProxy struct {transport Transport // 传输层（stdio 或 SSE）serverInfo ServerInfo // 服务端信息tools []ToolSchema // 缓存的工具列表mu sync.RWMutex}// Initialize 与 MCP Server 建立连接并握手func(p *MCPProxy) Initialize(ctx context.Context) error {// 发送 initialize 请求resp, err := p.transport.Call(ctx, JSONRPCRequest{JSONRPC: ”2.0”,ID: 1,Method: ”initialize”,Params: map[string]interface{}{”protocolVersion”: ”2024-11-05”,”capabilities”: map[string]interface{}{},”clientInfo”: map[string]interface{}{”name”: ”wanwu-agent”,”version”: ”1.0.0”,},},})if err != nil {return fmt.Errorf(”mcp initialize failed: %w”, err)}// 解析服务端能力p.serverInfo = parseServerInfo(resp.Result)// 发送 initialized 通知（无需响应）p.transport.Notify(ctx, JSONRPCRequest{JSONRPC: ”2.0”,Method: ”notifications/initialized”,})return nil}// ListTools 获取 MCP Server 提供的工具列表func(p *MCPProxy) ListTools(ctx context.Context) ([]ToolSchema, error) {p.mu.RLock()if len(p.tools) > 0 {defer p.mu.RUnlock()return p.tools, nil // 使用缓存}p.mu.RUnlock()resp, err := p.transport.Call(ctx, JSONRPCRequest{JSONRPC: ”2.0”,ID: 2,Method: ”tools/list”,})if err != nil {return nil, err}var result struct {Tools []ToolSchema `json:”tools”`}json.Unmarshal(resp.Result, &result)// 缓存工具列表（TTL 5 分钟）p.mu.Lock()p.tools = result.Toolsp.mu.Unlock()return result.Tools, nil}// CallTool 调用 MCP Server 上的工具func(p *MCPProxy) CallTool(ctx context.Context, name string, args map[string]interface{}) (string, error) {resp, err := p.transport.Call(ctx, JSONRPCRequest{JSONRPC: ”2.0”,ID: generateID(),Method: ”tools/call”,Params: map[string]interface{}{”name”: name,”arguments”: args,},})if err != nil {return ””, fmt.Errorf(”mcp tool call failed: %w”, err)}// 解析结果（MCP 返回的是 content 数组）var result struct {Content []struct {Type string `json:”type”`Text string `json:”text”`} `json:”content”`IsError bool `json:”isError”`}json.Unmarshal(resp.Result, &result)if result.IsError {return ””, fmt.Errorf(”tool returned error: %s”, result.Content[0].Text)}// 拼接所有 text 类型的内容var sb strings.Builderfor _, c := range result.Content {if c.Type == ”text” {sb.WriteString(c.Text)}}return sb.String(), nil}
3.3 统一工具注册中心
// 教学重建版（对应真实路径：internal/mcp-service/ 工具注册逻辑）// Registry 统一管理内置工具和 MCP 工具// 对 Agent 引擎来说，两者没有区别type Registry struct {builtinTools map[string]BuiltinTool // 内置工具mcpProxies map[string]*MCPProxy // MCP Server 连接}// GetTools 返回某个 Agent 可用的所有工具// Agent 引擎调用这个方法获取工具列表，不关心工具来源func(r *Registry) GetTools(agentID string) []ToolSchema {var tools []ToolSchema// 1. 内置工具for _, t := range r.builtinTools {if t.IsEnabledFor(agentID) {tools = append(tools, t.Schema())}}// 2. MCP 工具（从各 MCP Server 动态获取）for serverName, proxy := range r.mcpProxies {mcpTools, err := proxy.ListTools(context.Background())if err != nil {log.Warnf(”list tools from mcp server %s failed: %v”, serverName, err)continue}// 给工具名加前缀，避免不同 Server 的工具名冲突for _, t := range mcpTools {t.Name = serverName + ”__” + t.Name // 如 ”crm__query_customer”tools = append(tools, t)}}return tools}// Execute 根据工具名路由到正确的执行器func(r *Registry) Execute(ctx context.Context, toolName string, args map[string]interface{}) (string, error) {// 判断是内置工具还是 MCP 工具if parts := strings.SplitN(toolName, ”__”, 2); len(parts) == 2 {// MCP 工具：serverName__toolNameserverName, realToolName := parts[0], parts[1]proxy, ok := r.mcpProxies[serverName]if !ok {return ””, fmt.Errorf(”mcp server not found: %s”, serverName)}return proxy.CallTool(ctx, realToolName, args)}// 内置工具tool, ok := r.builtinTools[toolName]if !ok {return ””, fmt.Errorf(”unknown tool: %s”, toolName)}return tool.Run(ctx, args)}
📌 设计亮点：
serverName__toolName的命名空间设计非常巧妙。不同 MCP Server 可能有同名工具（比如都有search），加前缀后变成crm__search和erp__search，彻底避免冲突。Agent 引擎完全不需要知道工具来自哪里。
3.4 MCP 网关与沙箱
// 教学重建版（对应真实路径：internal/mcp-service/ 沙箱执行逻辑）// Sandbox 为每个 MCP Server 提供隔离的执行环境// 防止恶意或有缺陷的工具影响主服务type Sandbox struct {containerID string // Docker 容器 ID（可选）process *exec.Cmd // 子进程（stdio 模式）resourceLimits ResourceLimits // CPU/内存/网络限制}// Start 启动一个 MCP Server 沙箱func(s *Sandbox) Start(ctx context.Context, config ServerConfig) error {if config.Transport == ”stdio” {// stdio 模式：启动子进程cmd := exec.CommandContext(ctx, config.Command, config.Args...)cmd.Env = append(os.Environ(), config.Env...)// 资源限制（Linux cgroups）cmd.SysProcAttr = &syscall.SysProcAttr{// 设置 CPU 和内存限制}stdin, _ := cmd.StdinPipe()stdout, _ := cmd.StdoutPipe()if err := cmd.Start(); err != nil {return fmt.Errorf(”start mcp server failed: %w”, err)}s.process = cmds.transport = NewStdioTransport(stdin, stdout)}if config.Transport == ”sse” {// SSE 模式：直接连接远程 HTTP 端点s.transport = NewSSETransport(config.URL, config.Headers)}return nil}
💡 安全设计：万悟的 MCP 网关支持两种隔离级别：
进程级隔离（stdio）：每个 MCP Server 是独立子进程，崩溃不影响主服务 容器级隔离（Docker）：生产环境推荐，完全隔离文件系统和网络
四、动手跑通（实践篇）
4.1 探索 MCP 广场
进入万悟平台 →「MCP 广场」 浏览已接入的工具分类：
选择一个工具（如 web_search），查看： - 工具描述和参数 Schema - 调用示例 - 所属 MCP Server 信息
▲ 万悟 MCP 广场：按分类浏览、搜索、一键启用工具
4.2 给 Agent 接入 MCP 工具
进入「智能体」→ 选择第 5 篇创建的 Agent 「工具配置」→「MCP 工具」 从广场中勾选： - ✅ web__search（网页搜索） - ✅mysql__query（数据库查询，需配置连接）保存 → 测试对话
4.3 开发一个自定义 MCP Server（核心实践！）
🎯 目标：用 Python 写一个"天气查询"MCP Server，接入万悟
# weather_mcp_server.py# 依赖：pip install mcp httpx# MCP Python SDK: https://github.com/modelcontextprotocol/python-sdkfrom mcp.server import Serverfrom mcp.server.stdio import stdio_serverfrom mcp.types import Tool, TextContentimport httpxapp = Server(”weather-tool”)# ===== 定义工具 =====@app.list_tools()async def list_tools() -> list[Tool]:return [Tool(name=”get_weather”,description=”查询指定城市的当前天气信息，包括温度、湿度、风力”,inputSchema={”type”: ”object”,”properties”: {”city”: {”type”: ”string”,”description”: ”城市名称，如'北京'、'上海'”}},”required”: [”city”]})]# ===== 实现工具逻辑 =====@app.call_tool()async def call_tool(name: str, arguments: dict) -> list[TextContent]:if name != ”get_weather”:raise ValueError(f”Unknown tool: {name}”)city = arguments[”city”]# 调用免费天气 API（示例用 wttr.in）async with httpx.AsyncClient() as client:resp = await client.get(f”https://wttr.in/{city}?format=j1”,timeout=10)data = resp.json()current = data[”current_condition”][0]result = (f”城市：{city}\n”f”温度：{current['temp_C']}°C（体感 {current['FeelsLikeC']}°C）\n”f”天气：{current['weatherDesc'][0]['value']}\n”f”湿度：{current['humidity']}%\n”f”风力：{current['windspeedKmph']} km/h {current['winddir16Point']}”)return [TextContent(type=”text”, text=result)]# ===== 启动服务 =====async def main():async with stdio_server() as (read_stream, write_stream):await app.run(read_stream, write_stream, app.create_initialization_options())if __name__ == ”__main__”:import asyncioasyncio.run(main())
4.4 注册到万悟
在万悟的 MCP Server 配置文件中添加：
# 对应真实路径：internal/mcp-service/config/（MCP Server 配置）servers:# ... 已有配置 ...- name: weathertransport: stdiocommand: python3args:- /opt/mcp-servers/weather_mcp_server.pyenv:- PYTHONPATH=/opt/mcp-serversenabled: true
重启 MCP 网关服务后，Agent 自动发现 weather__get_weather 工具。
4.5 测试
在 Agent 对话中：
用户：北京今天天气怎么样？适合出差吗？预期推理链：
Thought: 用户问天气，我有 weather 工具Action: weather__get_weather(city=”北京”)Observation: 城市：北京 / 温度：32°C / 天气：晴 / 湿度：45% ...Thought: 32度晴天，可以回答Final Answer: 北京今天晴，32°C，湿度45%，风力较小。天气适合出差，但注意防晒和补水。
4.6 对比实验
五、自己造一个 Mini 版（Deep Dive）
🎯 目标：不用 SDK，用 50 行 Python 实现一个最简 MCP Server（stdio 模式）
# mini_mcp_server.py# 零依赖！仅用标准库实现 MCP 协议（stdio 传输）import sys, jsondef handle_request(req):method = req.get(”method”)req_id = req.get(”id”)if method == ”initialize”:return {”jsonrpc”: ”2.0”, ”id”: req_id, ”result”: {”protocolVersion”: ”2024-11-05”,”capabilities”: {”tools”: {}},”serverInfo”: {”name”: ”mini-tool”, ”version”: ”0.1.0”}}}if method == ”tools/list”:return {”jsonrpc”: ”2.0”, ”id”: req_id, ”result”: {”tools”: [{”name”: ”add”,”description”: ”计算两个数的和”,”inputSchema”: {”type”: ”object”,”properties”: {”a”: {”type”: ”number”, ”description”: ”第一个数”},”b”: {”type”: ”number”, ”description”: ”第二个数”}},”required”: [”a”, ”b”]}}]}}if method == ”tools/call”:params = req.get(”params”, {})if params.get(”name”) == ”add”:args = params.get(”arguments”, {})result = args.get(”a”, 0) + args.get(”b”, 0)return {”jsonrpc”: ”2.0”, ”id”: req_id, ”result”: {”content”: [{”type”: ”text”, ”text”: str(result)}],”isError”: False}}return {”jsonrpc”: ”2.0”, ”id”: req_id, ”error”: {”code”: -32601, ”message”: ”Method not found”}}# 主循环：从 stdin 读 JSON-RPC，处理后写到 stdoutfor line in sys.stdin:line = line.strip()if not line:continuetry:req = json.loads(line)except json.JSONDecodeError:continue# 通知类消息（无 id）不需要响应if ”id” not in req:continueresp = handle_request(req)sys.stdout.write(json.dumps(resp) + ”\n”)sys.stdout.flush()
测试：
# 终端 1：启动 MCP Serverpython3 mini_mcp_server.py# 终端 2：手动发送请求（模拟 MCP Client）echo '{”jsonrpc”:”2.0”,”id”:1,”method”:”initialize”,”params”:{}}' | python3 mini_mcp_server.pyecho '{”jsonrpc”:”2.0”,”id”:2,”method”:”tools/list”}' | python3 mini_mcp_server.pyecho '{”jsonrpc”:”2.0”,”id”:3,”method”:”tools/call”,”params”:{”name”:”add”,”arguments”:{”a”:17,”b”:25}}}' | python3 mini_mcp_server.py# 预期输出：{”jsonrpc”:”2.0”,”id”:3,”result”:{”content”:[{”type”:”text”,”text”:”42”}],”isError”:false}}
🎉 这 50 行就是一个完整的 MCP Server。它实现了协议的三个核心方法：
initialize、tools/list、tools/call。万悟的 MCP 生态中，每个工具服务的骨架都是这样的——只是工具逻辑更复杂、传输方式可能是 SSE、加了认证和限流而已。
六、总结 & 延伸阅读
本文要点回顾
✅ MCP = AI 工具的"USB 接口"，实现工具与 Agent 的解耦 ✅ 三角色：Host（Agent）→ Client（协议代理）→ Server（工具服务） ✅ 底层通信：JSON-RPC 2.0，传输支持 stdio（本地）和 SSE（远程） ✅ 万悟通过 serverName__toolName命名空间实现多 Server 工具隔离✅ MCP 网关提供进程级/容器级沙箱，保障安全 ✅ 开发 MCP Server 只需实现 3 个方法：initialize / tools/list / tools/call
课后作业
在万悟 MCP 广场中启用 3 个不同分类的工具，测试 Agent 调用 用 Python MCP SDK 开发一个自定义工具（如查汇率、查快递） 将自定义工具注册到万悟，验证 Agent 能自动发现并调用 跑通 Mini MCP Server，理解 JSON-RPC 通信流程 思考：如果一个 MCP Server 挂了，Agent 应该怎么处理？（提示：熔断、降级）
下一篇预告
第 7 篇：工作流引擎：让 AI 按"流程图"干活我们将深入万悟的工作流模块，看看 DAG 编排、条件分支、并行执行是如何实现的，以及它和 Agent 的适用场景有何不同。
参考资源
元景万悟 GitHub：https://github.com/UnicomAI/wanwu MCP 官方规范：https://modelcontextprotocol.io/specification MCP Python SDK：https://github.com/modelcontextprotocol/python-sdk MCP TypeScript SDK：https://github.com/modelcontextprotocol/typescript-sdk JSON-RPC 2.0 规范：https://www.jsonrpc.org/specification
📱 关注公众号，追更不迷路
本系列文章首发于微信公众号「农夫三拳有点癫」，每周更新源码拆解与架构实战。
在微信扫描下方二维码即可关注：
