MCP+A2A协作：工具调用与Agent通信的完整架构

摘要MCP负责工具调用A2A负责Agent通信两者协作构建完整的AI Agent架构。本文详解MCPA2A混合架构设计提供工具调用与Agent协作的统一框架代码。MCPA2A协作: 工具调用与Agent通信的完整架构上周有个朋友跟我吐槽说他做了个客服系统Agent能查订单能查物流但碰到退款问题就傻眼了因为退款流程在另一个专门的Agent上。他问我两个Agent之间怎么通信工具调用和Agent间通信能不能用同一套协议这个问题其实非常典型。很多人刚接触Agent开发的时候会把MCP和A2A搞混甚至觉得用一个就够了。我当时也是这么想的后来踩了不少坑才明白这俩解决的根本不是同一个问题。今天这篇就来聊聊MCP和A2A怎么协同工作以及怎么用它们搭建一个真正能跑的客服Agent系统。MCP和A2A各自管什么先说结论。MCP解决的是Agent和工具之间的问题A2A解决的是Agent和Agent之间的问题。MCPModel Context Protocol是Anthropic提出的协议核心是让大模型能标准化地调用外部工具。你写一个MCP Server暴露查数据库、调API这些能力Agent通过MCP Client去连接就能用这些工具。它管的是Agent怎么用工具。A2AAgent-to-Agent是Google提出的协议核心是让不同的Agent能互相发现、互相通信、互相协作。Agent A干不了的活可以转交给Agent B去干。它管的是Agent怎么找帮手。一个是人和工具的关系一个是人和同事的关系。完全不同维度的事。对比维度MCPA2A解决的问题Agent与工具之间的调用Agent与Agent之间的通信通信方向Agent - 工具单向调用Agent - Agent双向通信核心概念Tool、Resource、PromptAgent Card、Task、Message传输方式stdio、SSE、Streamable HTTPHTTP、JSON-RPC状态管理无状态每次调用独立有状态Task有生命周期典型场景查数据库、调API、读文件任务转交、协作完成、信息同步整体架构设计我们要搭建的系统是这样的。一个客服Agent作为入口接收用户的问题。它能通过MCP查订单数据库查物流系统。当碰到退款这种专业问题时它通过A2A把任务转交给退款Agent。退款Agent处理完之后结果通过A2A返回给客服Agent客服Agent再回复用户。用户 | v 客服Agent (入口) |--- MCP --- 订单数据库Server (查订单) |--- MCP --- 物流系统Server (查物流) |--- A2A --- 退款Agent (处理退款) |--- A2A --- 投诉Agent (处理投诉)客服Agent同时是MCP Client和A2A Client。它用MCP调用工具用A2A跟其他Agent通信。实战步骤第一步 搭建MCP Server订单查询先写一个简单的MCP Server提供订单查询能力。# mcp_order_server.py# 订单查询MCP Server基于FastMCP实现frommcp.server.fastmcpimportFastMCPimportjson# 创建MCP Server实例名称为order-servicemcpFastMCP(order-service)# 模拟订单数据库实际项目中替换为真实数据库连接ORDERS_DB{ORD-001:{product:无线鼠标,price:89,status:已发货,tracking:SF1234567},ORD-002:{product:机械键盘,price:299,status:待发货,tracking:None},ORD-003:{product:显示器支架,price:159,status:已签收,tracking:SF7654321},}mcp.tool()defquery_order(order_id:str)-str:根据订单号查询订单信息 Args: order_id: 订单编号格式为ORD-XXX Returns: 订单详情的JSON字符串 # 从模拟数据库中查找订单orderORDERS_DB.get(order_id)iforderisNone:# 订单不存在时返回友好提示returnjson.dumps({error:f订单{order_id}不存在},ensure_asciiFalse)# 返回订单完整信息returnjson.dumps({order_id:order_id,**order},ensure_asciiFalse)mcp.tool()deflist_orders()-str:列出所有订单 Returns: 所有订单的JSON列表字符串 # 遍历数据库构建订单列表orders[{order_id:oid,**info}foroid,infoinORDERS_DB.items()]returnjson.dumps(orders,ensure_asciiFalse)if__name____main__:# 启动MCP Server使用stdio传输方式mcp.run(transportstdio)这个Server暴露了两个工具query_order和list_orders。客服Agent可以通过MCP协议调用它们。第二步 搭建退款AgentA2A Server退款Agent需要实现A2A协议能接收其他Agent发来的任务。# refund_agent.py# 退款Agent实现A2A协议的服务端fromfastapiimportFastAPIfrompydanticimportBaseModelfromtypingimportDict,Anyimportuuidimportjson appFastAPI(titleRefund Agent)# A2A协议中的Agent Card描述Agent的能力信息AGENT_CARD{name:refund-agent,# Agent名称description:专门处理退款申请的Agent,# Agent描述version:1.0.0,# 版本号capabilities:{# 能力声明streaming:False,# 不支持流式输出pushNotifications:False,# 不支持推送通知},skills:[# 技能列表{id:refund_process,# 技能IDname:退款处理,# 技能名称description:处理用户的退款申请审核退款条件,# 技能描述}],}# 存储任务状态的字典key为task_idtasks_store:Dict[str,Dict[str,Any]]{}classTaskSend(BaseModel):# A2A协议中发送任务的请求体jsonrpc:str2.0# JSON-RPC版本method:str# 方法名如tasks/sendparams:Dict[str,Any]# 请求参数id:str# 请求ID用于匹配响应app.get(/.well-known/agent.json)asyncdefget_agent_card():返回Agent Card供其他Agent发现能力returnAGENT_CARDapp.post(/)asyncdefhandle_task(task_send:TaskSend):处理A2A任务请求 支持的方法: - tasks/send: 发送新任务 - tasks/get: 查询任务状态 methodtask_send.methodifmethodtasks/send:# 提取任务内容paramstask_send.params task_idparams.get(id,str(uuid.uuid4()))# 生成或复用task_idmessageparams.get(message,{})# 从消息中提取文本内容text_contentforpartinmessage.get(parts,[]):ifpart.get(type)text:text_contentpart.get(text,)# 模拟退款处理逻辑# 实际项目中这里会调用退款审核流程refund_resultprocess_refund(text_content)# 构建任务状态记录task_record{id:task_id,# 任务IDstatus:{state:completed},# 任务状态已完成artifacts:[{# 任务产出物parts:[{# 产出内容分片type:text,text:refund_result,# 退款处理结果文本}]}],}# 存入任务存储tasks_store[task_id]task_recordreturn{jsonrpc:2.0,result:task_record,id:task_send.id}elifmethodtasks/get:# 查询已有任务状态task_idtask_send.params.get(id)tasktasks_store.get(task_id)iftaskisNone:return{jsonrpc:2.0,error:{code:-32602,message:任务不存在},id:task_send.id}return{jsonrpc:2.0,result:task,id:task_send.id}defprocess_refund(content:str)-str:处理退款逻辑的模拟函数 Args: content: 用户的退款请求文本 Returns: 退款处理结果文本 # 这里模拟退款审核规则if质量问题incontent:return退款已批准质量问题全额退款预计3个工作日到账。退款单号: RF-str(uuid.uuid4())[:8]elif不喜欢incontent:return退款已提交审核非质量问题需扣除运费预计5个工作日处理。退款单号: RF-str(uuid.uuid4())[:8]else:return退款申请已受理客服将在24小时内联系您确认退款细节。退款单号: RF-str(uuid.uuid4())[:8]if__name____main__:importuvicorn# 启动退款Agent监听8001端口uvicorn.run(app,host0.0.0.0,port8001)第三步 搭建客服AgentMCP Client A2A Client客服Agent是核心它既要通过MCP调用工具又要通过A2A跟退款Agent通信。# customer_service_agent.py# 客服Agent同时作为MCP Client和A2A ClientimportasyncioimportjsonimporthttpxfrommcpimportClientSession,StdioServerParametersfrommcp.client.stdioimportstdio_clientclassCustomerServiceAgent:客服Agent整合MCP工具调用和A2A Agent通信def__init__(self):# MCP会话引用初始化时为Noneself.mcp_sessionNone# 退款Agent的A2A地址self.refund_agent_urlhttp://localhost:8001asyncdefconnect_mcp(self):连接MCP Server建立工具调用通道# 配置MCP Server的启动参数server_paramsStdioServerParameters(commandpython,# 启动命令args[mcp_order_server.py],# 启动参数指定Server脚本)# 通过stdio建立连接asyncwithstdio_client(server_params)as(read,write):asyncwithClientSession(read,write)assession:# 初始化MCP会话awaitsession.initialize()self.mcp_sessionsession# 列出可用工具确认连接成功toolsawaitsession.list_tools()print(f已连接MCP Server可用工具:{[t.namefortintools.tools]})# 保持会话等待用户输入awaitself.chat_loop()asyncdefquery_order_via_mcp(self,order_id:str)-str:通过MCP查询订单信息 Args: order_id: 订单编号 Returns: 订单信息字符串 # 调用MCP工具query_orderresultawaitself.mcp_session.call_tool(query_order,# 工具名称{order_id:order_id}# 工具参数)# 提取工具返回的文本内容returnresult.content[0].textasyncdeftransfer_to_refund_agent(self,user_message:str)-str:通过A2A协议将退款任务转交给退款Agent Args: user_message: 用户的原始消息 Returns: 退款Agent返回的处理结果 # 构建A2A任务请求体遵循JSON-RPC 2.0格式task_request{jsonrpc:2.0,# JSON-RPC版本method:tasks/send,# A2A方法发送任务params:{id:ftask-{asyncio.get_event_loop().time()},# 生成任务IDmessage:{# 任务消息体role:user,# 消息角色parts:[{# 消息分片type:text,text:user_message,# 用户原始消息文本}]}},id:req-001# 请求ID}# 通过HTTP发送A2A请求到退款Agentasyncwithhttpx.AsyncClient()asclient:responseawaitclient.post(self.refund_agent_url/,# 退款Agent的A2A端点jsontask_request,# 请求体timeout30.0# 超时时间30秒)resultresponse.json()# 检查是否有错误iferrorinresult:returnf退款Agent处理失败:{result[error][message]}# 从返回结果中提取退款处理文本artifactsresult.get(result,{}).get(artifacts,[])forartifactinartifacts:forpartinartifact.get(parts,[]):ifpart.get(type)text:returnpart[text]return退款Agent未返回有效结果asyncdefchat_loop(self):客服对话主循环处理用户输入print(\n 客服Agent已就绪 )print(输入格式说明:)print( 查订单: 查询 ORD-001)print( 退款: 退款 ORD-001 质量问题)print( 退出: quit)print(*30)whileTrue:# 获取用户输入user_inputawaitasyncio.get_event_loop().run_in_executor(None,input,\n用户 )ifuser_input.strip().lower()quit:print(客服Agent退出)break# 简单的意图判断实际项目用LLM做意图识别ifuser_input.startswith(查询):# 提取订单号格式为查询 ORD-001order_iduser_input.replace(查询,).strip()# 通过MCP查询订单resultawaitself.query_order_via_mcp(order_id)print(f客服 订单查询结果:{result})elifuser_input.startswith(退款):# 退款请求通过A2A转交给退款Agentprint(客服 检测到退款需求正在转接退款专员...)resultawaitself.transfer_to_refund_agent(user_input)print(f客服 退款专员回复:{result})else:print(客服 我可以帮您查订单或处理退款请说明您的需求。)asyncdefmain():# 创建客服Agent实例agentCustomerServiceAgent()# 连接MCP并启动对话循环awaitagent.connect_mcp()if__name____main__:# 启动客服Agentasyncio.run(main())效果验证运行这个系统分三步。第一步启动退款Agent。python refund_agent.py第二步验证退款Agent的Agent Card是否可访问。curlhttp://localhost:8001/.well-known/agent.json第三步启动客服Agent。python customer_service_agent.py然后你可以在终端里输入查询 ORD-001客服Agent会通过MCP查到订单信息。输入退款 ORD-001 质量问题客服Agent会通过A2A把任务转给退款Agent拿到退款处理结果后回复给你。独家踩坑: MCP和A2A共用HTTP服务器的路由冲突这个坑我踩了整整两天。一开始我觉得既然MCP和A2A都用HTTP那我把它们放在同一个FastAPI应用里不就行了省得开两个端口。于是我把MCP Server的HTTP模式和A2A Server写在同一个FastAPI app里。结果MCP的SSE连接和A2A的任务请求互相干扰路由全部乱套。问题出在哪呢MCP的Streamable HTTP传输模式会在根路径/“上处理SSE和JSON-RPC请求而A2A也在根路径”/上处理JSON-RPC请求。两个协议都在同一个路径上监听请求进来之后FastAPI不知道该交给谁处理。我试过给A2A加一个/api前缀但MCP Client连接的时候默认就是连根路径改前缀导致MCP Client连不上。最终解决方案是这样的。MCP和A2A必须跑在独立的端口上。MCP Server用stdio或独立端口的HTTPA2A Server用另一个端口。两者在进程层面隔离互不干扰。如果你非要放在一个进程里可以用FastAPI的子应用挂载。# 错误写法: MCP和A2A路由冲突# app FastAPI()# app.mount(/, mcp_http_app) # MCP挂在根路径# app.post(/)(handle_a2a_task) # A2A也在根路径冲突!# 正确写法: 用子应用隔离main_appFastAPI()# MCP挂载到/mcp子路径下main_app.mount(/mcp,mcp_http_app)# A2A挂载到/a2a子路径下main_app.mount(/a2a,a2a_app)# Agent Card挂在根路径的固定位置main_app.get(/.well-known/agent.json)asyncdefcard():returnAGENT_CARD但说实话最省心的方案还是分端口部署。生产环境里用Docker Compose把每个Agent放在独立容器里端口完全隔离根本不用担心路由冲突。常见问题与避坑问题1: A2A任务超时怎么处理退款Agent处理可能很慢客服Agent不能一直等。解决方案是A2A协议本身支持异步任务。你可以在tasks/send的时候返回一个pending状态的任务然后用tasks/get轮询或者用webhook回调。我上面的代码为了简化是同步等待的生产环境一定要改成异步模式。问题2: MCP的stdio模式和A2A的HTTP模式能混用吗可以而且推荐这么做。MCP用stdio的好处是进程隔离干净工具Server挂了不影响主进程。A2A用HTTP是因为Agent之间是跨进程通信必须走网络。我在代码里就是这么做的MCP走stdioA2A走HTTP。问题3: 多个Agent怎么发现彼此A2A协议规定每个Agent在/.well-known/agent.json路径上暴露自己的Agent Card。你可以写一个Agent注册中心维护所有Agent的地址和Card信息。客服Agent启动时先去注册中心拉取可用Agent列表需要转交任务时根据Card里的skills字段找到合适的Agent。小结这篇我们搭了一个完整的客服Agent系统。客服Agent通过MCP查订单查物流通过A2A把退款任务转给专业Agent。核心认知是MCP管工具调用A2A管Agent通信两者不是替代关系而是互补关系。踩的坑是HTTP路由冲突根本原因是两个协议都默认用根路径。解决方案是分端口部署或者用子应用隔离。下一篇我们来聊ACP协议另一种Agent通信方案看看它和A2A有什么不同。相关推荐A2A实战搭建两个Agent的通信链路多Agent编排实战用MCPA2A构建AI团队MCP协议全景Host、Client、Server架构详解