前言每加一个工具就要改 Agent 代码上一篇我们实现了 A2A 协议Agent 之间可以通信协作了。但还有一个没解决的问题工具管理。现在每加一个新工具需要写工具函数用register_tool注册重启 Agent祈祷 LLM 能理解新工具的描述MCPModel Context Protocol的目标是让工具变成可插拔的插件——Agent 启动时自动发现可用工具不需要改一行代码。一、MCP 协议的核心架构┌─────────────────────────────────────────────┐ │ MCP 架构简化版 │ ├─────────────────────────────────────────────┤ │ │ │ AgentMCP 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 协议实现VERSION2.0staticmethoddefrequest(method:str,params:dictNone,id:strNone)-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:dictNone)-dict:return{jsonrpc:JSONRPC.VERSION,error:{code:code,message:message,data:dataor{}},id:id,}staticmethoddefis_request(msg:dict)-bool:returnmethodinmsgstaticmethoddefis_response(msg:dict)-bool:returnresultinmsgorerrorinmsgMCP 标准方法定义classMCPMethods:MCP 协议规定的标准方法INITIALIZEinitialize# 初始化连接TOOLS_LISTtools/list# 获取工具列表TOOLS_CALLtools/call# 调用工具RESOURCES_LISTresources/list# 获取资源列表PINGping# 心跳三、MCP Server工具的商店货架classMCPServer:MCP 服务端——暴露工具给 Agent 调用def__init__(self,name:str,version:str1.0.0):self.namename self.versionversion 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 Serverself.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 请求路由到对应方法methodraw.get(method,)req_idraw.get(id,)handlerself._method_handlers.get(method)ifnothandler:returnJSONRPC.error(req_id,-32601,fMethod not found:{method})try:resulthandler(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_nameparams.get(name,)argumentsparams.get(arguments,{})toolself.tools.get(tool_name)ifnottool:raiseValueError(fTool not found:{tool_name})resulttool[handler](**arguments)return{content:[{type:text,text:str(result)}],isError:False,}def_handle_ping(self,params:dict)-dict:return{status:ok}四、MCP ClientAgent 的工具发现引擎classMCPClient:MCP 客户端——Agent 通过 Client 发现和调用远程工具def__init__(self,agent:IntegratedAgent):self.agentagent self.servers:dict[str,MCPServer]{}# 本示例用内存直连生产用 HTTPself.discovered_tools:dict[str,dict]{}# tool_name - {server, definition}defconnect_server(self,server:MCPServer):连接 MCP Serverself.servers[server.name]server# 发送 initialize 请求self._send_and_receive(server,MCPMethods.INITIALIZE,{})# 自动发现工具respself._send_and_receive(server,MCPMethods.TOOLS_LIST,{})toolsresp.get(result,{}).get(tools,[])fortool_defintools:nametool_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 请求并等待响应requestJSONRPC.request(method,params)returnserver.handle_request(request)defcall_tool(self,tool_name:str,arguments:dict)-str:调用远程工具infoself.discovered_tools.get(tool_name)ifnotinfo:returnfError: Tool {tool_name} not foundserverself.servers[info[server]]respself._send_and_receive(server,MCPMethods.TOOLS_CALL,{name:tool_name,arguments:arguments,})iferrorinresp:returnfMCP Error:{resp[error][message]}contentresp.get(result,{}).get(content,[])ifcontent:returncontent[0].get(text,str(content))returnstr(resp)def_register_to_agent(self):将发现的工具注册到 Agent 的工具系统forname,infoinself.discovered_tools.items():schemainfo[definition].get(inputSchema,{})propsschema.get(properties,{})# 为每个远程工具创建一个包装函数defmake_wrapper(tool_name):defwrapper(**kwargs):returnself.call_tool(tool_name,kwargs)returnwrapper# 注册到 Agentfromtoolsimportregister_tool,TOOL_REGISTRY register_tool(namename,descriptioninfo[definition][description],parameters[{name:pname,type:pinfo.get(type,string),description:pinfo.get(description,),required:pnameinschema.get(required,[])}forpname,pinfoinprops.items()],categorymcp,)(make_wrapper(name))五、完整示例搭建两个 MCP Server# 创建搜索服务 search_serverMCPServer(search-service)search_server.register_tool(nameweb_search,description搜索互联网获取信息。用于查询最新新闻、事实、人物。,parameters[{name:query,type:string,description:搜索关键词,required:True},{name:max_results,type:number,description:结果数默认5,required:False},],handlerlambdaquery,max_results5:f搜索结果({max_results}条): ...)# 创建通知服务 notify_serverMCPServer(notification-service)notify_server.register_tool(namesend_email,description发送邮件。用于通知、报告分发。,parameters[{name:to,type:string,description:收件人邮箱,required:True},{name:subject,type:string,description:邮件主题,required:True},{name:body,type:string,description:邮件正文,required:True},],handlerlambdato,subject,body:f邮件已发送至{to})# Agent 连接 Server agentIntegratedAgent(tools{},llm_clientclient)mcpMCPClient(agent)mcp.connect_server(search_server)mcp.connect_server(notify_server)# Agent 现在自动拥有了 web_search 和 send_email 两个工具# 不需要改 Agent 代码resultagent.run(搜索今天AI新闻并发送摘要到bosszyentor.com)六、MCP vs Function Calling 对比维度MCPFunction Calling原生工具管理热插拔启动时自动发现硬编码需要重启跨 Agent 共享一个 Server 多个 Agent 共享每个 Agent 独立配置安全隔离Server 级沙箱函数级校验协议标准JSON-RPC 2.0业界标准各厂商自定义生态可发布到 MCP Registry无生态延迟多一层 RPC 调用5-20ms直接函数调用MCP 的代价是多一层 RPC 调用但换来的是工具生态的标准化。对于生产环境这个代价完全值得。七、总结MCP 协议的核心价值让工具变成插件让 Agent 拥有应用商店。JSON-RPC 通信轻量、标准化、业经验证Server/Client 架构工具服务化一个 Server 多个 Agent 共享自动发现Agent 启动时自动拉取可用工具零代码扩展安全隔离每个 MCP Server 是独立的沙箱和 Function Calling 的关系MCP 是工具的上层管理和分发机制Function Calling 是底层调用协议。两者互补——MCP 管有哪些工具Function Calling 管怎么调用工具。如果觉得有用欢迎点赞 收藏 关注。下一篇文章将是这个系列的阶段性总结。《手搓生产级 AI Agent 系统》系列第 5 篇上一篇A2A 协议——让两个 Agent 互相通信 下一篇系列总结 生产级 Agent 架构全景