核心内容摘要
圣女司幼幽-造相Z-Turbo代码审查实战：Java八股文知识库应用
Agent2Agent 协议2025年4月9日Google 正式发布了Agent2Agent Protocol 以下简称 “A2A”。
该协议为不同类型的智能体之间搭建了一座高效沟通与协作的桥梁无论是独立Agent与独立Agent、独立Agent与企业Agent亦或是企业Agent与企业Agent都能借助该协议实现通信交互和协作。
算是比较新的技术有
1 四大特性1Secure collaboration (安全协作): A2A协议全程保障智能体间的通信与数据安全通过身份验证、数据加密和访问控制三重机制有效防范未授权访问及数据泄露风险。
2Task and state mgmt (任务与状态管理): 这表明协议能够高效管理任务的全生命周期及其状态流转。
例如从任务提交开始直到最终完成、失败或被取消协议都能全程跟踪并妥善处理。
3UX negotiation (用户体验协商):该功能支持智能体间协作协商通过动态调整服务策略来持续优化用户体验。
4Capability discovery (能力发现): A2A 协议实现了智能体之间的自动能力发现机制。
通过 AgentCard 这类数字名片智能体可以主动发布自身功能而其他智能体则能基于这些公开信息筛选最合适的协作伙伴。
2 Agent2Agent 核心概念Client Agent 和 Server Agent 交互的过程中会涉及到一些 EntityAgentCard、Task 、Artifact等下面做个介绍。
1.
1 AgentSkillAgentSkill描述代理的具体能力或功能模块例如处理特定任务的技能。
它包括技能名称、描述、示例、输入/输出模式等。
在A2A协议中技能是代理卡片AgentCard的组成部分用于细粒度服务发现。
支持扩展标签和示例便于代理间匹配调用。
代码示例from python_a2a import AgentSkill # 定义一个代理技能 ticket_skill AgentSkill( namebook_ticket, description预订火车票的技能, examples[预订从上海到北京的火车票], input_modes[text/plain], output_modes[text/plain] ) print(ticket_skill) print(ticket_skill.to_dict())运行结果AgentSkill(namebook_ticket, description预订火车票的技能, id321ec43d-043f-
f730b33356, tags[], examples[预订从上海到北京的火车票], input_modes[text/plain], output_modes[text/plain]) {id: 321ec43d-043f-
f730b33356, name: book_ticket, description: 预订火车票的技能, tags: [], examples: [预订从上海到北京的火车票], inputModes: [text/plain], outputModes: [text/plain]}
1.
2 AgentCardAgentCard是A2A协议中定义Server Agent元数据的核心组件作为代理发现和服务注册的关键载体。
该卡片详细记录了代理的基础信息名称、描述、URL、版本、功能特性技能列表、流式传输支持等以及输入输出模式等关键参数。
在A2A生态系统中AgentCard承担着服务入口点的角色一方面支持JSON序列化以实现高效网络传输另一方面为Client Agent提供决策依据使其能够基于不同Server Agent的能力描述包括认证机制等智能选择最适合的任务执行对象。
代码示例:from python_a2a import AgentCard, AgentSkill # 创建一个代理技能 ticket_skill AgentSkill( namebook_ticket, description预订火车票的技能, examples[预订从上海到北京的火车票], input_modes[text/plain], output_modes[text/plain] ) # 创建代理卡片 agent_card AgentCard( nameTicketAgent, description一个可以预订票务的代理, urlhttp://
127.
0.
1:5009, version
1.
0, skills[ticket_skill], capabilities{streaming: True} ) # 打印代理卡片的字典表示用于序列化 print(agent_card) print(agent_card.to_dict())输出示例AgentCard(nameTicketAgent, description一个可以预订票务的代理, urlhttp://
127.
0.
1:5009, version
1.
0, authenticationNone, capabilities{streaming: True}, default_input_modes[text/plain], default_output_modes[text/plain], skills[AgentSkill(namebook_ticket, description预订火车票的技能, id487e3ee
cbd-4567-b083-a9627a486d1c, tags[], examples[预订从上海到北京的火车票], input_modes[text/plain], output_modes[text/plain])], providerNone, documentation_urlNone) {name: TicketAgent, description: 一个可以预订票务的代理, url: http://
127.
0.
1:5009, version:
1.
0, capabilities: {streaming: True}, defaultInputModes: [text/plain], defaultOutputModes: [text/plain], skills: [{id: 487e3ee
cbd-4567-b083-a9627a486d1c, name: book_ticket, description: 预订火车票的技能, tags: [], examples: [预订从上海到北京的火车票], inputModes: [text/plain], outputModes: [text/plain]}]}
1.
3 TaskTask是由Client Agent创建并管理的实体具有明确的状态属性其状态更新由Server Agent负责维护。
每个Task都对应特定的目标或结果。
在执行过程中Client Agent和Server Agent通过Message进行通信Server Agent执行后产生的输出结果称为Artifact。
每个Task都包含唯一的sessionId标识。
多个Task可共享相同sessionId表示这些Task属于同一个Session便于统一管理和跟踪相关任务的执行流程。
代码示例:from python_a2a import Task, Message, MessageRole, TextContent # 创建任务 message Message(contentTextContent(text查询天气), roleMessageRole.USER) task Task(messagemessage.to_dict()) print(task)运行结果Task(id5e63778f-a
f07-a
b47614d0d1b, session_id7211f559-d
f
d2-c74976f86390, statusTaskStatus(stateTaskState.SUBMITTED: submitted, messageNone, timestamp
T23:20:
44.
, message{content: {text: 查询天气, type: ContentType.TEXT: text}, role: user, message_id: 8a3c
-b70d-cf1434ad1caf}, history[], artifacts[], metadata{})
1.
4 TaskStateTaskState 枚举类TaskState 是一个枚举类用于定义任务的各种可能状态包括SUBMITTED已提交COMPLETED已完成FAILED已失败作为任务生命周期管理的基础组件该枚举类确保了状态的一致性和代码的可读性。
TaskState 状态表格状态名称值中文描述SUBMITTEDsubmitted任务已提交等待处理。
WAITINGwaiting任务正在等待例如等待外部资源或输入。
INPUT_REQUIREDinput-required任务需要额外用户输入以继续执行。
COMPLETEDcompleted任务已成功完成结果可用。
CANCELEDcanceled任务被取消未完成执行。
FAILEDfailed任务执行失败可能包含错误信息。
UNKNOWNunknown未知状态通常用于处理无效或未识别的状态。
注意状态名称TaskState 枚举成员如 TaskState.SUBMITTED提供类型安全的代码引用值对应的字符串值如submitted用于序列化或系统间交互中文描述说明各状态在任务生命周期中的功能和使用场景便于开发者理解代码示例from python_a2a import TaskState # 只需相关导入 # 检查任务状态 if TaskState.COMPLETED completed: print(任务完成) state TaskState.SUBMITTED print(转换后的状态值, state.value) print(state)运行结果任务完成 转换后的状态值 submitted TaskState.SUBMITTED
1.
5 TaskStatusTaskStatus 是用于表示 A2A 任务当前状态的对象包含三个关键属性状态枚举TaskState附加消息时间戳该对象主要用于跟踪任务执行进度支持序列化和格式转换功能能够动态反映任务处理状态。
使用
注意事项TaskStatus 必须包含一个有效的 TaskState 作为其 state 字段TaskState 是 TaskStatus 的必要依赖项示例代码展示了 TaskState 和 TaskStatus 的典型结合
使用方法。
代码示例from python_a2a import TaskStatus, TaskState status_completed TaskStatus( stateTaskState.COMPLETED, message{info: 任务成功完成} ) status_failed TaskStatus( stateTaskState.FAILED, message{error: 无法处理请求} ) # 打印字典表示 print(完成状态, status_completed.to_dict()) print(失败状态, status_failed.to_dict())输出日志完成状态 {state: completed, timestamp:
T22:55:
3
883274, message: {info: 任务成功完成}} 失败状态 {state: failed, timestamp:
T22:55:
3
883274, message: {error: 无法处理请求}}
1.
6 A2AServerA2AServer作为A2A协议的核心实现类负责构建代理服务器。
它继承自BaseA2AServer主要功能包括任务处理handle_task、消息处理handle_message和路由配置setup_routes。
该类集成了任务存储管理、流式订阅支持并兼容Google A2A模式通过Flask路由支持、任务处理逻辑和错误处理机制确保代理间通信的可靠性。
任务处理说明当继承A2AServer时系统会自动处理任务请求。
通常情况下无需手动创建Task对象A2AServer会自动将传入请求解析为Task对象并传递给handle_task方法进行处理。
handle_task方法功能 该方法负责解析任务输入、执行查询处理、封装处理结果并返回最终的task任务对象。
代码示例from python_a2a import A2AServer, run_server, AgentCard, AgentSkill, TaskStatus, TaskState # 定义代理卡片 ticket_card AgentCard( nameTicketAgentServer, description票务代理, urlhttp://
127.
0.
1:5009/a2a, skills[AgentSkill(namebook_ticket, description预订票务)] ) # 自定义 A2AServer 子类 class TicketServer(A2AServer): def __init__(self): super().__init__(agent_cardticket_card) def handle_task(self, task): print(f任务状态{task.status.state}) return task # 启动服务器 if __name__ __main__: server TicketServer() print(f[{server.agent_card.name}] 创建服务成功) run_server(server, host
127.
0.
1, port5009, debugFalse)
1.
7 artifacts在 A2A 协议中artifacts 是 Task 对象的核心字段用于存储任务执行后生成的输出结果。
该字段是一个列表结构其中每个元素都是一个产物对象通常以字典形式存储并包含指向多个内容片段的 parts 键。
artifacts 作为任务结果的结构化容器支持存储文本内容、函数调用结果和错误信息等多种数据类型确保客户端能够准确解析并有效利用代理生成的输出内容。
task.artifacts [ { parts: [ {type: text, text: 处理结果}, {type: error, message: 错误描述}, {type: function_response, name: func_name, response: {...}} ] } ]常见类型text纯文本结果如用户查询的响应。
error错误信息包含错误描述。
function_response函数调用结果包含函数名和返回数据。
name 和 response 用于函数响应。
function_call发起的函数调用包含函数名和参数。
代码示例服务端from python_a2a import A2AServer, run_server, AgentCard, AgentSkill, TaskStatus, TaskState # 定义代理卡片 ticket_card AgentCard( nameTicketAgentServer, description票务代理, urlhttp://
127.
0.
1:5010, skills[AgentSkill(namebook_ticket, description预订票务)] ) # 自定义 A2AServer 子类 class TicketServer(A2AServer): def __init__(self): super().__init__(agent_cardticket_card) def handle_task(self, task): print(收到A2A任务的task:, task) #默认写法获取任务内容 query (task.message or {}).get(content, {}).get(text, ) if 上海 in query and 北京 in query: # 这里的结果可以来自于 MCP 模块这里我们直接模拟结果 train_result 上海到北京的火车票已经预订成功 G1001,10车1A else: train_result 请输入明确的出发地和目的地。
task.artifacts [{parts: [{type: text, text: train_result}]}] task.status TaskStatus(stateTaskState.COMPLETED) print(f[{self.agent_card.name} 日志] 任务处理完毕) print(f[{self.agent_card.name} 日志] 输出结果task: {task}) print(f[{self.agent_card.name} 日志] 输出结果task.artifacts: {task.artifacts}) return task # 启动服务器 if __name__ __main__: server TicketServer() print(f[{server.agent_card.name}] 启动成功服务地址: {server.agent_card.url}) run_server(server, host
127.
0.
1, port5010, debugTrue)客户端import asyncio from python_a2a import A2AClient async def main(): ticket_client A2AClient(http://
127.
0.
1:
#预订火车票 ticket_query 预订一张从北京到上海的火车票 print(f[主控客户端日志]预订票务 - {ticket_query}) ticket_result ticket_client.ask(ticket_query) print(f[主控客户端日志] 收到票务预订结果: {ticket_result}) if __name__ __main__: asyncio.run(main())先启动服务端然后启动客户端最终得到的结果如下[主控客户端日志]预订票务 - 预订一张从北京到上海的火车票 [主控客户端日志] 收到票务预订结果: 上海到北京的火车票已经预订成功 G1001,10车1A
1.
8 AgentNetworkAgentNetwork 是 A2A 协议中的核心网络管理类主要负责集中管理和发现 A2A 兼容代理。
其主要功能包括维护代理列表支持通过 URL 或客户端实例添加代理缓存代理元数据如 AgentCard
核心价值简化多代理协作流程提供完整的代理管理功能添加/获取/列表/发现/移除适用于构建分布式代理系统消除手动管理多个客户端的复杂性核心特性添加代理通过 add 方法支持 URL自动创建 A2AClient或现有客户端。
代理元数据自动缓存 AgentCard便于查询代理能力。
发现代理通过 discover_agents 从 URL 列表自动添加有效代理。
扩展性支持头信息headers和异常处理。
代码示例from python_a2a import AgentNetwork network AgentNetwork(nameMyNetwork) network.add(TicketAgent, http://
127.
0.
1:
print(fagent network--{network.agent_cards}) print(**
# 调用 client network.get_agent(TicketAgent) print(client.ask(预订一张从北京到上海的火车票))运行结果INFO:python_a2a.client.network:Added agent TicketAgent from URL: http://
127.
0.
1:5010 agent network--{TicketAgent: AgentCard(nameTicketAgentServer, description票务代理, urlhttp://
127.
0.
1:5010, version
1.
0, authenticationNone, capabilities{google_a2a_compatible: True, parts_array_format: True, pushNotifications: False, stateTransitionHistory: False, streaming: True}, default_input_modes[text/plain], default_output_modes[text/plain], skills[AgentSkill(namebook_ticket, description预订票务, id65d84e15-c
ffd-9a
caf46d88e9, tags[], examples[], input_modes[text/plain], output_modes[text/plain])], providerNone, documentation_urlNone)} ******************************************************************************** 火车票已经预订成功 G1001,10车1A
1.
9 AIAgentRouter定义AIAgentRouter 是一个定义在 router.py 文件中的智能路由类它通过分析查询意图和上下文信息自动选择最合适的代理进行处理。
作用多代理网络支持智能路由功能可根据语义分析、历史上下文及缓存机制自动将用户查询精准匹配至最适合的代理节点。
系统以代理名称及置信度作为响应格式实现代理选择的自动化处理。
核心特性LLM 驱动使用 LLM 客户端如 ChatOpenAI生成路由提示分析查询匹配代理描述和技能。
上下文支持包含对话历史max_history_tokens 限制令牌数。
缓存优化类似查询使用缓存减少 LLM 调用。
回退机制LLM 失败时使用关键词匹配回退路由。
系统提示自定义提示指导路由决策。
代码示例from python_a2a import AIAgentRouter, AgentNetwork from langchain_openai import ChatOpenAI from agent_learn.config import Config confConfig() # 创建网络 network AgentNetwork(nameMyNetwork) network.add(TicketAgent, http://
127.
0.
1:
# 创建模型 llm ChatOpenAI(base_urlconf.base_url, api_keyconf.api_key, modelconf.model_name, temperature
0.
# 创建路由器 router AIAgentRouter(llm_clientllm, agent_networknetwork) agent_name, confidence router.route_query(预订票) print(agent_name, confidence)输出结果INFO:python_a2a.client.network:Added agent TicketAgent from URL: http://
127.
0.
1:5010 TicketAgent
0.
MCP协议MCPModel Context Protocol模型上下文协议是Anthropic于2024年1月推出的开放协议旨在为大型语言模型LLM与外部数据源及工具之间建立安全、高效的双向连接通道。
Anthropic致力于将MCP协议打造成为AI领域的通用接口类似于互联网中的HTTP协议。
通过MCP协议各类工具和数据源能够实现无缝对接最终达成广泛互联的生态系统。
1 MCP 核心概念MCP协议有两个核心角色客户端与服务端。
MCP服务端 (Tool Provider)角色工具的提供者。
职责将一个或多个本地函数例如Python函数包装起来通过一个标准的MCP接口暴露出去。
它监听来自客户端的请求执行对应的函数并返回结果。
例子一个天气查询服务、一个计算服务、一个数据库访问服务。
MCP客户端 (Tool Consumer)角色工具的调用者或消费者。
职责连接到MCP服务端查询可用的工具列表自发现并根据需要调用这些工具。
例子大模型Agent、自动化脚本、任何需要远程执行功能的应用程序。
区别与联系简单来说MCP 是“给 AI 装工具”而 A2A 是“让 AI 找外援”。
特性MCP (Model Context Protocol)A2A (Agent-to-Agent)角色定义Tool (工具)被动等待调用。
Agent (代理)具备独立思考和任务管理能力。
Token 消耗所有的工具描述和中间过程都在主 AI 的上下文中。
主 AI 只负责中转任务子 Agent 独立消耗 Token。
复杂任务处理难以处理需要长时间异步运行或多步确认的任务。
天生支持异步任务和复杂的内部流程。
解耦程度高度依赖主模型的指令遵循能力。
完全解耦。
子 Agent 可以用不同的模型、不同的语言实现。
权限隔离主 AI 拥有所有工具的访问权存在安全风险。
权限被封装在子 Agent 内部主 AI 只能看到最终结果。
1 为什么不只用多个 MCP想象一下你要开发一个“公司全能助理”全 MCP 方案你给一个 LLM 挂载了 50 个 MCP 工具财务、考勤、订票、代码、数据库。
痛点LLM 的上下文会被海量的工具描述塞满Context Window 爆炸且工具越多LLM 越容易产生“幻觉”分不清该用哪一个。
A2A 方案主助理只挂载 5 个“垂直领域的 Agent”。
优势当你说“帮我报销并订张票”时主助理只需分发任务。
具体到你的代码TicketServer不仅仅是一个函数它是一个独立的服务节点。
它可以有自己的排队系统、任务状态TaskStatus监控甚至在订票失败时自动尝试另一种方案而不需要主 AI 一直盯着它看。
综上所述还是取决于具体的任务使用 MCP当你有一些简单的、原子化的功能如查天气、读文件、算数学、查数据库一行数据。
使用 A2A当你需要构建一个复杂的分布式系统其中每个模块都需要独立维护状态、拥有独立逻辑或者你希望通过“专家分工”来降低主模型的推理负担。