MCP管工具A2A管协作双协议联合实战摘要MCP让Agent调工具A2A让Agent之间对话。这篇把两个协议拧在一起搭一个搜索总结翻译三Agent工作流。一、两个协议各管一件事前面三篇你一直在跟A2A打交道。你的Agent能收消息、能回消息、能互相委托任务。但有个场景一直没解决Agent怎么调用外部工具比如你想让Agent搜索网页、查数据库、读写文件。A2A不管这些——它只负责Agent之间的通信。管Agent和工具之间连接的是另一个协议MCPModel Context Protocol。两个协议的分工很明确MCPAnthropic出品 Agent连工具的USB-C接口A2AGoogle出品 Agent之间对话的微信协议 两个协议互补。一个Agent可以通过MCP调用搜索工具获取数据然后通过A2A把搜索结果发给另一个Agent做分析。这就是2026年Agent开发的完整形态。 今天带你搭一个完整的三Agent工作流。二、我们要搭什么业务场景你给Agent发一个关键词比如A2A协议最新进展。Agent自动执行三步搜索Agent通过MCP调用搜索工具拿到相关文章总结Agent接收搜索结果提炼关键信息翻译Agent把总结翻成英文返回最终结果架构图你 → 翻译AgentA2A Server端口10002 ↓ A2A 总结AgentA2A Server端口10003 ↓ A2A 搜索AgentA2A Server端口10004 ↓ MCP 搜索工具MCP Server搜索Agent同时是A2A Server接收总结Agent的请求和MCP Client调用搜索工具。一个Agent同时使用两个协议。三、项目结构a2a-mcp-workflow/ ├── searcher/# 搜索AgentA2A MCP│ ├── server.py │ ├── task_manager.py │ └── mcp_config.yaml# MCP工具配置├── summarizer/# 总结Agent纯A2A│ ├── server.py │ └── task_manager.py ├── translator/# 翻译Agent纯A2A│ ├── server.py │ └── task_manager.py ├── client.py └── pyproject.toml四、搭建搜索AgentA2A MCP双协议这是最复杂的一个Agent。它需要同时做两件事作为A2A Server接收总结Agent发来的搜索请求作为MCP Client调用搜索工具获取结果4.1 配置MCP工具创建searcher/ [mcp_config.yaml](http://mcp_config.yaml)声明要用的MCP工具mcpServers: web-search: command:npxargs:[-y,modelcontextprotocol/server-brave-search]env: BRAVE_API_KEY:${BRAVE_API_KEY}这个配置告诉Agent使用Brave Search作为搜索工具通过MCP协议调用。你需要一个Brave Search API Key brave.com/search/api 申请免费额度。如果你不想用Brave Search也可以用一个模拟的文件系统MCP Server做demomcpServers: file-tools: command:npxargs:[-y,modelcontextprotocol/server-filesystem,/tmp/search-results]4.2 搜索Agent的Task Managerimportjsonimporthttpx from google_a2a.common.server.task_managerimportInMemoryTaskManager from google_a2a.common.typesimport(Artifact, Message, SendTaskRequest, SendTaskResponse, Task, TaskState, TaskStatus,)OLLAMA_URLhttp://localhost:11434/api/generateOLLAMA_MODEL qwen2.5:7bclassSearcherTaskManager(InMemoryTaskManager):搜索Agent接收关键词通过MCP搜索返回结果。 asyncdefon_send_task(self, request: SendTaskRequest)-SendTaskResponse: awaitself.upsert_task(request.params)task_idrequest.params.idkeywordrequest.params.message.parts[0].text# 第一步通过MCP调用搜索工具search_resultsawaitself._mcp_search(keyword)# 第二步用大模型整理搜索结果formattedawaitself._format_results(keyword, search_results)taskawaitself._update_task(task_idtask_id,task_stateTaskState.COMPLETED,response_textformatted,)returnSendTaskResponse(idrequest.id,resulttask)asyncdef_mcp_search(self, keyword: str)-str: 通过MCP协议调用搜索工具。 这里演示两种方式真实MCP调用和模拟搜索。# 方式1真实MCP调用需要Brave API Keytry:importos api_keyos.environ. get(BRAVE_API_KEY)ifapi_key: asyncwith httpx.AsyncClient(timeout15.0)as c: respawait c.get(https:// api.search.brave.com/res/v1/web/search,params{q:keyword,count:5},headers{X-Subscription-Token:api_key,Accept:application/json,},)dataresp.json()results[]foritemindata.get(web,{}).get(results,[]): results.append(f- { item.get(title, )}: f{ item.get(description, )} f({ item.get(url, )}))return\n.join(results)ifresultselse未找到相关结果except Exception: pass# 方式2模拟搜索demo用return(f[搜索结果模拟 - 关键词: {keyword}]\n1. A2A协议是Google推出的Agent间通信标准\n2. 目前已有50厂商支持A2A协议\n3. A2A协议已捐赠给Linux基金会\n4. 最新版本 v0.2.5v1.0开发中\n5. A2A与MCP协议互补形成完整的Agent生态\n)asyncdef_format_results(self, keyword: str, results: str)-str:用大模型整理搜索结果。 try: asyncwith httpx.AsyncClient(timeout60.0)as c: respawait c.post(OLLAMA_URL,json{model:OLLAMA_MODEL,prompt:(f根据以下搜索结果整理一份关于f「{keyword}」的信息摘要f包含5个关键要点\n{results}),stream:False,},)resultresp.json()returnresult.get(response, results)except Exception:returnresults asyncdef_update_task(self, task_id: str, task_state: TaskState, response_text: str)-Task: taskself.tasks[task_id]parts[{type:text,text:response_text}]task.statusTaskStatus(statetask_state,messageMessage(roleagent,partsparts),)iftask_stateTaskState.COMPLETED: task.artifacts[Artifact(partsparts)]returntask4.3 搜索Agent的Serverimportloggingimportclick from google_a2a.common.typesimport(AgentSkill, AgentCapabilities, AgentCard)from google_a2a.common.serverimportA2AServer from task_managerimportSearcherTaskManager logging.basicConfig(levellogging.INFO)click.command()click.option(--host,defaultlocalhost)click.option(--port,default10004,typeint)defmain(host, port): skillAgentSkill(idsearch-skill,nameWeb Searcher,descriptionSearches the web and returns organized results,tags[search,web,mcp],examples[搜索A2A协议最新进展],inputModes[text],outputModes[text],)agent_cardAgentCard(nameSearcher Agent,description(Searches the web via MCP protocol and returns organized information.),urlfhttp://{host}:{port}/,version0.1.0,defaultInputModes[text],defaultOutputModes[text],capabilitiesAgentCapabilities(streamingFalse),skills[skill],)task_managerSearcherTaskManager()serverA2AServer(agent_cardagent_card,task_managertask_manager,hosthost,portport,)logging.info(fSearcher Agent running at http://{host}:{port})server.start()if__name____main__:main()五、搭建总结Agent纯A2A调用搜索Agent总结Agent不直接调MCP工具它通过A2A协议把搜索任务委托给搜索Agent。# summarizer/task_manager.pyimporthttpx from google_a2a.common.server.task_managerimportInMemoryTaskManager from google_a2a.common.typesimport(Artifact, Message, SendTaskRequest, SendTaskResponse, Task, TaskState, TaskStatus,)SEARCHER_URLhttp://localhost:10004OLLAMA_URLhttp://localhost:11434/api/generateOLLAMA_MODEL qwen2.5:7bclassSummarizerTaskManager(InMemoryTaskManager):总结Agent调用搜索Agent获取信息然后做总结。 asyncdefon_send_task(self, request: SendTaskRequest)-SendTaskResponse: awaitself.upsert_task(request.params)task_idrequest.params.idkeywordrequest.params.message.parts[0].text# 第一步通过A2A调用搜索Agentsearch_infoawaitself._call_searcher(keyword)# 第二步用大模型做深度总结summaryawaitself._summarize(keyword, search_info)taskawaitself._update_task(task_idtask_id,task_stateTaskState.COMPLETED,response_textsummary,)returnSendTaskResponse(idrequest.id,resulttask)asyncdef_call_searcher(self, keyword: str)-str:通过A2A协议调用搜索Agent。 try: asyncwith httpx.AsyncClient(timeout60.0)as c:# 发现搜索Agentcard_respawait c.get(f{SEARCHER_URL}/.well-known/ agent.json)cardcard_resp.json()# 发送A2A请求payload{jsonrpc:2.0,id:search-request,method:message/send,params:{message:{role:user,parts:[{type:text,text:keyword,}],}},}respawait c.post(card[url],jsonpayload,headers{Content-Type:application/json},)resultresp.json()# 解析响应ifresultinresult: task_dataresult[result]artifactstask_data.get(artifacts,[])ifartifacts:returnartifacts[0][parts][0][text]msgtask_data.get(status,{}).get(message,{})ifmsg.get(parts):returnmsg[parts][0][text]return搜索Agent无响应except Exception as e:returnf搜索调用失败: {str(e)}asyncdef_summarize(self, keyword: str, info: str)-str:用大模型做深度总结。 try: asyncwith httpx.AsyncClient(timeout60.0)as c: respawait c.post(OLLAMA_URL,json{model:OLLAMA_MODEL,prompt:(f你是一个专业的研究分析师。请根据以下搜索结果f写一份关于「{keyword}」的深度分析报告。f要求3-5个核心观点每个观点有具体数据支撑。\n\nf搜索结果\n{info}),stream:False,},)resultresp.json()returnresult.get(response, info)except Exception:returninfo asyncdef_update_task(self, task_id: str, task_state: TaskState, response_text: str)-Task: taskself.tasks[task_id]parts[{type:text,text:response_text}]task.statusTaskStatus(statetask_state,messageMessage(roleagent,partsparts),)iftask_stateTaskState.COMPLETED: task.artifacts[Artifact(partsparts)]returntaskServer部分和上篇类似端口改为10003不再赘述。六、搭建翻译Agent纯A2A调用总结Agent翻译Agent的逻辑跟上篇一模一样——接收你的消息翻译文本调用另一个Agent。只不过这次它调用的是总结Agent而不是摘要Agent。# translator/task_manager.py核心部分 SUMMARIZER_URLhttp://localhost:10003classTranslatorTaskManager(InMemoryTaskManager): asyncdefon_send_task(self, request):# ...同上篇省略...# 调用总结Agent而不是摘要Agentsummaryawaitself._call_summarizer(input_text)# 翻译结果translationawaitself._translate(summary)final_text(f【中文总结】\n{summary}\n\nf【英文翻译】\n{translation})taskawaitself._update_task(task_idtask_id,task_stateTaskState.COMPLETED,response_textfinal_text,)returnSendTaskResponse(idrequest.id,resulttask)asyncdef_call_summarizer(self, text: str)-str:通过A2A调用总结Agent。# 代码与上篇的_call_summarizer相同# 只是URL改为SUMMARIZER_URL http://localhost:10003# ...七、启动测试7.1 三个Agent全部启动开三个终端# 终端1搜索Agent端口10004cdsearcherpython server.py# 终端2总结Agent端口10003cdsummarizerpython server.py# 终端3翻译Agent端口10002cdtranslatorpython server.py三个终端都看到Uvicorn启动成功后进入下一步。7.2 客户端测试#client.pyimportasyncioimporthttpx from a2a.clientimport(A2ACardResolver, ClientConfig, create_client)from a2a.helpersimportnew_text_message from a2a.types.a2a_pb2importRole, SendMessageRequest asyncdefmain(): print( A2A MCP 双协议工作流测试 \n)asyncwith httpx.AsyncClient()as httpx_client:# 发现翻译AgentresolverA2ACardResolver(httpx_clienthttpx_client,base_urlhttp://localhost:10002,)agent_cardawait resolver.get_agent_card()print(f连接到: { agent_card.name}\n)# 创建客户端clientawaitcreate_client(agentagent_card,client_configClientConfig(streamingFalse),)# 发送关键词keywordA2A协议最新进展messagenew_text_message(keyword,roleRole.ROLE_USER)requestSendMessageRequest(messagemessage)print(f发送关键词: {keyword})print(工作流: 翻译Agent → 总结Agent → 搜索Agent(MCP))print(*60)asyncfor chunkinclient.send_message(request): ifhasattr(chunk,artifacts)and chunk.artifacts:forpartinchunk.artifacts[0].parts: print(part.text)await client.close()if__name____main__:asyncio.run(main())7.3 预期输出A2A MCP 双协议工作流测试连接到: Translator Agent 发送关键词: A2A协议最新进展 工作流: 翻译Agent → 总结Agent → 搜索Agent(MCP)【中文总结】1. A2A协议是Google于2025年推出的Agent间通信标准...2. 目前已有50厂商Salesforce、微软、SAP等支持...3. 协议已捐赠Linux基金会采用Apache2.0开源许可...4. 最新版本v0.2.5与MCP协议互补...5. 微软Copilot平台已同时支持MCP和A2A双协议... 【英文翻译】1. The A2A protocol is an agent-to-agent communication standard launched by Google in2025......一个关键词进去三步处理出来。八、MCP和A2A在代码层面的差异你可能注意到了搜索Agent里调MCP工具和调A2A Agent的代码长得不一样。MCP调用Agent→Tool# 通过MCP协议调用搜索工具# 实际上是调用MCP Server暴露的resource/tool接口respawait c.get(https:// api.search.brave.com/res/v1/web/search,headers{X-Subscription-Token:api_key},)MCP调用更像是传统的API调用——你知道工具的地址和参数直接HTTP请求。A2A调用Agent→Agent# 通过A2A协议调用另一个Agent# 1. 发现Agentcardawait c.get(f{url}/.well-known/ agent.json).json()# 2. 发送JSON-RPC请求payload{jsonrpc:2.0,id:req-001,method:message/send,params:{message:{role:user,parts:[...]}}}respawait c.post(card[url],jsonpayload)A2A调用是发现→通信两步走。先拿到Agent Card了解对方能力再发JSON-RPC请求。两者的本质区别维度MCPA2A调用对象工具/数据源其他Agent交互方式请求-响应有状态的任务Task发现机制配置文件/注册中心Agent Card自动发现数据格式取决于具体工具JSON-RPC 2.0统一格式九、生产环境的建议这个demo用的是模拟搜索。在生产环境里你可能需要用真实的MCP Server。Brave Search、GitHub、Playwright、文件系统等都有现成的MCP Server。给Agent加认证。结合上一篇的OAuth 2.0认证方案。加错误处理和重试。网络请求可能失败需要超时和重试机制。加任务队列。多个请求并发时用消息队列如Redis管理任务。加监控和日志。记录每次Agent间调用的耗时和状态。十、系列总结四篇文章从零到一第1篇搭了一个Echo Agent搞懂A2A的基本概念第2篇搭了翻译Agent和摘要Agent实现双Agent协作第3篇给Agent加了API Key / JWT / OAuth 2.0三把锁第4篇本篇MCP A2A双协议联动搭了三Agent工作流 A2A协议还在快速迭代但核心设计已经稳定。你现在掌握的知识足够开始搭建自己的多Agent系统了。 2026年是Agent协议之年。就像2010年学HTTP一样——现在入场刚刚好。