前言：每加一个工具就要改 Agent 代码？

上一篇我们实现了 A2A 协议，Agent 之间可以通信协作了。但还有一个没解决的问题：工具管理。

现在每加一个新工具，需要：
1. 写工具函数
2. 用 @register_tool 注册
3. 重启 Agent
4. 祈祷 LLM 能理解新工具的描述

MCP（Model Context Protocol）的目标是让工具变成可插拔的插件——Agent 启动时自动发现可用工具，不需要改一行代码。

一、MCP 协议的核心架构

┌─────────────────────────────────────────────┐
│              MCP 架构（简化版）               │
├─────────────────────────────────────────────┤
│                                              │
│  Agent（MCP Client）                         │
│  ├─ 工具发现 → 向 Server 请求工具列表         │
│  ├─ 工具调用 → 通过 JSON-RPC 调用远程工具     │
│  └─ 自动注册 → 下载的工具自动加入 TOOLS 字典  │
│                                              │
│  MCP Server A          MCP Server B          │
│  ├─ search             ├─ send_email         │
│  ├─ calculator         ├─ create_sheet       │
│  └─ read_file          └─ generate_report    │
│                                              │
│  通信协议：JSON-RPC 2.0 over HTTP/stdio      │
└─────────────────────────────────────────────┘

二、JSON-RPC 通信层

MCP 使用 JSON-RPC 2.0 作为通信协议：

import json, uuid, time
from dataclasses import dataclass, asdict
from typing import Optional, Callable
from enum import Enum

class JSONRPC:
    """JSON-RPC 2.0 协议实现"""
    VERSION = "2.0"

    @staticmethod
    def request(method: str, params: dict = None, id: str = None) -> dict:
        return {
            "jsonrpc": JSONRPC.VERSION,
            "method": method,
            "params": params or {},
            "id": id or str(uuid.uuid4())[:8],
        }

    @staticmethod
    def response(id: str, result: dict) -> dict:
        return {
            "jsonrpc": JSONRPC.VERSION,
            "result": result,
            "id": id,
        }

    @staticmethod
    def error(id: str, code: int, message: str, data: dict = None) -> dict:
        return {
            "jsonrpc": JSONRPC.VERSION,
            "error": {"code": code, "message": message, "data": data or {}},
            "id": id,
        }

    @staticmethod
    def is_request(msg: dict) -> bool:
        return "method" in msg

    @staticmethod
    def is_response(msg: dict) -> bool:
        return "result" in msg or "error" in msg

MCP 标准方法定义

class MCPMethods:
    """MCP 协议规定的标准方法"""
    INITIALIZE = "initialize"       # 初始化连接
    TOOLS_LIST = "tools/list"       # 获取工具列表
    TOOLS_CALL = "tools/call"       # 调用工具
    RESOURCES_LIST = "resources/list"  # 获取资源列表
    PING = "ping"                   # 心跳

三、MCP Server：工具的"商店货架"

class MCPServer:
    """MCP 服务端——暴露工具给 Agent 调用"""

    def __init__(self, name: str, version: str = "1.0.0"):
        self.name = name
        self.version = version
        self.tools: dict[str, dict] = {}  # tool_name -> {definition, handler}
        self._method_handlers: dict[str, Callable] = {
            MCPMethods.INITIALIZE: self._handle_initialize,
            MCPMethods.TOOLS_LIST: self._handle_tools_list,
            MCPMethods.TOOLS_CALL: self._handle_tools_call,
            MCPMethods.PING: self._handle_ping,
        }

    def register_tool(self, name: str, description: str,
                      parameters: list[dict], handler: Callable):
        """注册一个工具到这个 MCP Server"""
        self.tools[name] = {
            "definition": {
                "name": name,
                "description": description,
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        p["name"]: {
                            "type": p.get("type", "string"),
                            "description": p.get("description", ""),
                        }
                        for p in parameters
                    },
                    "required": [p["name"] for p in parameters if p.get("required", True)],
                },
            },
            "handler": handler,
        }

    def handle_request(self, raw: dict) -> dict:
        """处理 JSON-RPC 请求，路由到对应方法"""
        method = raw.get("method", "")
        req_id = raw.get("id", "")

        handler = self._method_handlers.get(method)
        if not handler:
            return JSONRPC.error(req_id, -32601, f"Method not found: {method}")

        try:
            result = handler(raw.get("params", {}))
            return JSONRPC.response(req_id, result)
        except Exception as e:
            return JSONRPC.error(req_id, -32000, str(e))

    def _handle_initialize(self, params: dict) -> dict:
        return {
            "protocolVersion": "2024-11-05",
            "serverInfo": {"name": self.name, "version": self.version},
            "capabilities": {"tools": {}},
        }

    def _handle_tools_list(self, params: dict) -> dict:
        return {
            "tools": [t["definition"] for t in self.tools.values()]
        }

    def _handle_tools_call(self, params: dict) -> dict:
        tool_name = params.get("name", "")
        arguments = params.get("arguments", {})

        tool = self.tools.get(tool_name)
        if not tool:
            raise ValueError(f"Tool not found: {tool_name}")

        result = tool["handler"](**arguments)
        return {
            "content": [{"type": "text", "text": str(result)}],
            "isError": False,
        }

    def _handle_ping(self, params: dict) -> dict:
        return {"status": "ok"}

四、MCP Client：Agent 的工具发现引擎

class MCPClient:
    """MCP 客户端——Agent 通过 Client 发现和调用远程工具"""

    def __init__(self, agent: IntegratedAgent):
        self.agent = agent
        self.servers: dict[str, MCPServer] = {}  # 本示例用内存直连，生产用 HTTP
        self.discovered_tools: dict[str, dict] = {}  # tool_name -> {server, definition}

    def connect_server(self, server: MCPServer):
        """连接 MCP Server"""
        self.servers[server.name] = server

        # 发送 initialize 请求
        self._send_and_receive(server, MCPMethods.INITIALIZE, {})

        # 自动发现工具
        resp = self._send_and_receive(server, MCPMethods.TOOLS_LIST, {})
        tools = resp.get("result", {}).get("tools", [])

        for tool_def in tools:
            name = tool_def["name"]
            self.discovered_tools[name] = {
                "server": server.name,
                "definition": tool_def,
            }
            print(f"  [DISCOVER] {name} from {server.name}")

        # 自动注册到 Agent 的工具系统
        self._register_to_agent()

    def _send_and_receive(self, server: MCPServer,
                          method: str, params: dict) -> dict:
        """发送 JSON-RPC 请求并等待响应"""
        request = JSONRPC.request(method, params)
        return server.handle_request(request)

    def call_tool(self, tool_name: str, arguments: dict) -> str:
        """调用远程工具"""
        info = self.discovered_tools.get(tool_name)
        if not info:
            return f"Error: Tool '{tool_name}' not found"

        server = self.servers[info["server"]]
        resp = self._send_and_receive(server, MCPMethods.TOOLS_CALL, {
            "name": tool_name,
            "arguments": arguments,
        })

        if "error" in resp:
            return f"MCP Error: {resp['error']['message']}"

        content = resp.get("result", {}).get("content", [])
        if content:
            return content[0].get("text", str(content))
        return str(resp)

    def _register_to_agent(self):
        """将发现的工具注册到 Agent 的工具系统"""
        for name, info in self.discovered_tools.items():
            schema = info["definition"].get("inputSchema", {})
            props = schema.get("properties", {})

            # 为每个远程工具创建一个包装函数
            def make_wrapper(tool_name):
                def wrapper(**kwargs):
                    return self.call_tool(tool_name, kwargs)
                return wrapper

            # 注册到 Agent
            from tools import register_tool, TOOL_REGISTRY
            register_tool(
                name=name,
                description=info["definition"]["description"],
                parameters=[
                    {"name": pname, "type": pinfo.get("type", "string"),
                     "description": pinfo.get("description", ""),
                     "required": pname in schema.get("required", [])}
                    for pname, pinfo in props.items()
                ],
                category="mcp",
            )(make_wrapper(name))

五、完整示例：搭建两个 MCP Server

# ====== 创建搜索服务 ======
search_server = MCPServer("search-service")
search_server.register_tool(
    name="web_search",
    description="搜索互联网获取信息。用于查询最新新闻、事实、人物。",
    parameters=[
        {"name": "query", "type": "string",
         "description": "搜索关键词", "required": True},
        {"name": "max_results", "type": "number",
         "description": "结果数，默认5", "required": False},
    ],
    handler=lambda query, max_results=5: f"搜索结果({max_results}条): ..."
)

# ====== 创建通知服务 ======
notify_server = MCPServer("notification-service")
notify_server.register_tool(
    name="send_email",
    description="发送邮件。用于通知、报告分发。",
    parameters=[
        {"name": "to", "type": "string",
         "description": "收件人邮箱", "required": True},
        {"name": "subject", "type": "string",
         "description": "邮件主题", "required": True},
        {"name": "body", "type": "string",
         "description": "邮件正文", "required": True},
    ],
    handler=lambda to, subject, body: f"邮件已发送至{to}"
)

# ====== Agent 连接 Server ======
agent = IntegratedAgent(tools={}, llm_client=client)
mcp = MCPClient(agent)
mcp.connect_server(search_server)
mcp.connect_server(notify_server)

# Agent 现在自动拥有了 web_search 和 send_email 两个工具
# 不需要改 Agent 代码
result = agent.run("搜索今天AI新闻，并发送摘要到boss@zyentor.com")

六、MCP vs Function Calling 对比

维度 MCP Function Calling（原生）
工具管理 热插拔，启动时自动发现 硬编码，需要重启
跨 Agent 共享 一个 Server 多个 Agent 共享 每个 Agent 独立配置
安全隔离 Server 级沙箱 函数级校验
协议标准 JSON-RPC 2.0，业界标准 各厂商自定义
生态 可发布到 MCP Registry 无生态
延迟 多一层 RPC 调用（+5-20ms） 直接函数调用

MCP 的代价是多一层 RPC 调用，但换来的是工具生态的标准化。 对于生产环境，这个代价完全值得。

七、总结

MCP 协议的核心价值：让工具变成插件，让 Agent 拥有"应用商店"。

  1. JSON-RPC 通信：轻量、标准化、业经验证
  2. Server/Client 架构：工具服务化，一个 Server 多个 Agent 共享
  3. 自动发现：Agent 启动时自动拉取可用工具，零代码扩展
  4. 安全隔离：每个 MCP Server 是独立的沙箱

和 Function Calling 的关系：MCP 是工具的上层管理和分发机制，Function Calling 是底层调用协议。两者互补——MCP 管"有哪些工具"，Function Calling 管"怎么调用工具"。

如果觉得有用，欢迎 点赞 + 收藏 + 关注。下一篇文章将是这个系列的阶段性总结。

📌 《手搓生产级 AI Agent 系统》系列第 5 篇
上一篇：A2A 协议——让两个 Agent 互相通信
🔜 下一篇：系列总结 + 生产级 Agent 架构全景