MCP 协议从零实现：打造 Agent 的“应用商店“——工具热插拔与标准化生态

前言：每加一个工具就要改 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 作为通信协议：

importjson,uuid,timefromdataclassesimportdataclass,asdictfromtypingimportOptional,CallablefromenumimportEnumclassJSONRPC:"""JSON-RPC 2.0 协议实现"""VERSION="2.0"@staticmethoddefrequest(method:str,params:dict=None,id:str=None)->dict:return{"jsonrpc":JSONRPC.VERSION,"method":method,"params":paramsor{},"id":idorstr(uuid.uuid4())[:8],}@staticmethoddefresponse(id:str,result:dict)->dict:return{"jsonrpc":JSONRPC.VERSION,"result":result,"id":id,}@staticmethoddeferror(id:str,code:int,message:str,data:dict=None)->dict:return{"jsonrpc":JSONRPC.VERSION,"error":{"code":code,"message":message,"data":dataor{}},"id":id,}@staticmethoddefis_request(msg:dict)->bool:return"method"inmsg@staticmethoddefis_response(msg:dict)->bool:return"result"inmsgor"error"inmsg

MCP 标准方法定义

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

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

classMCPServer:"""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,}defregister_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",""),}forpinparameters},"required":[p["name"]forpinparametersifp.get("required",True)],},},"handler":handler,}defhandle_request(self,raw:dict)->dict:"""处理 JSON-RPC 请求，路由到对应方法"""method=raw.get("method","")req_id=raw.get("id","")handler=self._method_handlers.get(method)ifnothandler:returnJSONRPC.error(req_id,-32601,f"Method not found:{method}")try:result=handler(raw.get("params",{}))returnJSONRPC.response(req_id,result)exceptExceptionase:returnJSONRPC.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"]fortinself.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)ifnottool:raiseValueError(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 的工具发现引擎

classMCPClient:"""MCP 客户端——Agent 通过 Client 发现和调用远程工具"""def__init__(self,agent:IntegratedAgent):self.agent=agent self.servers:dict[str,MCPServer]={}# 本示例用内存直连，生产用 HTTPself.discovered_tools:dict[str,dict]={}# tool_name -> {server, definition}defconnect_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",[])fortool_defintools: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)returnserver.handle_request(request)defcall_tool(self,tool_name:str,arguments:dict)->str:"""调用远程工具"""info=self.discovered_tools.get(tool_name)ifnotinfo:returnf"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"inresp:returnf"MCP Error:{resp['error']['message']}"content=resp.get("result",{}).get("content",[])ifcontent:returncontent[0].get("text",str(content))returnstr(resp)def_register_to_agent(self):"""将发现的工具注册到 Agent 的工具系统"""forname,infoinself.discovered_tools.items():schema=info["definition"].get("inputSchema",{})props=schema.get("properties",{})# 为每个远程工具创建一个包装函数defmake_wrapper(tool_name):defwrapper(**kwargs):returnself.call_tool(tool_name,kwargs)returnwrapper# 注册到 Agentfromtoolsimportregister_tool,TOOL_REGISTRY register_tool(name=name,description=info["definition"]["description"],parameters=[{"name":pname,"type":pinfo.get("type","string"),"description":pinfo.get("description",""),"required":pnameinschema.get("required",[])}forpname,pinfoinprops.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=lambdaquery,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=lambdato,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 的代价是多一层 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 架构全景