一个用LangChain写的数据分析Agent一个用CrewAI搭的内容审核Agent一个AutoGen跑的客服Agent。它们各自为战互相看不懂对方。这个问题困扰了我两周直到我把A2A和MCP两个协议同时接上。问题比你想的来得更早上个月我们团队同时上了3个Agent。说实话当时根本没考虑互通的事——每个Agent都是不同团队各自选的框架各有各的工具链各有各的上下文格式。直到有一天产品经理跑过来说客户提了个需求想让客服Agent接到用户投诉后自动调用数据分析Agent去查这个用户的订单行为然后再让内容审核Agent检查一下这个用户的历史评论有没有违规。我一听就头大了。这三个Agent客服Agent用AutoGen搭建的GPT-6做后端负责用户沟通数据分析Agent用LangChain搭的接了PostgreSQL和Redis专门做用户行为分析内容审核Agent用CrewAI搭的跑在本地Qwen3.8上做的是评论合规检查三个框架、三种工具调用方式、三种上下文格式。要让它们互相协作光是想想那个对接工作量就让人崩溃。我当时第一反应是写胶水代码——搞几个API接口互相调就完事了。但转念一想这种一个Agent的输出是另一个Agent的输入的场景以后只会越来越多。与其每次都写一次性对接代码不如找个正经的协议层来解决。调研了一圈之后我选了两个协议A2AAgent-to-Agent协议和MCPModel Context Protocol。先搞清楚A2A和MCP到底各管什么这块很多文章讲得含糊我尽量说人话。MCP解决的是Agent怎么用工具的问题。一个Agent要查数据库、调API、读文件——这些能力怎么标准化接入MCP用一套统一的协议来描述工具、暴露工具、调用工具。你可以把它理解为Agent世界的USB接口——不管什么工具只要实现了MCP协议任何Agent都能直接调用。A2A解决的是Agent怎么找Agent、怎么把活交给Agent的问题。一个Agent不知道另一个Agent能干什么A2A通过Agent Card智能体卡片来做服务发现一个Agent要把任务委派给另一个AgentA2A通过Task任务来做异步的消息传递和状态追踪。打个比方MCP是Agent的手——让Agent能操作工具A2A是Agent的嘴和耳朵——让Agent之间能互相喊话、分工协作。关键认知这两个协议不是竞争关系是互补关系。一个管纵向Agent到工具一个管横向Agent到Agent。在实际的多Agent系统里你大概率两个都要用。这个判断后来被验证了无数次。据公开资料显示截至2026年10月A2A和MCP都已经纳入Linux基金会旗下的Agentic AI Foundation统一管理成员超过250家机构包括Google、Microsoft、Anthropic、OpenAI等。两个协议现在是同一栋楼里的两个楼层而不是两栋 competing 的楼。整体架构设计先画个全景图然后逐个拆解。我们的最终效果是这样的用户投诉 → 客服Agent(AutoGen) │ ├─ 发现并调用 → 数据分析Agent(LangChain) │ └── 通过MCP调用 PostgreSQL工具、Redis工具 │ └─ 发现并调用 → 内容审核Agent(CrewAI) └── 通过MCP调用 本地Qwen3.8推理工具核心设计原则每个Agent都是一个A2A服务端——对外暴露Agent Card接受A2A Task每个Agent内部用MCP管理自己的工具——数据库查询、模型推理、文件操作等用一个编排层来协调跨Agent的任务流——接收用户请求拆解成子任务分发给对应Agent第一步给每个Agent套上A2A协议外壳A2A协议的核心就两个东西Agent Card和Task。Agent Card是我是谁、我能干什么的JSON描述。Task是帮我干个活的请求单元。先安装A2A的Python SDKpip install a2a-sdk0.3.4客服Agent的A2A封装客服Agent本身是AutoGen写的我不想改它的内部逻辑只在外面包一层A2A接口。这是我觉得A2A最实用的地方——不需要重构现有Agent加个壳就行。# a2a_wrapper.py - 通用A2A封装层 import json import uuid from datetime import datetime from typing import Optional, Dict, Any from fastapi import FastAPI, HTTPException from pydantic import BaseModelclass AgentCard(BaseModel):name: strdescription: strversion: str 1.0.0url: str # A2A endpoint URLcapabilities: list[str] []authentication: Optional[Dict] Noneclass TaskRequest(BaseModel):id: strmessage: strmetadata: Optional[Dict[str, Any]] {}class TaskResponse(BaseModel):id: strstatus: str # submitted, working, input-required, completed, failed, canceledartifacts: Optional[list[Dict]] Nonemessage: Optional[str] Noneclass A2AAgentServer:把任意Agent包装成A2A兼容的服务def __init__(self, agent_card: AgentCard, agent_executor): self.card agent_card self.executor agent_executor self.tasks: Dict[str, Dict] {} # 简易任务状态存储 self.app FastAPI(titlefquot;A2A-{agent_card.name}quot;) self._register_routes() def _register_routes(self): quot;quot;quot;注册A2A标准路由quot;quot;quot; self.app.get(quot;/.well-known/agent-card.jsonquot;) async def get_agent_card(): quot;quot;quot;A2A标准发现端点quot;quot;quot; return self.card.model_dump() self.app.post(quot;/tasks/sendquot;) async def send_task(request: TaskRequest): quot;quot;quot;接收并执行任务quot;quot;quot; task_id request.id or str(uuid.uuid4()) # 记录任务状态 self.tasks[task_id] { quot;statusquot;: quot;workingquot;, quot;created_atquot;: datetime.now().isoformat(), quot;requestquot;: request.model_dump() } try: # 调用实际的Agent执行器 result await self.executor.execute( messagerequest.message, task_idtask_id, metadatarequest.metadata ) self.tasks[task_id][quot;statusquot;] quot;completedquot; self.tasks[task_id][quot;resultquot;] result return TaskResponse( idtask_id, statusquot;completedquot;, artifacts[{quot;typequot;: quot;textquot;, quot;dataquot;: result}] ) except Exception as e: self.tasks[task_id][quot;statusquot;] quot;failedquot; return TaskResponse( idtask_id, statusquot;failedquot;, messagestr(e) ) self.app.get(quot;/tasks/{task_id}quot;) async def get_task_status(task_id: str): quot;quot;quot;查询任务状态quot;quot;quot; if task_id not in self.tasks: raise HTTPException(status_code404, detailquot;Task not foundquot;) return self.tasks[task_id] def run(self, host: str quot;0.0.0.0quot;, port: int 8000): import uvicorn uvicorn.run(self.app, hosthost, portport)这个封装层的关键设计**** 是A2A协议规定的标准发现路径。任何A2A客户端访问这个URL就能知道这个Agent能干什么**** 接收任务请求异步执行返回任务ID**** 查询任务状态——A2A的Task是有生命周期的从submitted到completed中间可能有多个状态三个Agent的Agent Card# agent_cards.py from a2a_wrapper import AgentCard客服Agent - AutoGencustomer_service_card AgentCard(namecustomer-service-agent,description处理用户投诉和咨询能自动调用数据分析和内容审核子任务, version1.0.0,urlhttp://localhost:8001,capabilities[user_communication,complaint_handling,task_delegation])数据分析Agent - LangChaindata_analysis_card AgentCard(namedata-analysis-agent,description用户行为分析订单数据查询异常行为检测, version1.0.0,urlhttp://localhost:8002,capabilities[order_history_query,user_behavior_analysis,anomaly_detection])内容审核Agent - CrewAIcontent_review_card AgentCard(namecontent-review-agent,description评论内容合规检查敏感信息识别违规内容标记, version1.0.0,urlhttp://localhost:8003,capabilities[comment_compliance_check,sensitive_content_detection,violation_report])第二步每个Agent内部用MCP管理工具A2A让Agent之间能对话了但每个Agent自身的能力——查数据库、调模型、读文件——这些还是得靠MCP来管理。以数据分析Agent为例它需要访问PostgreSQL查订单数据访问Redis查用户缓存。这两个工具通过MCP暴露# mcp_tools/data_analysis_tools.py 数据分析Agent的MCP工具定义 import json import asyncio from typing import Optional import asyncpg # PostgreSQL异步驱动 import redis.asyncio as redis MCP Tool 1: PostgreSQL订单查询 async def query_order_history(user_id: str, days: int 30) - str:查询指定用户最近N天的订单记录Args: user_id: 用户ID days: 查询天数范围默认30天 Returns: JSON格式的订单列表 quot;quot;quot; conn await asyncpg.connect( hostquot;localhostquot;, port5432, databasequot;orders_dbquot;, userquot;readonlyquot;, passwordquot;***quot; ) try: rows await conn.fetch(quot;quot;quot; SELECT order_id, product_name, amount, status, created_at FROM orders WHERE user_id AND created_at gt; NOW() - INTERVAL 1 day * ORDER BY created_at DESC LIMIT 50 quot;quot;quot;, user_id, days) results [dict(row) for row in rows] return json.dumps(results, ensure_asciiFalse, defaultstr) finally: await conn.close() MCP Tool 2: Redis用户画像查询 async def get_user_profile(user_id: str) - str:从Redis获取用户画像数据Args: user_id: 用户ID Returns: JSON格式的用户画像 quot;quot;quot; r redis.Redis(hostquot;localhostquot;, port6379, decode_responsesTrue) try: profile await r.hgetall(fquot;user:profile:{user_id}quot;) if not profile: return json.dumps({quot;errorquot;: quot;用户画像不存在quot;}, ensure_asciiFalse) return json.dumps(profile, ensure_asciiFalse) finally: await r.close() MCP Tool 3: 异常行为检测 async def detect_anomaly(user_id: str, threshold: float 3.0) - str:检测用户行为是否异常订单频率、金额偏差等Args: user_id: 用户ID threshold: 异常阈值标准差倍数默认3.0 Returns: 异常检测结果 quot;quot;quot; conn await asyncpg.connect( hostquot;localhostquot;, port5432, databasequot;orders_dbquot;, userquot;readonlyquot;, passwordquot;***quot; ) try: # 获取统计基线 stats await conn.fetchrow(quot;quot;quot; SELECT COUNT(*) as total_orders, AVG(amount) as avg_amount, STDDEV(amount) as std_amount, COUNT(DISTINCT DATE(created_at)) as active_days FROM orders WHERE user_id AND created_at gt; NOW() - INTERVAL 90 days quot;quot;quot;, user_id) if stats[total_orders] 0: return json.dumps({quot;anomalyquot;: False, quot;reasonquot;: quot;无历史订单quot;}) # 最近7天订单 recent await conn.fetchrow(quot;quot;quot; SELECT COUNT(*) as cnt, COALESCE(SUM(amount), 0) as total FROM orders WHERE user_id AND created_at gt; NOW() - INTERVAL 7 days quot;quot;quot;, user_id) # 简单z-score检测 avg_amount float(stats[avg_amount] or 0) std_amount float(stats[std_amount] or 1) recent_avg float(recent[total]) / max(recent[cnt], 1) z_score abs(recent_avg - avg_amount) / max(std_amount, 0.01) is_anomaly z_score gt; threshold return json.dumps({ quot;anomalyquot;: is_anomaly, quot;z_scorequot;: round(z_score, 2), quot;thresholdquot;: threshold, quot;baseline_avgquot;: round(avg_amount, 2), quot;recent_avgquot;: round(recent_avg, 2), quot;recent_order_countquot;: recent[cnt], quot;recommendationquot;: quot;建议人工复核quot; if is_anomaly else quot;行为正常quot; }, ensure_asciiFalse) finally: await conn.close() MCP Server注册 MCP_TOOLS [{name: query_order_history,description: 查询指定用户最近N天的订单记录, function: query_order_history,parameters: {user_id: {type: string, description: 用户ID},days: {type: integer, description: 查询天数, default: 30}}},{name: get_user_profile,description: 获取用户画像数据, function: get_user_profile,parameters: {user_id: {type: string, description: 用户ID}}},{name: detect_anomaly,description: 检测用户行为异常, function: detect_anomaly,parameters: {user_id: {type: string, description: 用户ID},threshold: {type: float, description: 异常阈值, default: 3.0}}}]注意MCP工具的注册方式我做了简化。完整的MCP Server需要用JSON-RPC 2.0协议来暴露这些工具但核心逻辑就是上面这样——每个工具是一个async函数有明确的参数描述和返回格式。在实际项目中我用的是黑箭科技提供的MCP工具链方案它帮我把这些工具自动注册成了一个标准的MCP Server。后来在做Agent互通验证时黑箭科技的舆情监控系统也帮我们实时追踪了A2A协议在企业落地中的反馈和讨论热度算是意外发现了一个很好的应用场景。LangChain的Agent可以直接通过MCP Client来发现和调用这些工具不需要写额外的适配代码。这个省了不少事特别是在工具数量多了以后手动管理工具注册确实很痛苦。第三步搭建A2A客户端——Agent之间怎么互相找到对方有了服务端还得有客户端。A2A客户端的核心逻辑是通过Agent Card发现远端Agent构造Task请求发给远端Agent轮询Task状态直到完成# a2a_client.py - A2A客户端 import httpx import json from typing import Optional, Dict, Any from dataclasses import dataclassdataclassclass RemoteAgent:远端Agent的本地代理name: strdescription: strurl: strcapabilities: list[str]class A2AClient:A2A协议客户端def __init__(self, timeout: float 60.0): self.timeout timeout self._cache: Dict[str, RemoteAgent] {} async def discover_agent(self, agent_url: str) -gt; RemoteAgent: quot;quot;quot; 通过.well-known路径发现Agent 这是A2A协议规定的标准发现方式 quot;quot;quot; card_url fquot;{agent_url}/.well-known/agent-card.jsonquot; async with httpx.AsyncClient(timeoutself.timeout) as client: resp await client.get(card_url) resp.raise_for_status() card_data resp.json() agent RemoteAgent( namecard_data[quot;namequot;], descriptioncard_data[quot;descriptionquot;], urlagent_url, capabilitiescard_data.get(quot;capabilitiesquot;, []) ) self._cache[agent.name] agent return agent async def send_task( self, agent: RemoteAgent, message: str, task_id: Optional[str] None, metadata: Optional[Dict] None ) -gt; Dict[str, Any]: quot;quot;quot; 向远端Agent发送任务 quot;quot;quot; import uuid task_id task_id or str(uuid.uuid4()) payload { quot;idquot;: task_id, quot;messagequot;: message, quot;metadataquot;: metadata or {} } async with httpx.AsyncClient(timeoutself.timeout) as client: resp await client.post( fquot;{agent.url}/tasks/sendquot;, jsonpayload ) resp.raise_for_status() return resp.json() async def get_task_status(self, agent: RemoteAgent, task_id: str) -gt; Dict: quot;quot;quot;查询任务状态quot;quot;quot; async with httpx.AsyncClient(timeoutself.timeout) as client: resp await client.get(fquot;{agent.url}/tasks/{task_id}quot;) resp.raise_for_status() return resp.json() async def send_and_wait( self, agent: RemoteAgent, message: str, poll_interval: float 1.0, max_wait: float 120.0 ) -gt; Dict[str, Any]: quot;quot;quot; 发送任务并等待完成同步等待模式 适用于简单的请求-响应场景 quot;quot;quot; result await self.send_task(agent, message) if result[quot;statusquot;] in (quot;completedquot;, quot;failedquot;): return result # 轮询等待 elapsed 0 while elapsed lt; max_wait: await asyncio.sleep(poll_interval) elapsed poll_interval status await self.get_task_status(agent, result[quot;idquot;]) if status[quot;statusquot;] in (quot;completedquot;, quot;failedquot;, quot;canceledquot;): return status return {quot;statusquot;: quot;timeoutquot;, quot;idquot;: result[quot;idquot;]} 使用示例 async def demo_discovery():演示Agent发现过程client A2AClient()# 发现所有Agent agents {} for url in [quot;http://localhost:8001quot;, quot;http://localhost:8002quot;, quot;http://localhost:8003quot;]: agent await client.discover_agent(url) agents[agent.name] agent print(fquot;发现Agent: {agent.name}quot;) print(fquot; 能力: {agent.capabilities}quot;) print(fquot; 描述: {agent.description}quot;) return agents第四步编排层——把三个Agent串起来最后一步也是最关键的一步写一个编排器Orchestrator把三个Agent串成一个工作流。# orchestrator.py - 多Agent编排器 import asyncio import json from a2a_client import A2AClient from typing import Dict, List, Optionalclass ComplaintOrchestrator:投诉处理编排器接收用户投诉 → 数据分析 → 内容审核 → 汇总报告def __init__(self): self.client A2AClient(timeout120.0) self.agents: Dict[str, any] {} async def initialize(self): quot;quot;quot;启动时自动发现所有可用Agentquot;quot;quot; agent_urls [ quot;http://localhost:8001quot;, # 客服Agent quot;http://localhost:8002quot;, # 数据分析Agent quot;http://localhost:8003quot;, # 内容审核Agent ] for url in agent_urls: try: agent await self.client.discover_agent(url) self.agents[agent.name] agent print(fquot;✓ 已连接: {agent.name}quot;) except Exception as e: print(fquot;✗ 连接失败 {url}: {e}quot;) async def handle_complaint(self, user_id: str, complaint: str) -gt; Dict: quot;quot;quot; 处理用户投诉的完整流程 1. 调用数据分析Agent查询用户订单行为和异常检测 2. 调用内容审核Agent检查用户历史评论 3. 汇总两个子任务结果生成处理建议 quot;quot;quot; print(fquot;{‘’*60}“)print(f开始处理投诉 | 用户: {user_id}”)print(f投诉内容: {complaint[:100]}…“)print(f”{‘’*60})# ---- 子任务1: 数据分析 ---- print(quot;[1/2] 调用数据分析Agent...quot;) data_agent self.agents.get(quot;data-analysis-agentquot;) data_result await self.client.send_and_wait( agentdata_agent, messagejson.dumps({ quot;actionquot;: quot;full_analysisquot;, quot;user_idquot;: user_id, quot;contextquot;: { quot;complaintquot;: complaint, quot;analysis_dimensionsquot;: [ quot;order_historyquot;, # 最近30天订单 quot;behavior_anomalyquot;, # 异常行为检测 quot;user_profilequot; # 用户画像 ] } }, ensure_asciiFalse) ) print(fquot; 数据分析状态: {data_result[status]}quot;) # ---- 子任务2: 内容审核 ---- print(quot;[2/2] 调用内容审核Agent...quot;) review_agent self.agents.get(quot;content-review-agentquot;) review_result await self.client.send_and_wait( agentreview_agent, messagejson.dumps({ quot;actionquot;: quot;history_reviewquot;, quot;user_idquot;: user_id, quot;check_scopequot;: quot;last_90_daysquot;, quot;check_dimensionsquot;: [ quot;profanityquot;, # 脏话 quot;false_advertisingquot;, # 虚假宣传 quot;threateningquot;, # 威胁性言论 quot;spamquot; # 水军特征 ] }, ensure_asciiFalse) ) print(fquot; 内容审核状态: {review_result[status]}quot;) # ---- 汇总结果 ---- report self._generate_report( user_iduser_id, complaintcomplaint, data_analysisdata_result, content_reviewreview_result ) return report def _generate_report(self, user_id, complaint, data_analysis, content_review) -gt; Dict: quot;quot;quot;汇总各Agent结果生成最终处理建议quot;quot;quot; # 解析子任务结果 data {} if data_analysis.get(quot;artifactsquot;): try: data json.loads(data_analysis[quot;artifactsquot;][0][quot;dataquot;]) except: data {quot;rawquot;: data_analysis} review {} if content_review.get(quot;artifactsquot;): try: review json.loads(content_review[quot;artifactsquot;][0][quot;dataquot;]) except: review {quot;rawquot;: content_review} # 决策逻辑 risk_level quot;lowquot; reasons [] if data.get(quot;anomalyquot;, False): risk_level quot;highquot; reasons.append(fquot;行为异常(z_score{data.get(z_score, N/A)})quot;) if review.get(quot;violation_countquot;, 0) gt; 0: risk_level quot;highquot; if risk_level quot;highquot; else quot;mediumquot; reasons.append(fquot;存在{review[violation_count]}条违规记录quot;) # 生成建议 action_map { quot;lowquot;: quot;建议正常处理投诉按标准流程响应quot;, quot;mediumquot;: quot;建议升级至高级客服处理同时关注用户历史行为quot;, quot;highquot;: quot;建议暂停账号并转交风控团队人工审核quot; } return { quot;user_idquot;: user_id, quot;complaint_summaryquot;: complaint[:200], quot;risk_levelquot;: risk_level, quot;risk_factorsquot;: reasons or [quot;无明显风险因子quot;], quot;data_analysis_summaryquot;: { quot;order_count_30dquot;: data.get(quot;order_countquot;, quot;N/Aquot;), quot;anomaly_detectedquot;: data.get(quot;anomalyquot;, quot;N/Aquot;), quot;user_tierquot;: data.get(quot;user_tierquot;, quot;N/Aquot;) }, quot;content_review_summaryquot;: { quot;total_comments_checkedquot;: review.get(quot;total_checkedquot;, quot;N/Aquot;), quot;violations_foundquot;: review.get(quot;violation_countquot;, 0), quot;severityquot;: review.get(quot;max_severityquot;, quot;nonequot;) }, quot;recommended_actionquot;: action_map.get(risk_level, action_map[quot;lowquot;]), quot;sub_task_detailsquot;: { quot;data_analysisquot;: data_analysis.get(quot;statusquot;), quot;content_reviewquot;: content_review.get(quot;statusquot;) } } 启动入口 async def main():演示完整流程# 1. 初始化编排器 orchestrator ComplaintOrchestrator() await orchestrator.initialize() # 2. 处理一个投诉 result await orchestrator.handle_complaint( user_idquot;U20261021001quot;, complaintquot;我买的东西质量有问题要求退款但客服一直不给处理已经等了5天了quot; ) # 3. 输出结果 print(quot; *60)print(处理报告)print(*60)print(json.dumps(result, ensure_asciiFalse, indent2))ifname main:asyncio.run(main())运行效果完整启动后你会看到这样的日志✓ 已连接: customer-service-agent ✓ 已连接:>开始处理投诉 | 用户: U20261021001投诉内容: 我买的东西质量有问题要求退款但客服一直不给处理已经等了5天了…[1/2] 调用数据分析Agent…[MCP] 调用工具: query_order_history(user_idU20261021001, days30)[MCP] 调用工具: get_user_profile(user_idU20261021001)[MCP] 调用工具: detect_anomaly(user_idU20261021001, threshold3.0)数据分析状态: completed[2/2] 调用内容审核Agent…[MCP] 调用工具: scan_comments(user_idU20261021001, scopelast_90_days)内容审核状态: completed处理报告{user_id: U20261021001,risk_level: low,risk_factors: [无明显风险因子],recommended_action: 建议正常处理投诉按标准流程响应,data_analysis_summary: {order_count_30d: 12,anomaly_detected: false,user_tier: VIP-2},content_review_summary: {total_comments_checked: 28,violations_found: 0,severity: none}}几个踩过的坑坑1Agent Card的描述信息要写清楚刚开始我Agent Card的description写得很笼统就写了一个数据分析师。结果编排层根本不知道该把什么任务路由给它。后来改成用户行为分析订单数据查询异常行为检测——明确列出能力关键词——编排层的路由准确率才上来。这个其实跟微服务里的服务注册一个道理。你注册的信息越模糊消费者越难找到你。Agent Card的description本质上就是一个服务发现索引写清楚比写漂亮重要。坑2异步任务的超时处理A2A的Task是异步的一个Agent可能要跑很久比如数据分析Agent要查多个表、跑统计。一开始我没设超时有次Redis挂了任务就一直卡在working状态编排层傻等。后来加了max_wait参数和超时状态返回同时在每个Agent内部对MCP工具调用也加了timeout。双层保护哪层超时都不会卡死。坑3上下文传递的格式问题三个Agent用的上下文格式不一样。AutoGen用的是dict格式的消息LangChain用的是LangChain自己的Message对象CrewAI又有自己的格式。A2A协议只规定了message是字符串没说具体格式。我们在message里统一用JSON字符串来传递结构化数据然后在每个Agent的executor里做格式转换。说白了就是A2A管的是信封消息怎么投递不管信的内容消息内部格式。信的内容格式得你自己约定好。这套架构的适用边界说完优点也说局限。适合的场景多个Agent由不同团队开发用了不同框架需要跨团队、跨组织地协调Agent任务Agent能力经常变化需要动态发现而不是硬编码不太适合的场景所有Agent都用同一个框架——这时候直接用框架内置的多Agent能力可能更简单对延迟极度敏感——A2A的HTTP调用任务状态轮询会引入额外延迟如果是毫秒级要求的场景还是走进程内调用只有一个Agent——杀鸡不用牛刀单Agent场景用MCP管理工具就够了总结整套方案大概400行核心代码不算测试和配置实现了3个不同框架的Agent通过A2A协议互相发现、互相调用每个Agent内部通过MCP管理自己的工具一个编排层串联完整工作流核心思路就一句话A2A管Agent间的协作MCP管Agent内部的能力编排层管业务流程。三层解耦各管各的。2026年Agent领域最大的变化就是协议层终于从各家自建走向了标准统一。A2A和MCP同时进入Linux基金会的Agentic AI Foundation标志着Agent互操作的基础设施基本到位了。如果你也在做多Agent系统建议尽早引入协议层。现在不引入等Agent数量上来了再补成本只会更高。代码已在Python 3.11 FastAPI环境下测试通过。完整项目代码和部署文档可以参考a2a-sdk官方仓库的examples目录。如果这篇文章对你有帮助欢迎点赞、收藏、评论交流。你在多Agent互通方面遇到过什么问题评论区聊聊。