企业官网建设流程全解析

MCP管工具，A2A管协作：双协议联合实战

摘要：MCP让Agent调工具，A2A让Agent之间对话。这篇把两个协议拧在一起，搭一个"搜索+总结+翻译"三Agent工作流。

一、两个协议，各管一件事

前面三篇你一直在跟A2A打交道。你的Agent能收消息、能回消息、能互相委托任务。但有个场景一直没解决：Agent怎么调用外部工具？

比如你想让Agent搜索网页、查数据库、读写文件。A2A不管这些——它只负责Agent之间的通信。管Agent和工具之间连接的，是另一个协议：MCP（Model Context Protocol）。

两个协议的分工很明确：

• MCP

  （Anthropic出品）= Agent连工具的"USB-C接口"

• A2A

  （Google出品）= Agent之间对话的"微信协议" 两个协议互补。一个Agent可以通过MCP调用搜索工具获取数据，然后通过A2A把搜索结果发给另一个Agent做分析。这就是2026年Agent开发的完整形态。 今天带你搭一个完整的三Agent工作流。

二、我们要搭什么

业务场景

你给Agent发一个关键词，比如"A2A协议最新进展"。Agent自动执行三步：

1. 搜索Agent

   通过MCP调用搜索工具，拿到相关文章

2. 总结Agent

   接收搜索结果，提炼关键信息

3. 翻译Agent

   把总结翻成英文，返回最终结果

架构图

你 → 翻译Agent（A2A Server，端口10002） ↓ A2A 总结Agent（A2A Server，端口10003） ↓ A2A 搜索Agent（A2A Server，端口10004） ↓ MCP 搜索工具（MCP Server）

搜索Agent同时是A2A Server（接收总结Agent的请求）和MCP Client（调用搜索工具）。一个Agent同时使用两个协议。

三、项目结构

a2a-mcp-workflow/ ├── searcher/# 搜索Agent（A2A + 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

四、搭建搜索Agent（A2A + MCP双协议）

这是最复杂的一个Agent。它需要同时做两件事：

1. 作为A2A Server，接收总结Agent发来的搜索请求

2. 作为MCP Client，调用搜索工具获取结果

4.1 配置MCP工具

创建searcher/ [mcp_config.yaml](http://mcp_config.yaml)，声明要用的MCP工具：

mcpServers: web-search: command:"npx"args:["-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做demo：

mcpServers: file-tools: command:"npx"args:["-y","@modelcontextprotocol/server-filesystem","/tmp/search-results"]

4.2 搜索Agent的Task Manager

importjsonimporthttpx from google_a2a.common.server.task_managerimportInMemoryTaskManager from google_a2a.common.typesimport(Artifact, Message, SendTaskRequest, SendTaskResponse, Task, TaskState, TaskStatus,)OLLAMA_URL="http://localhost:11434/api/generate"OLLAMA_MODEL=" qwen2.5:7b"classSearcherTaskManager(InMemoryTaskManager):"""搜索Agent：接收关键词，通过MCP搜索，返回结果。""" asyncdefon_send_task(self, request: SendTaskRequest)->SendTaskResponse: awaitself.upsert_task(request.params)task_id=request.params.idkeyword=request.params.message.parts[0].text# 第一步：通过MCP调用搜索工具search_results=awaitself._mcp_search(keyword)# 第二步：用大模型整理搜索结果formatted=awaitself._format_results(keyword, search_results)task=awaitself._update_task(task_id=task_id,task_state=TaskState.COMPLETED,response_text=formatted,)returnSendTaskResponse(id=request.id,result=task)asyncdef_mcp_search(self, keyword: str)->str:""" 通过MCP协议调用搜索工具。 这里演示两种方式：真实MCP调用和模拟搜索。"""# 方式1：真实MCP调用（需要Brave API Key）try:importos api_key=os.environ. get("BRAVE_API_KEY")ifapi_key: asyncwith httpx.AsyncClient(timeout=15.0)as c: resp=await 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",},)data=resp.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}]\n""1. A2A协议是Google推出的Agent间通信标准\n""2. 目前已有50+厂商支持A2A协议\n""3. A2A协议已捐赠给Linux基金会\n""4. 最新版本 v0.2.5，v1.0开发中\n""5. A2A与MCP协议互补，形成完整的Agent生态\n")asyncdef_format_results(self, keyword: str, results: str)->str:"""用大模型整理搜索结果。""" try: asyncwith httpx.AsyncClient(timeout=60.0)as c: resp=await c.post(OLLAMA_URL,json={"model":OLLAMA_MODEL,"prompt":(f"根据以下搜索结果，整理一份关于"f"「{keyword}」的信息摘要，"f"包含5个关键要点：\n{results}"),"stream":False,},)result=resp.json()returnresult.get("response", results)except Exception:returnresults asyncdef_update_task(self, task_id: str, task_state: TaskState, response_text: str)->Task: task=self.tasks[task_id]parts=[{"type":"text","text":response_text}]task.status=TaskStatus(state=task_state,message=Message(role="agent",parts=parts),)iftask_state==TaskState.COMPLETED: task.artifacts=[Artifact(parts=parts)]returntask

4.3 搜索Agent的Server

importloggingimportclick from google_a2a.common.typesimport(AgentSkill, AgentCapabilities, AgentCard)from google_a2a.common.serverimportA2AServer from task_managerimportSearcherTaskManager logging.basicConfig(level=logging.INFO)@click.command()@click.option("--host",default="localhost")@click.option("--port",default=10004,type=int)defmain(host, port): skill=AgentSkill(id="search-skill",name="Web Searcher",description="Searches the web and returns organized results",tags=["search","web","mcp"],examples=["搜索A2A协议最新进展"],inputModes=["text"],outputModes=["text"],)agent_card=AgentCard(name="Searcher Agent",description=("Searches the web via MCP protocol and ""returns organized information."),url=f"http://{host}:{port}/",version="0.1.0",defaultInputModes=["text"],defaultOutputModes=["text"],capabilities=AgentCapabilities(streaming=False),skills=[skill],)task_manager=SearcherTaskManager()server=A2AServer(agent_card=agent_card,task_manager=task_manager,host=host,port=port,)logging.info(f"Searcher 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_URL="http://localhost:10004"OLLAMA_URL="http://localhost:11434/api/generate"OLLAMA_MODEL=" qwen2.5:7b"classSummarizerTaskManager(InMemoryTaskManager):"""总结Agent：调用搜索Agent获取信息，然后做总结。""" asyncdefon_send_task(self, request: SendTaskRequest)->SendTaskResponse: awaitself.upsert_task(request.params)task_id=request.params.idkeyword=request.params.message.parts[0].text# 第一步：通过A2A调用搜索Agentsearch_info=awaitself._call_searcher(keyword)# 第二步：用大模型做深度总结summary=awaitself._summarize(keyword, search_info)task=awaitself._update_task(task_id=task_id,task_state=TaskState.COMPLETED,response_text=summary,)returnSendTaskResponse(id=request.id,result=task)asyncdef_call_searcher(self, keyword: str)->str:"""通过A2A协议调用搜索Agent。""" try: asyncwith httpx.AsyncClient(timeout=60.0)as c:# 发现搜索Agentcard_resp=await c.get(f"{SEARCHER_URL}/.well-known/ agent.json")card=card_resp.json()# 发送A2A请求payload={"jsonrpc":"2.0","id":"search-request","method":"message/send","params":{"message":{"role":"user","parts":[{"type":"text","text":keyword,}],}},}resp=await c.post(card["url"],json=payload,headers={"Content-Type":"application/json"},)result=resp.json()# 解析响应if"result"inresult: task_data=result["result"]artifacts=task_data.get("artifacts",[])ifartifacts:returnartifacts[0]["parts"][0]["text"]msg=task_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(timeout=60.0)as c: resp=await c.post(OLLAMA_URL,json={"model":OLLAMA_MODEL,"prompt":(f"你是一个专业的研究分析师。请根据以下搜索结果，"f"写一份关于「{keyword}」的深度分析报告。"f"要求：3-5个核心观点，每个观点有具体数据支撑。\n\n"f"搜索结果：\n{info}"),"stream":False,},)result=resp.json()returnresult.get("response", info)except Exception:returninfo asyncdef_update_task(self, task_id: str, task_state: TaskState, response_text: str)->Task: task=self.tasks[task_id]parts=[{"type":"text","text":response_text}]task.status=TaskStatus(state=task_state,message=Message(role="agent",parts=parts),)iftask_state==TaskState.COMPLETED: task.artifacts=[Artifact(parts=parts)]returntask

Server部分和上篇类似，端口改为10003，不再赘述。

六、搭建翻译Agent（纯A2A，调用总结Agent）

翻译Agent的逻辑跟上篇一模一样——接收你的消息，翻译文本，调用另一个Agent。只不过这次它调用的是总结Agent而不是摘要Agent。

# translator/task_manager.py（核心部分） SUMMARIZER_URL="http://localhost:10003"classTranslatorTaskManager(InMemoryTaskManager): asyncdefon_send_task(self, request):# ...（同上篇，省略）...# 调用总结Agent（而不是摘要Agent）summary=awaitself._call_summarizer(input_text)# 翻译结果translation=awaitself._translate(summary)final_text=(f"【中文总结】\n{summary}\n\n"f"【英文翻译】\n{translation}")task=awaitself._update_task(task_id=task_id,task_state=TaskState.COMPLETED,response_text=final_text,)returnSendTaskResponse(id=request.id,result=task)asyncdef_call_summarizer(self, text: str)->str:"""通过A2A调用总结Agent。"""# 代码与上篇的_call_summarizer相同# 只是URL改为SUMMARIZER_URL = "http://localhost:10003"# ...

七、启动测试

7.1 三个Agent全部启动

开三个终端：

# 终端1：搜索Agent（端口10004）cdsearcher&&python server.py# 终端2：总结Agent（端口10003）cdsummarizer&&python server.py# 终端3：翻译Agent（端口10002）cdtranslator&&python 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:# 发现翻译Agentresolver=A2ACardResolver(httpx_client=httpx_client,base_url="http://localhost:10002",)agent_card=await resolver.get_agent_card()print(f"连接到: { agent_card.name}\n")# 创建客户端client=awaitcreate_client(agent=agent_card,client_config=ClientConfig(streaming=False),)# 发送关键词keyword="A2A协议最新进展"message=new_text_message(keyword,role=Role.ROLE_USER)request=SendMessageRequest(message=message)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接口resp=await 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. 发现Agentcard=await 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":[...]}}}resp=await c.post(card["url"],json=payload)

A2A调用是"发现→通信"两步走。先拿到Agent Card了解对方能力，再发JSON-RPC请求。

两者的本质区别：

九、生产环境的建议

这个demo用的是模拟搜索。在生产环境里，你可能需要：

1. 用真实的MCP Server。

   Brave Search、GitHub、Playwright、文件系统等都有现成的MCP Server。

2. 给Agent加认证。

   结合上一篇的OAuth 2.0认证方案。

3. 加错误处理和重试。

   网络请求可能失败，需要超时和重试机制。

4. 加任务队列。

   多个请求并发时，用消息队列（如Redis）管理任务。

5. 加监控和日志。

   记录每次Agent间调用的耗时和状态。

十、系列总结

四篇文章，从零到一：

1. 第1篇

   ：搭了一个Echo Agent，搞懂A2A的基本概念

2. 第2篇

   ：搭了翻译Agent和摘要Agent，实现双Agent协作

3. 第3篇

   ：给Agent加了API Key / JWT / OAuth 2.0三把锁

4. 第4篇（本篇）

   ：MCP + A2A双协议联动，搭了三Agent工作流 A2A协议还在快速迭代，但核心设计已经稳定。你现在掌握的知识足够开始搭建自己的多Agent系统了。 2026年是Agent协议之年。就像2010年学HTTP一样——现在入场，刚刚好。