1. A2A协议的核心思想智能体间的沟通桥梁想象一下你有一个团队团队里每个人都有自己擅长的技能。A2A协议就是为这样的智能体团队设计的。它解决的核心问题包括智能体发现团队里有那么多专家我怎么知道谁能帮我解决这个问题就像一个服务注册中心智能体会在这里登记自己的能力任务委托如果一个任务太复杂我一个人搞不定能不能把它拆分成小任务分给不同的专家去做A2A允许智能体把任务派发给其他智能体状态管理任务分出去了我怎么知道它们进行到哪一步了有没有遇到困难A2A提供了跟踪任务进度的机制数据交换智能体之间如何安全、高效地传递数据和信息它定义了智能体之间数据传输的格式和方式价值体现A2A的核心价值在于实现智能体的分布式协作和能力复用。通过A2A我们可以构建一个由多个专业智能体组成的网络共同解决更复杂、更庞大的问题就像一个高效的虚拟团队。2. A2A与MCP协议的区别你可能还听说过另一个协议叫MCPModel Context Protocol。那么A2A和MCP有什么不同呢MCPModel Context Protocol更关注单个智能体如何利用工具和资源来完成自己的任务。它定义的是智能体内部如何与外部世界比如API、数据库、文件等进行交互。可以把MCP想象成智能体内部的工具箱管理协议。A2AAgent2Agent Protocol关注的是智能体之间如何协作和委托任务。它定义的是不同智能体之间如何进行对话和分工。A2A更像是智能体之间的沟通协议或团队协作协议。互补关系这两种协议并不是相互独立的它们是互补的。一个A2A智能体在收到任务后可以使用MCP协议去调用各种工具来完成自己的那部分工作然后再通过A2A协议将结果返回给发起者或者委托给其他智能体。所以你可以理解为A2A是智能体对外的协作标准MCP是智能体对内的工具使用标准。3. A2A的核心技术特征A2A协议为了实现上述功能通常会包含以下技术特征JSON-RPC通信这是一种基于JSON格式的远程过程调用Remote Procedure Call机制。简单来说就是智能体之间通过发送JSON格式的消息来调用对方的功能并接收返回的结果。它定义了请求和响应的统一格式。Agent Card发现每个智能体都会有一个类似名片的东西叫做Agent Card。这个卡片上会声明这个智能体的名称、版本、它能做什么即它提供的技能。通过这些Agent Card其他智能体可以发现并了解网络中有哪些智能体以及它们的能力。这就像一个服务注册与发现机制。流式处理对于长时间运行的任务A2A支持流式处理。这意味着智能体可以在任务执行过程中实时地发送状态更新和中间结果而不是等到任务全部完成才一次性返回。这对于需要实时反馈或进度监控的复杂任务非常有用。企业级安全A2A也考虑到了安全问题通常会包含完整的认证验证智能体身份、授权控制智能体能做什么和推送通知机制确保智能体之间的通信是安全可靠的。4. 环境搭建与准备在开始编写代码之前我们需要确保开发环境已经准备就绪。4.1 软件准备Python 3.12请确保你的电脑上安装了Python 3.12或更高版本推荐3.13版本。你可以在命令行输入python --version或python3 --version来检查。python_a2a 库这是Google A2A协议的Python实现库我们将用它来创建智能体和网络。4.2 安装 python_a2a 库打开你的命令行或终端Terminal输入以下命令来安装 python_a2apip install python_a2a如果安装成功你不会看到任何错误信息。5. 构建你的A2A智能体数学专家团队在这个部分我们将创建三个简单的A2A智能体它们各自负责一个数学计算任务计算一个数的正弦Sine、余弦Cosine和正切Tangent。这就像我们团队里的三位数学专家5.1 创建正弦计算智能体Sine Agent首先我们来创建 agent1.py。这个智能体将负责计算给定数字的正弦值。在你的项目文件夹中创建一个名为 agent1.py 的文件并粘贴以下代码from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState import math import re agent( nameSine Agent, # 代理名称 descriptionProvides the sine of a number, # 代理功能描述 version1.0.0 # 版本号 ) class SineAgent(A2AServer): 计算给定数字的正弦值 skill( nameGet Sine, # 技能名称 descriptionGet the sine of a number, # 技能描述 tags[sine, sin] # 关键词标签 ) def get_sine(self, number: float) - str: return fThe sine of {number} is {math.sin(number)} def handle_task(self, task): # 从任务消息中提取文本 input_message task.message[content][text] # 正则匹配数字 match re.search(r([-]?[0-9]*\.?[0-9]), input_message) if not match: task.status TaskStatus(stateTaskState.FAILED, error_message输入中未找到数字。) return task try: number float(match.group(1)) print(f正弦代理收到数字 {number}) sine_output self.get_sine(number) task.artifacts [{ parts: [{type: text, text: sine_output}] }] task.status TaskStatus(stateTaskState.COMPLETED) except ValueError: task.status TaskStatus(stateTaskState.FAILED, error_message数字格式错误。) except Exception as e: task.status TaskStatus(stateTaskState.FAILED, error_messagef发生错误: {str(e)}) return task if __name__ __main__: agent_instance SineAgent() print(Sine Agent starting on port 4737...) run_server(agent_instance, port4737)代码解释agent装饰器这是用来声明一个Python类是一个A2A智能体的。你需要提供 name智能体名称、description智能体功能描述和 version版本号。skill装饰器这是用来声明智能体提供的一个具体能力的。name 是技能的名称description 是技能的描述tags 是关键词列表有助于其他智能体或路由器发现这个技能。get_sine(self, number)方法这是一个普通的Python方法实现了计算正弦的逻辑。handle_task(self, task)方法这是所有A2A智能体必须实现的核心方法。当其他智能体或客户端向这个智能体发送任务时handle_task 会被调用。它接收一个 task 对象这个对象包含了任务的详细信息比如 task.message[content][text] 可以获取到请求的文本内容。我们使用正则表达式 re.search 从输入文本中提取数字。然后调用 self.get_sine(number) 执行实际的计算。最后将计算结果包装在 task.artifacts 中artifacts 可以理解为任务的产出或结果并将 task.status 设置为 TaskState.COMPLETED 表示任务完成。run_server(agent, port)这个函数会启动一个HTTP服务器让我们的A2A智能体可以监听来自其他智能体的请求。每个智能体都需要运行在不同的端口上。如何运行它保存 agent1.py 文件。打开你的命令行/终端。导航到你保存 agent1.py 的目录。运行命令python agent1.py你会看到类似 Sine Agent starting on port 4737... 的输出。这意味着你的正弦智能体已经在运行了它正在监听 http://localhost:4737。提示这个进程会一直运行直到你手动关闭它通常是按 CtrlC。在接下来的步骤中你需要让所有智能体同时运行。5.2 创建余弦计算智能体Cosine Agent接下来我们创建 agent2.py。它和 agent1.py 非常相似只是计算的是余弦值并且监听在不同的端口。在你的项目文件夹中创建一个名为 agent2.py 的文件并粘贴以下代码from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState import math import re agent( nameCosine Agent, # 代理名称 descriptionProvides the cosine of a number, # 代理功能描述 version1.0.0 # 版本号 ) class CosineAgent(A2AServer): 计算给定数字的余弦值 skill( nameGet Cos, # 技能名称 descriptionGet the cosine of a number, # 技能描述 tags[cos, cosine] # 关键词标签 ) def get_cosine(self, number: float) - str: return fThe cosine of {number} is {math.cos(number)} def handle_task(self, task): # 从任务消息中提取文本 input_message task.message[content][text] # 正则匹配数字 match re.search(r([-]?[0-9]*\.?[0-9]), input_message) if not match: task.status TaskStatus(stateTaskState.FAILED, error_message输入中未找到数字。) return task try: number float(match.group(1)) print(f余弦代理收到数字 {number}) cosine_output self.get_cosine(number) task.artifacts [{ parts: [{type: text, text: cosine_output}] }] task.status TaskStatus(stateTaskState.COMPLETED) except ValueError: task.status TaskStatus(stateTaskState.FAILED, error_message数字格式错误。) except Exception as e: task.status TaskStatus(stateTaskState.FAILED, error_messagef发生错误: {str(e)}) return task if __name__ __main__: agent_instance CosineAgent() print(Cosine Agent starting on port 4738...) run_server(agent_instance, port4738)如何运行它保存 agent2.py 文件。打开一个新的命令行/终端窗口不要关闭之前运行 agent1.py 的窗口。导航到你保存 agent2.py 的目录。运行命令python agent2.py你会看到类似 Cosine Agent starting on port 4738... 的输出。5.3 创建正切计算智能体Tangent Agent最后我们创建 agent3.py。它负责计算正切值并监听在第三个端口。在你的项目文件夹中创建一个名为 agent3.py 的文件并粘贴以下代码from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState import math import re agent( nameTangent Agent, # 代理名称 descriptionProvides the tangent of a number, # 代理功能描述 version1.0.0 # 版本号 ) class TangentAgent(A2AServer): 计算给定数字的正切值 skill( nameGet Tangent, # 技能名称 descriptionGet the tangent of a number, # 技能描述 tags[tangent, tan] # 关键词标签 ) def get_tangent(self, number: float) - str: return fThe tangent of {number} is {math.tan(number)} def handle_task(self, task): # 从任务消息中提取文本 input_message task.message[content][text] # 正则匹配数字 match re.search(r([-]?[0-9]*\.?[0-9]), input_message) if not match: task.status TaskStatus(stateTaskState.FAILED, error_message输入中未找到数字。) return task try: number float(match.group(1)) print(f正切代理收到数字 {number}) tangent_output self.get_tangent(number) task.artifacts [{ parts: [{type: text, text: tangent_output}] }] task.status TaskStatus(stateTaskState.COMPLETED) except ValueError: task.status TaskStatus(stateTaskState.FAILED, error_message数字格式错误。) except Exception as e: task.status TaskStatus(stateTaskState.FAILED, error_messagef发生错误: {str(e)}) return task if __name__ __main__: agent_instance TangentAgent() print(Tangent Agent starting on port 4739...) run_server(agent_instance, port4739)如何运行它保存 agent3.py 文件。打开第三个命令行/终端窗口保持前两个窗口运行。导航到你保存 agent3.py 的目录。运行命令python agent3.py你会看到类似 Tangent Agent starting on port 4739... 的输出。恭喜现在你已经成功启动了三个独立的A2A智能体它们分别运行在 4737、4738 和 4739 端口上随时准备接收任务。请确保这三个终端窗口都保持打开状态因为接下来的 main.py 将会与它们通信。6. 构建智能体网络与路由器现在我们有了三个独立的数学智能体但是我们怎么知道一个查询比如计算正弦值应该发送给哪个智能体呢这就是 智能体网络AgentNetwork 和 智能体路由器AIAgentRouter 的作用。智能体网络负责管理我们所有的智能体而智能体路由器则负责根据用户的查询智能地将任务路由到最合适的智能体。6.1 创建主调度器main.py创建一个名为 main.py 的文件并粘贴以下代码from python_a2a import AgentNetwork, A2AClient, AIAgentRouter, TaskStatus, TaskState # --- 第1部分初始化代理网络 --- # 创建代理网络作为所有代理的注册中心 network AgentNetwork(nameMath Assistant Network) # 添加三个数学代理指定代理名和监听地址 network.add(Sine, http://localhost:4737) network.add(Cosine, http://localhost:4738) network.add(Tangent, http://localhost:4739) print(Agent Network initialized with Sine, Cosine, and Tangent agents.) # --- 第2部分创建路由器简化演示--- # 简单关键词路由函数根据关键词选择代理 def simple_keyword_router(query: str, available_agents: dict) - tuple[str | None, float]: query_lower query.lower() if sine in query_lower or sin in query_lower: return Sine, 0.95 elif cosine in query_lower or cos in query_lower: return Cosine, 0.95 elif tangent in query_lower or tan in query_lower: return Tangent, 0.95 return None, 0.0 # 直接用关键词路由绕过需要LLM客户端的AIAgentRouter def route_and_ask(query: str, agent_network: AgentNetwork): agent_name, confidence simple_keyword_router(query, agent_network.agents) if agent_name: print(f\nRouting query: {query} to {agent_name} with {confidence:.2f} confidence (via keyword matching).) agent_client agent_network.get_agent(agent_name) if agent_client: try: response_task agent_client.ask(query) if response_task.status.state TaskState.COMPLETED: response_text for artifact in response_task.artifacts: for part in artifact.get(parts, []): if part.get(type) text: response_text part.get(text, ) \n print(fAgent Response: {response_text.strip()}) elif response_task.status.state TaskState.FAILED: print(fAgent Task Failed: {response_task.status.error_message}) else: print(fAgent Task Status: {response_task.status.state}) except Exception as e: print(fError communicating with agent {agent_name}: {e}) else: print(fError: Agent {agent_name} not found in the network.) else: print(f\nNo suitable agent found for query: {query} (via keyword matching).) # --- 第3部分测试查询 --- # 测试不同查询观察路由结果 queries [ What is the sine of 0.7854?, Calculate cos for 1.5708, Tan of 0.78545, What is the value of sin(3.14159)?, Compute cosine of 0, Find the tangent of 45 degrees (which is pi/4 or 0.7854), What is 2 2? # 无匹配代理 ] for q in queries: route_and_ask(q, network)代码解释AgentNetwork这是我们的智能体总览或注册表。我们通过 network.add(AgentName, http://localhost:port) 将之前创建的智能体注册到这个网络中。这样main.py 就知道了每个智能体的名称和它们的网络地址。simple_keyword_router(query, available_agents)函数为了简化教程我们没有使用真正的LLM。这个函数模拟了一个非常简单的路由器它检查查询字符串中是否包含 sine, cos, tan 等关键词。如果找到匹配的关键词它就返回对应的智能体名称和模拟的置信度。在真正的 AIAgentRouter 中llm_client 会帮助路由器理解更复杂的自然语言查询并根据智能体的 Agent Card (特别是 tags 和 description) 决定最佳路由。route_and_ask(query, agent_network)函数这个函数首先调用 simple_keyword_router 来决定哪个智能体应该处理当前的 query。然后它使用 agent_network.get_agent(agent_name) 从网络中获取对应智能体的客户端对象。最后通过 agent_client.ask(query) 向指定的智能体发送任务。它还会打印出智能体的响应并检查任务状态 (TaskState.COMPLETED 或 TaskState.FAILED)。如何运行它保存 main.py 文件。确保 agent1.py、agent2.py 和 agent3.py 都在它们各自的终端窗口中运行着。如果没有运行main.py 将无法连接到它们。打开第四个命令行/终端窗口。导航到你保存 main.py 的目录。运行命令python main.py预期结果你会在 main.py 的终端窗口中看到路由决策和来自各个智能体的响应。同时在运行 agent1.py、agent2.py 或 agent3.py 的终端窗口中你会看到它们收到请求时打印的 print 信息。例如Agent Network initialized with Sine, Cosine, and Tangent agents. Routing query: What is the sine of 0.7854? to Sine with 0.95 confidence (via keyword matching). Agent Response: The sine of 0.7854 is 0.7070775558117769 Routing query: Calculate cos for 1.5708 to Cosine with 0.95 confidence (via keyword matching). Agent Response: The cosine of 1.5708 is 0.0007963267948966192 Routing query: Tan of 0.78545 to Tangent with 0.95 confidence (via keyword matching). Agent Response: The tangent of 0.78545 is 1.0001592654388886 ... (and so on for other queries) No suitable agent found for query: What is 2 2? (via keyword matching).如果你在某个智能体的终端窗口中看到类似 正弦代理收到数字 0.7854 的输出那就表示请求确实被路由到了正确的智能体并且该智能体成功处理了请求。7. A2A系统设计与价值拓展通过前面的实战相信你已经对A2A协议有了初步的认识。现在让我们回到理论层面更深入地探讨如何设计一个健壮、高效且有价值的A2A系统。7.1 系统整体架构智能体层 (Agent Layer)这是我们刚刚创建的 SineAgent、CosineAgent 和 TangentAgent 所处的层级。它们是提供具体能力的最小单元。在实际系统中会有各种各样专业的智能体比如图像识别智能体、数据分析智能体等。协议层 (Protocol Layer)这是A2A协议本身发挥作用的地方。A2A Gateway 是智能体通信的入口和出口服务发现 机制通过 Agent Card 实现让智能体能找到彼此通信管理 则确保消息的正确传递。编排层 (Orchestration Layer)这是整个系统的大脑。智能体路由器 (AIAgentRouter)我们 main.py 中的 simple_keyword_router 就是一个简化的路由器。在复杂系统中它会利用LLM来理解用户意图并根据Agent Card和实时的负载情况将任务分配给最合适的智能体或智能体组合。任务调度器 (Task Scheduler)负责任务的生命周期管理比如任务的优先级、并发执行、失败重试等。状态管理器 (State Manager)跟踪复杂任务在不同智能体之间流转时的执行状态。应用层 (Application Layer)这是用户直接交互的部分。它可以是一个聊天机器人、一个自动化平台或者一个企业应用通过调用编排层来利用底层智能体的能力。7.2 智能体分类与交互模式智能体分类体系一个大中型A2A系统智能体需要有清晰的分类就像一个组织严密的团队。智能体网络 (Agent Network) ├── 领域专家智能体 (Domain Expert Agents) │ ├── 数学计算智能体 (Math Agents) # 我们的Sine, Cosine, Tangent就是这个 │ │ ├── 三角函数智能体 (Sine, Cosine, Tangent) │ │ ├── 统计分析智能体 │ │ └── 微积分智能体 │ ├── 数据处理智能体 (Data Processing Agents) │ ├── 文本分析智能体 (NLP Agents) │ └── 图像处理智能体 (Vision Agents) ├── 协调智能体 (Coordination Agents) │ ├── 主控制器智能体 (Master Controller) │ ├── 工作流管理智能体 (Workflow Manager) │ └── 结果聚合智能体 (Result Aggregator) └── 基础设施服务智能体 (Infrastructure Agents) ├── 认证鉴权智能体 (Auth Agent) ├── 日志审计智能体 (Audit Agent) └── 监控告警智能体 (Monitor Agent)领域专家智能体负责特定领域的任务比如我们之前的数学计算。协调智能体负责管理和编排其他智能体比如主控制器智能体可以分解复杂任务工作流管理智能体可以协调多个步骤结果聚合智能体则负责汇总所有子任务的结果。基础设施服务智能体提供整个系统公共的基础服务如身份认证、日志记录、监控等。智能体交互模式智能体之间可以以多种方式协作点对点模式 (P2P)最简单直接一个智能体直接调用另一个智能体来完成单一任务。就像我们的 main.py 直接调用 SineAgent。适用场景简单、单一的查询任务。编排模式 (Orchestration)当任务需要多步骤才能完成时由一个协调智能体如主控制器将任务分解并按顺序或并行地委托给多个智能体执行最后再聚合结果。用户请求 → 主控制器 → 任务分解 → 多智能体并行/串行执行 → 结果聚合适用场景复杂的多步骤任务如分析一份报告并生成摘要。协作模式 (Collaboration)更复杂的交互智能体之间可能需要多轮对话、相互依赖、共享数据才能完成任务。这需要更高级的状态管理和通信机制。智能体A ↔ 智能体B ↔ 智能体C (相互依赖、数据共享)适用场景需要多轮交互、信息逐步完善的复杂任务如帮我计划一次旅行并根据我的偏好调整行程。7.3 任务执行流程与数据流任务生命周期管理一个任务从被接收到最终完成会经历一系列状态任务接收 → 任务解析 → 智能体选择 → 任务分解 → 并行/串行执行 → 状态监控 → 结果聚合 → 响应返回在任何一步都可能出现错误这就需要 错误处理 和 重试机制。A2A的流式处理在这里尤为重要可以实时报告进度和问题。路由决策机制智能体路由器如何选择智能体是核心。除了我们示例中的简单关键词匹配更高级的策略包括规则路由基于明确的规则如如果查询包含翻译则路由到翻译智能体。能力匹配根据Agent Card中声明的技能和标签进行精准匹配。负载均衡考虑智能体的当前工作负载将任务分配给空闲或负载较低的智能体。智能推理利用LLM来理解用户意图进行更复杂的、语义级别的路由甚至可以进行任务的动态分解和编排。数据流和通信机制A2A通过JSON-RPC通信通常支持以下数据交换模式同步交换即时请求即时响应。适用于快速计算任务。异步交换提交任务后立即返回一个任务ID客户端可以通过轮询或接收推送通知来获取结果。适用于长时间运行的任务。流式交换智能体在任务执行过程中持续发送中间结果或状态更新。通常适用于需要实时反馈的场景如代码生成、数据处理进度等。7.4 如何更好地设计A2A系统并凸显其价值一个精心设计的A2A系统能够极大地提升效率、灵活性和可扩展性。清晰的智能体职责划分设计原则每个智能体应该专注于一个明确的领域或完成一类任务遵循单一职责原则。避免一个智能体承担过多功能。价值体现这样可以提高智能体的复用性便于独立开发、测试和部署降低系统复杂度。当某个功能需要更新时只需修改对应的智能体不会影响其他部分。细粒度的Agent Card定义设计原则Agent Card不仅要包含智能体的基本信息还要详细描述其提供的技能的输入、输出、参数、预期用途和限制。价值体现更丰富的Agent Card能让路由器更准确地发现和匹配智能体减少误路由和失败率。它也为未来的自动化编排和任务分解提供了丰富的信息。智能路由与动态编排设计原则利用强大的LLM作为路由器的大脑使其不仅能理解用户查询还能根据实时情况如智能体负载、可用性动态选择最佳智能体或智能体组合。对于复杂任务允许LLM进行动态分解和编排。价值体现这是A2A系统实现智能的核心。它能让系统像一个真正的智能总管根据需求灵活调配资源处理高度个性化和非结构化的任务大幅提升用户体验和系统智能化水平。健壮的任务管理与状态跟踪设计原则引入专门的任务调度器和状态管理器确保任务能够被正确执行、监控进度、处理失败并进行重试。采用流式通信提供实时反馈。价值体现确保系统的可靠性和稳定性即使在复杂的多智能体协作场景下也能有效管理任务生命周期提高任务成功率和用户满意度。开放与可扩展性设计原则A2A协议本身就支持新的智能体随时加入网络。设计时应考虑如何轻松集成第三方智能体或服务。价值体现A2A系统可以像积木一样通过不断添加新的智能体来扩展其功能和能力适应未来不断变化的需求形成一个庞大而灵活的智能体生态系统。安全与治理设计原则内置身份认证、授权和审计机制确保智能体之间通信的安全。建立智能体行为的监管和治理框架。价值体现特别是在企业级应用中安全性是基石。一个安全的A2A系统能保护数据隐私防止未经授权的访问和恶意行为确保系统合规运行。总结其价值A2A协议不仅仅是智能体之间的通信方式它提供了一个构建分布式、可协作、高智能系统的框架。通过A2A我们可以将复杂的AI能力分解成模块化的智能体实现能力复用不同应用可以复用同一个专业智能体。高可扩展性通过添加新智能体轻松扩展系统功能。灵活性与韧性智能体可以独立部署、升级一个智能体失败不影响整个系统。智能编排利用LLM实现任务的智能分解和路由处理更复杂、更真实的现实世界问题。