摘要：MCP负责工具调用，A2A负责Agent通信，两者协作构建完整的AI Agent架构。本文详解MCP+A2A混合架构设计，提供工具调用与Agent协作的统一框架代码。

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

上周有个朋友跟我吐槽，说他做了个客服系统，Agent能查订单能查物流，但碰到退款问题就傻眼了，因为退款流程在另一个专门的Agent上。他问我，两个Agent之间怎么通信？工具调用和Agent间通信能不能用同一套协议？

这个问题其实非常典型。很多人刚接触Agent开发的时候，会把MCP和A2A搞混，甚至觉得用一个就够了。我当时也是这么想的，后来踩了不少坑才明白，这俩解决的根本不是同一个问题。今天这篇就来聊聊MCP和A2A怎么协同工作，以及怎么用它们搭建一个真正能跑的客服Agent系统。

MCP和A2A各自管什么

先说结论。MCP解决的是Agent和工具之间的问题，A2A解决的是Agent和Agent之间的问题。

MCP（Model Context Protocol）是Anthropic提出的协议，核心是让大模型能标准化地调用外部工具。你写一个MCP Server，暴露查数据库、调API这些能力，Agent通过MCP Client去连接，就能用这些工具。它管的是"Agent怎么用工具"。

A2A（Agent-to-Agent）是Google提出的协议，核心是让不同的Agent能互相发现、互相通信、互相协作。Agent A干不了的活，可以转交给Agent B去干。它管的是"Agent怎么找帮手"。

一个是人和工具的关系，一个是人和同事的关系。完全不同维度的事。

整体架构设计

我们要搭建的系统是这样的。一个客服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-servicemcp=FastMCP("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字符串 """# 从模拟数据库中查找订单order=ORDERS_DB.get(order_id)iforderisNone:# 订单不存在时返回友好提示returnjson.dumps({"error":f"订单{order_id}不存在"},ensure_ascii=False)# 返回订单完整信息returnjson.dumps({"order_id":order_id,**order},ensure_ascii=False)@mcp.tool()deflist_orders()->str:"""列出所有订单 Returns: 所有订单的JSON列表字符串 """# 遍历数据库，构建订单列表orders=[{"order_id":oid,**info}foroid,infoinORDERS_DB.items()]returnjson.dumps(orders,ensure_ascii=False)if__name__=="__main__":# 启动MCP Server，使用stdio传输方式mcp.run(transport="stdio")

这个Server暴露了两个工具，query_order和list_orders。客服Agent可以通过MCP协议调用它们。

第二步 搭建退款Agent（A2A Server）

退款Agent需要实现A2A协议，能接收其他Agent发来的任务。

# refund_agent.py# 退款Agent，实现A2A协议的服务端fromfastapiimportFastAPIfrompydanticimportBaseModelfromtypingimportDict,Anyimportuuidimportjson app=FastAPI(title="Refund 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",# 技能ID"name":"退款处理",# 技能名称"description":"处理用户的退款申请，审核退款条件",# 技能描述}],}# 存储任务状态的字典，key为task_idtasks_store:Dict[str,Dict[str,Any]]={}classTaskSend(BaseModel):# A2A协议中发送任务的请求体jsonrpc:str="2.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_CARD@app.post("/")asyncdefhandle_task(task_send:TaskSend):"""处理A2A任务请求 支持的方法: - tasks/send: 发送新任务 - tasks/get: 查询任务状态 """method=task_send.methodifmethod=="tasks/send":# 提取任务内容params=task_send.params task_id=params.get("id",str(uuid.uuid4()))# 生成或复用task_idmessage=params.get("message",{})# 从消息中提取文本内容text_content=""forpartinmessage.get("parts",[]):ifpart.get("type")=="text":text_content+=part.get("text","")# 模拟退款处理逻辑# 实际项目中这里会调用退款审核流程refund_result=process_refund(text_content)# 构建任务状态记录task_record={"id":task_id,# 任务ID"status":{"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}elifmethod=="tasks/get":# 查询已有任务状态task_id=task_send.params.get("id")task=tasks_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,host="0.0.0.0",port=8001)

第三步 搭建客服Agent（MCP 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_session=None# 退款Agent的A2A地址self.refund_agent_url="http://localhost:8001"asyncdefconnect_mcp(self):"""连接MCP Server，建立工具调用通道"""# 配置MCP Server的启动参数server_params=StdioServerParameters(command="python",# 启动命令args=["mcp_order_server.py"],# 启动参数，指定Server脚本)# 通过stdio建立连接asyncwithstdio_client(server_params)as(read,write):asyncwithClientSession(read,write)assession:# 初始化MCP会话awaitsession.initialize()self.mcp_session=session# 列出可用工具，确认连接成功tools=awaitsession.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_orderresult=awaitself.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":f"task-{asyncio.get_event_loop().time()}",# 生成任务ID"message":{# 任务消息体"role":"user",# 消息角色"parts":[{# 消息分片"type":"text","text":user_message,# 用户原始消息文本}]}},"id":"req-001"# 请求ID}# 通过HTTP发送A2A请求到退款Agentasyncwithhttpx.AsyncClient()asclient:response=awaitclient.post(self.refund_agent_url+"/",# 退款Agent的A2A端点json=task_request,# 请求体timeout=30.0# 超时时间30秒)result=response.json()# 检查是否有错误if"error"inresult:returnf"退款Agent处理失败:{result['error']['message']}"# 从返回结果中提取退款处理文本artifacts=result.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_input=awaitasyncio.get_event_loop().run_in_executor(None,input,"\n用户> ")ifuser_input.strip().lower()=="quit":print("客服Agent退出")break# 简单的意图判断，实际项目用LLM做意图识别ifuser_input.startswith("查询"):# 提取订单号，格式为"查询 ORD-001"order_id=user_input.replace("查询","").strip()# 通过MCP查询订单result=awaitself.query_order_via_mcp(order_id)print(f"客服> 订单查询结果:{result}")elifuser_input.startswith("退款"):# 退款请求，通过A2A转交给退款Agentprint("客服> 检测到退款需求，正在转接退款专员...")result=awaitself.transfer_to_refund_agent(user_input)print(f"客服> 退款专员回复:{result}")else:print("客服> 我可以帮您查订单或处理退款，请说明您的需求。")asyncdefmain():# 创建客服Agent实例agent=CustomerServiceAgent()# 连接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或独立端口的HTTP，A2A Server用另一个端口。两者在进程层面隔离，互不干扰。如果你非要放在一个进程里，可以用FastAPI的子应用挂载。

# 错误写法: MCP和A2A路由冲突# app = FastAPI()# app.mount("/", mcp_http_app) # MCP挂在根路径# app.post("/")(handle_a2a_task) # A2A也在根路径，冲突!# 正确写法: 用子应用隔离main_app=FastAPI()# 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走stdio，A2A走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编排实战：用MCP+A2A构建AI团队
• • MCP协议全景：Host、Client、Server架构详解