ADK A2A Basic 示例深度解析用本地子 Agent 与远程 A2A Agent 构建多智能体协作应用【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python本篇文章以 adk-python 仓库中contributing/samples/a2a/a2a_basic示例为核心完整拆解基于 Agent-to-AgentA2A协议的多智能体协作实现一个本地根 Agentroot_agent如何通过本地子 Agentroll_agent完成掷骰子再通过 HTTP 调用独立部署的远程 A2A 服务prime_agent完成素数判断并实现掷骰子 判素数的组合任务编排。读完本文你将掌握 ADK 中RemoteA2aAgent的连接方式、agent cardAgent 卡片的解析与部署要点、子 Agent 委托机制以及一套可直接复制运行的 A2A 跨服务多智能体实战方案。示例概览DicePrimeBot 的三层结构A2A Basic 示例演示了 ADKAgent Development Kit中的 Agent-to-Agent 架构展示多个 Agent 如何协作处理复杂任务。示例实现了一个既能掷骰子、又能判断数字是否为素数的智能体其整体由三个角色组成角色名称类型职责根 Agentroot_agent本地 LLM Agent主编排器根据用户请求把任务委托给专门的子 Agent掷骰子 Agentroll_agent本地子 Agent处理骰子投掷操作可配置骰子面数素数 Agentprime_agent远程 A2A Agent运行在独立 A2A 服务器上判断数字是否为素数其中roll_agent直接内嵌在主 Agent 进程中而prime_agent则通过 A2A 协议与一个运行在localhost:8001的独立服务通信这是理解整个示例的关键本地子 Agent 与远程 A2A Agent 是两种不同的集成方式。架构本地编排与跨服务通信如何串联示例的架构图来自 a2a_basic/README.md清晰地描述了数据流向┌─────────────────┐ ┌──────────────────┐ ┌────────────────────┐ │ Root Agent │───▶│ Roll Agent │ │ Remote Prime │ │ (Local) │ │ (Local) │ │ Agent │ │ │ │ │ │ (localhost:8001) │ │ │───▶│ │◀───│ │ └─────────────────┘ └──────────────────┘ └────────────────────┘从该图可以读出三条关键链路本地委托链路root_agent→roll_agent。根 Agent 通过sub_agents[roll_agent, prime_agent]声明子 AgentLLM 根据用户指令自主决定把任务转交给谁transfer 机制。远程调用链路root_agent→prime_agent远程服务器。prime_agent作为RemoteA2aAgent把请求封装成 A2A 协议消息经 HTTP 发送到http://localhost:8001/a2a/check_prime_agent上的远程 A2A 服务。组合编排链路对于掷骰子并判断是否为素数这类复合请求根 Agent 会先调用roll_agent获得结果再把结果作为参数传给prime_agent最后向用户汇总两段结论。本地子 Agentroll_agent 的实现与配置要点roll_agent在 contributing/samples/a2a/a2a_basic/agent.py 中定义核心代码如下import random from google.adk.agents.llm_agent import Agent from google.adk.agents.remote_a2a_agent import AGENT_CARD_WELL_KNOWN_PATH from google.adk.agents.remote_a2a_agent import RemoteA2aAgent from google.adk.tools.example_tool import ExampleTool from google.genai import types # --- Roll Die Sub-Agent --- def roll_die(sides: int) - int: Roll a die and return the rolled result. return random.randint(1, sides) roll_agent Agent( nameroll_agent, descriptionHandles rolling dice of different sizes., instruction You are responsible for rolling dice based on the users request. When asked to roll a die, you must call the roll_die tool with the number of sides as an integer. , tools[roll_die], generate_content_configtypes.GenerateContentConfig( safety_settings[ types.SafetySetting( # avoid false alarm about rolling dice. categorytypes.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, thresholdtypes.HarmBlockThreshold.OFF, ), ] ), )要点拆解函数工具roll_die(sides: int)是一个普通的 Python 函数直接作为tools[roll_die]传入 Agent。ADK 会自动把函数的签名参数名、类型、docstring转换为 LLM 可调用的 Function Calling 声明无需额外封装。明确的 instructioninstruction里明确指示必须把骰子面数作为整数调用 roll_die 工具这是保证 LLM 正确调用工具、不发生参数类型误用如传字符串的关键。安全设置generate_content_config.safety_settings把HARM_CATEGORY_DANGEROUS_CONTENT的拦截阈值设为OFF注释明确说明这是为了避免掷骰子这类与随机数/赌博语义接近的请求被安全模型误判。这是示例中一个非常实用的细节当业务场景与安全分类存在语义重叠时可以按需调整阈值。roll_die底层只是random.randint(1, sides)随机性由 Python 标准库提供无需额外依赖。远程 A2A Agentprime_agent 与 RemoteA2aAgent 源码解读prime_agent的定义同样位于 contributing/samples/a2a/a2a_basic/agent.pyfrom google.adk.agents.remote_a2a_agent import AGENT_CARD_WELL_KNOWN_PATH from google.adk.agents.remote_a2a_agent import RemoteA2aAgent prime_agent RemoteA2aAgent( nameprime_agent, descriptionAgent that handles checking if numbers are prime., agent_card( fhttp://localhost:8001/a2a/check_prime_agent{AGENT_CARD_WELL_KNOWN_PATH} ), )这里有两个值得深入的点agent_card 的三种指定方式从 remote_a2a_agent.py 的源码可以看出RemoteA2aAgent支持三种方式定位远程 Agent直接传入 AgentCard 对象agent_card为a2a.types.AgentCard实例传入 agent card JSON 的 URL 字符串以http://或https://开头如本示例传入 agent card JSON 的文件路径字符串本地文件路径。当agent_card是字符串时源码 L752-L755会被保存为_agent_card_source在运行时通过_resolve_agent_card_from_url或_resolve_agent_card_from_file解析源码 L935-L960。AGENT_CARD_WELL_KNOWN_PATH 的作用AGENT_CARD_WELL_KNOWN_PATH是 A2A 规范中 agent card 的标准发现路径。在 remote_a2a_agent.py 源码 L45-L49 中可以看到try: from a2a.utils.constants import AGENT_CARD_WELL_KNOWN_PATH except ImportError: # Fallback for older versions of a2a-sdk. AGENT_CARD_WELL_KNOWN_PATH /.well-known/agent.json也就是说prime_agent实际请求的 agent card 地址是http://localhost:8001/a2a/check_prime_agent/.well-known/agent.json这个路径正是远程 A2A 服务器暴露 Agent 描述信息能力、URL、输入输出模式等的约定位置。根 Agent 通过这张卡片才能发现远程 Agent 的存在及其能力这是 A2A 协议实现服务发现的核心机制。其他可选参数源码可见的扩展点从 RemoteA2aAgent.init签名 可以看出该类的构造参数还包括timeout默认 600 秒、a2a_client_factory自定义 A2A 客户端工厂、auth_scheme/auth_credential/credential_key对远程 Agent 调用的认证配置、full_history_when_stateless无状态 Agent 是否每次携带完整会话历史等。本示例使用默认配置即可但在生产环境对接需要鉴权的远程 Agent 时这些参数是重要的扩展入口。远程服务端实现check_prime_agent 的完整剖析远程 A2A 服务端位于 contributing/samples/a2a/a2a_basic/remote_a2a/check_prime_agent/包含三个文件agent.pyAgent 实现、agent.jsonAgent 卡片、__init__.py。素数判断算法与工具agent.py 中定义了一个异步函数工具check_primeasync def check_prime(nums: list[int]) - str: Check if a given list of numbers are prime. Args: nums: The list of numbers to check. Returns: A str indicating which number is prime. primes set() for number in nums: number int(number) if number 1: continue is_prime True for i in range(2, int(number**0.5) 1): if number % i 0: is_prime False break if is_prime: primes.add(number) return ( No prime numbers found. if not primes else f{, .join(str(num) for num in primes)} are prime numbers. )算法要点支持批量输入nums: list[int]一次性判断多个数字去重后返回素数集合试除法优化只需遍历到int(number ** 0.5)将时间复杂度控制在 O(√n)对示例场景足够高效边界处理number 1直接跳过1 及以下不是素数返回值语义化无素数时返回No prime numbers found.否则返回2, 3, 5 are prime numbers.这类可读文本。远程 Agent 的配置同一文件中远程服务端定义了自己的 Agentfrom google.adk import Agent root_agent Agent( namecheck_prime_agent, descriptioncheck prime agent that can check whether numbers are prime., instruction You check whether numbers are prime. When checking prime numbers, call the check_prime tool with a list of integers. Be sure to pass in a list of integers. You should never pass in a string. You should not rely on the previous history on prime results. , tools[ check_prime, ], generate_content_configtypes.GenerateContentConfig( safety_settings[ types.SafetySetting( # avoid false alarm about rolling dice. categorytypes.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, thresholdtypes.HarmBlockThreshold.OFF, ), ] ), )注意这里有两个值得学习的细节参数类型强约束instruction 反复强调必须以整数列表传入绝不能传字符串因为check_prime内部虽然做了int(number)强转但 LLM 若以字符串形式传参会导致批量逻辑出错。这与主 Agent 中roll_agent的 instruction 风格一致——通过 prompt 约束工具调用契约是 ADK 示例中的常见最佳实践。不依赖历史声明instruction 中明确不要依赖之前关于素数结果的对话历史这确保了每次判断都基于最新传入的参数重新计算避免模型凭记忆作答。agent card 文件agent.json远程服务的 Agent 卡片位于 contributing/samples/a2a/a2a_basic/remote_a2a/check_prime_agent/agent.json{ capabilities: {}, defaultInputModes: [ text/plain ], defaultOutputModes: [ application/json ], description: An agent specialized in checking whether numbers are prime. It can efficiently determine the primality of individual numbers or lists of numbers., name: check_prime_agent, skills: [ { id: prime_checking, name: Prime Number Checking, description: Check if numbers in a list are prime using efficient mathematical algorithms, tags: [ mathematical, computation, prime, numbers ] } ], url: http://localhost:8001/a2a/check_prime_agent, version: 1.0.0 }字段含义name/descriptionAgent 的标识与描述供 A2A 客户端即主 Agent 侧的RemoteA2aAgent发现和理解url远程 Agent 的 RPC 端点地址是整张卡片中最关键的字段下文部署一节详述skills声明 Agent 具备的能力此处为prime_checking带tags便于检索匹配defaultInputModes/defaultOutputModes声明默认输入为纯文本、默认输出为 JSONcapabilities当前为空对象表示未声明额外能力如流式、状态持久化等。根 Agent 编排委托逻辑与少样本工具根 Agent 负责把用户请求路由到正确的子 Agent并支持链式操作。其定义同样在 contributing/samples/a2a/a2a_basic/agent.py 中example_tool ExampleTool([ { input: { role: user, parts: [{text: Roll a 6-sided die.}], }, output: [ {role: model, parts: [{text: I rolled a 4 for you.}]} ], }, { input: { role: user, parts: [{text: Is 7 a prime number?}], }, output: [{ role: model, parts: [{text: Yes, 7 is a prime number.}], }], }, { input: { role: user, parts: [{text: Roll a 10-sided die and check if its prime.}], }, output: [ { role: model, parts: [{text: I rolled an 8 for you.}], }, { role: model, parts: [{text: 8 is not a prime number.}], }, ], }, ]) root_agent Agent( nameroot_agent, instruction You are a helpful assistant that can roll dice and check if numbers are prime. You delegate rolling dice tasks to the roll_agent and prime checking tasks to the prime_agent. Follow these steps: 1. If the user asks to roll a die, delegate to the roll_agent. 2. If the user asks to check primes, delegate to the prime_agent. 3. If the user asks to roll a die and then check if the result is prime, call roll_agent first, then pass the result to prime_agent. Always clarify the results before proceeding. , global_instruction( You are DicePrimeBot, ready to roll dice and check prime numbers. ), sub_agents[roll_agent, prime_agent], tools[example_tool], generate_content_configtypes.GenerateContentConfig( safety_settings[ types.SafetySetting( # avoid false alarm about rolling dice. categorytypes.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, thresholdtypes.HarmBlockThreshold.OFF, ), ] ), )编排机制拆解sub_agents声明sub_agents[roll_agent, prime_agent]同时注册了本地子 Agent 和远程 A2A Agent。对根 Agent 而言两者都是可转交transfer的下游节点LLM 根据用户意图自主选择。instruction中的显式路由规则三步规则覆盖了三种请求形态掷骰子 / 判素数 / 组合任务特别是第 3 步明确了组合任务的调用顺序——先roll_agent后prime_agent再把结果串联。global_instruction角色设定You are DicePrimeBot, ready to roll dice and check prime numbers.为根 Agent 注入统一人格与能力边界与instruction的任务流程分工明确。ExampleTool少样本引导ExampleTool是 ADK 内置工具实现位于 example_tool.py它并不执行真实操作而是在每次 LLM 请求时把示例追加为指令引导模型模仿输入-输出模式。其process_llm_request会基于当前用户输入从示例列表中挑选最匹配的通过example_util.build_example_si构建系统指令。示例中三个问答对分别对应掷骰子判素数组合操作让模型在零训练的情况下也能稳定复现正确的回复格式与调用路径。环境搭建与运行两条命令启动整套系统运行该示例需要两个独立进程。前提是当前环境已安装 ADK 及其 A2A 相关依赖。第 1 步启动远程 A2A 服务在第一个终端执行# Start the remote a2a server that serves the check prime agent on port 8001 adk api_server --a2a --port 8001 contributing/samples/a2a/a2a_basic/remote_a2a--a2a标志表示以 A2A 模式启动 API 服务器--port 8001指定远程服务监听端口目录参数指向remote_a2a该目录下的check_prime_agent即被发布的服务 Agent。启动后服务会在http://localhost:8001/a2a/check_prime_agent暴露 A2A RPC 端点并在/.well-known/agent.json提供 Agent 卡片。第 2 步运行主 AgentWeb 服务器在另一个终端执行# In a separate terminal, run the adk web server adk web contributing/samples/a2aadk web启动 ADK 的本地 Web 开发服务器默认端口 8000并加载contributing/samples/a2a目录下的示例两个进程分别占用8000本地 Web UI与8001远程 A2A 服务端口互不冲突。交互示例两个服务都就绪后即可通过 Web UI 与根 Agent 对话。README 给出了三组典型交互覆盖全部编排路径简单掷骰子走 roll_agentUser: Roll a 6-sided die Bot: I rolled a 4 for you.素数判断走 prime_agent 远程调用User: Is 7 a prime number? Bot: Yes, 7 is a prime number.组合操作先 roll_agent 再 prime_agentUser: Roll a 10-sided die and check if its prime Bot: I rolled an 8 for you. Bot: 8 is not a prime number.第三个例子最能体现多智能体协作价值根 Agent 自动完成掷骰子 → 取结果 → 传给远程素数服务 → 汇总回复的全链路对用户而言只是发出了一句自然语言请求。代码结构速览contributing/samples/a2a/a2a_basic/ ├── agent.py # 主 Agentroll_agent / prime_agent / root_agent / ExampleTool ├── __init__.py # 导入 agent 模块 └── remote_a2a/ └── check_prime_agent/ ├── agent.py # 远程素数服务的 Agent 与 check_prime 工具 ├── agent.json # A2A Agent 卡片含 url 端点声明 └── __init__.py从仓库结构看contributing/samples/a2a目录下还提供了同主题的进阶变体例如 a2a_authA2A 认证、a2a_human_in_loop人类介入、a2a_state_forwarding状态转发、a2a_root 等可作为本示例的延伸学习材料。部署到其他环境必须更新 agent card 的 url 字段当把远程 A2A Agent 部署到不同环境如 Cloud Run、不同主机/端口时必须同步更新 agent.json 中的url字段本地开发{ url: http://localhost:8001/a2a/check_prime_agent, ... }Cloud Run 示例{ url: https://your-service-abc123-uc.a.run.app/a2a/check_prime_agent, ... }自定义主机/端口示例{ url: https://your-domain.com:9000/a2a/check_prime_agent, ... }重要提示url字段必须指向远程 A2A Agent 实际部署并可访问的 RPC 端点。同时主 Agent 侧构造RemoteA2aAgent时传入的 agent card URL 也必须与该服务器实际运行位置保持一致。也就是说存在两处一致性要求agent.json中的url指向真实部署地址主 Agent 侧RemoteA2aAgent(agent_card...)能通过/.well-known/agent.json拉取到这张卡片。两条链路任一处失配都会导致根 Agent 无法发现或无法调用远程服务。故障排查指南连接问题确保本地 ADK Web 服务器运行在 8000 端口确保远程 A2A 服务器运行在 8001 端口检查是否有防火墙阻止 localhost 连接验证 agent.json 中的url字段与实际部署位置一致验证传给RemoteA2aAgent构造函数的 agent card URL 与正在运行的 A2A 服务器相匹配。Agent 无响应同时检查本地 ADK Web 服务器8000与远程 A2A 服务器8001两侧的日志检查 Agent 的 instruction 是否清晰、无歧义尤其是工具调用的参数类型约束再次核对 agent.json 中的 RPC URL 是否正确且可访问。从实现层面补充一个定位技巧RemoteA2aAgent启动时会先解析 agent card源码中的_resolve_agent_card路径见 remote_a2a_agent.py。若卡片拉取失败或字段非法请求根本不会进入远程 Agent 的执行阶段因此排查时可以用curl http://localhost:8001/a2a/check_prime_agent/.well-known/agent.json之类的命令先验证卡片端点本身是否可达、返回是否完整。扩展方向README 为本示例列出了五条扩展路径结合源码可进一步落地增加更多数学运算仿照check_prime添加因式分解、平方根等工具与子 Agent创建更多远程 Agent复用remote_a2a/check_prime_agent的目录结构agent.py agent.json新建服务并在根 Agent 中注册新的RemoteA2aAgent实现更复杂的委托逻辑扩展root_agent的instruction路由规则支持多级、条件式任务分发添加持久化状态管理远程 Agent 可在agent.json的capabilities中声明状态支持配合RemoteA2aAgent的会话状态管理机制使用集成外部 API 或数据库把外部调用封装为函数工具挂载到对应子 Agent。小结A2A Basic 示例以最小可运行的形式展示了 ADK 多智能体协作的三层能力本地子 Agent 集成roll_agent 函数工具、远程 A2A Agent 集成RemoteA2aAgent agent card 服务发现、根 Agent 智能编排instruction 路由 ExampleTool 少样本引导。其中agent.json的url字段是连接本地与远程世界的桥梁也是部署迁移时最需要留意的配置项。掌握了这个示例你就具备了在 ADK 中搭建跨服务多智能体系统的基础骨架。【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 ADK A2A Basic 示例深度解析：用本地子 Agent 与远程 A2A Agent 构建多智能体协作应用
ADK A2A Basic 示例深度解析用本地子 Agent 与远程 A2A Agent 构建多智能体协作应用【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python本篇文章以 adk-python 仓库中contributing/samples/a2a/a2a_basic示例为核心完整拆解基于 Agent-to-AgentA2A协议的多智能体协作实现一个本地根 Agentroot_agent如何通过本地子 Agentroll_agent完成掷骰子再通过 HTTP 调用独立部署的远程 A2A 服务prime_agent完成素数判断并实现掷骰子 判素数的组合任务编排。读完本文你将掌握 ADK 中RemoteA2aAgent的连接方式、agent cardAgent 卡片的解析与部署要点、子 Agent 委托机制以及一套可直接复制运行的 A2A 跨服务多智能体实战方案。示例概览DicePrimeBot 的三层结构A2A Basic 示例演示了 ADKAgent Development Kit中的 Agent-to-Agent 架构展示多个 Agent 如何协作处理复杂任务。示例实现了一个既能掷骰子、又能判断数字是否为素数的智能体其整体由三个角色组成角色名称类型职责根 Agentroot_agent本地 LLM Agent主编排器根据用户请求把任务委托给专门的子 Agent掷骰子 Agentroll_agent本地子 Agent处理骰子投掷操作可配置骰子面数素数 Agentprime_agent远程 A2A Agent运行在独立 A2A 服务器上判断数字是否为素数其中roll_agent直接内嵌在主 Agent 进程中而prime_agent则通过 A2A 协议与一个运行在localhost:8001的独立服务通信这是理解整个示例的关键本地子 Agent 与远程 A2A Agent 是两种不同的集成方式。架构本地编排与跨服务通信如何串联示例的架构图来自 a2a_basic/README.md清晰地描述了数据流向┌─────────────────┐ ┌──────────────────┐ ┌────────────────────┐ │ Root Agent │───▶│ Roll Agent │ │ Remote Prime │ │ (Local) │ │ (Local) │ │ Agent │ │ │ │ │ │ (localhost:8001) │ │ │───▶│ │◀───│ │ └─────────────────┘ └──────────────────┘ └────────────────────┘从该图可以读出三条关键链路本地委托链路root_agent→roll_agent。根 Agent 通过sub_agents[roll_agent, prime_agent]声明子 AgentLLM 根据用户指令自主决定把任务转交给谁transfer 机制。远程调用链路root_agent→prime_agent远程服务器。prime_agent作为RemoteA2aAgent把请求封装成 A2A 协议消息经 HTTP 发送到http://localhost:8001/a2a/check_prime_agent上的远程 A2A 服务。组合编排链路对于掷骰子并判断是否为素数这类复合请求根 Agent 会先调用roll_agent获得结果再把结果作为参数传给prime_agent最后向用户汇总两段结论。本地子 Agentroll_agent 的实现与配置要点roll_agent在 contributing/samples/a2a/a2a_basic/agent.py 中定义核心代码如下import random from google.adk.agents.llm_agent import Agent from google.adk.agents.remote_a2a_agent import AGENT_CARD_WELL_KNOWN_PATH from google.adk.agents.remote_a2a_agent import RemoteA2aAgent from google.adk.tools.example_tool import ExampleTool from google.genai import types # --- Roll Die Sub-Agent --- def roll_die(sides: int) - int: Roll a die and return the rolled result. return random.randint(1, sides) roll_agent Agent( nameroll_agent, descriptionHandles rolling dice of different sizes., instruction You are responsible for rolling dice based on the users request. When asked to roll a die, you must call the roll_die tool with the number of sides as an integer. , tools[roll_die], generate_content_configtypes.GenerateContentConfig( safety_settings[ types.SafetySetting( # avoid false alarm about rolling dice. categorytypes.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, thresholdtypes.HarmBlockThreshold.OFF, ), ] ), )要点拆解函数工具roll_die(sides: int)是一个普通的 Python 函数直接作为tools[roll_die]传入 Agent。ADK 会自动把函数的签名参数名、类型、docstring转换为 LLM 可调用的 Function Calling 声明无需额外封装。明确的 instructioninstruction里明确指示必须把骰子面数作为整数调用 roll_die 工具这是保证 LLM 正确调用工具、不发生参数类型误用如传字符串的关键。安全设置generate_content_config.safety_settings把HARM_CATEGORY_DANGEROUS_CONTENT的拦截阈值设为OFF注释明确说明这是为了避免掷骰子这类与随机数/赌博语义接近的请求被安全模型误判。这是示例中一个非常实用的细节当业务场景与安全分类存在语义重叠时可以按需调整阈值。roll_die底层只是random.randint(1, sides)随机性由 Python 标准库提供无需额外依赖。远程 A2A Agentprime_agent 与 RemoteA2aAgent 源码解读prime_agent的定义同样位于 contributing/samples/a2a/a2a_basic/agent.pyfrom google.adk.agents.remote_a2a_agent import AGENT_CARD_WELL_KNOWN_PATH from google.adk.agents.remote_a2a_agent import RemoteA2aAgent prime_agent RemoteA2aAgent( nameprime_agent, descriptionAgent that handles checking if numbers are prime., agent_card( fhttp://localhost:8001/a2a/check_prime_agent{AGENT_CARD_WELL_KNOWN_PATH} ), )这里有两个值得深入的点agent_card 的三种指定方式从 remote_a2a_agent.py 的源码可以看出RemoteA2aAgent支持三种方式定位远程 Agent直接传入 AgentCard 对象agent_card为a2a.types.AgentCard实例传入 agent card JSON 的 URL 字符串以http://或https://开头如本示例传入 agent card JSON 的文件路径字符串本地文件路径。当agent_card是字符串时源码 L752-L755会被保存为_agent_card_source在运行时通过_resolve_agent_card_from_url或_resolve_agent_card_from_file解析源码 L935-L960。AGENT_CARD_WELL_KNOWN_PATH 的作用AGENT_CARD_WELL_KNOWN_PATH是 A2A 规范中 agent card 的标准发现路径。在 remote_a2a_agent.py 源码 L45-L49 中可以看到try: from a2a.utils.constants import AGENT_CARD_WELL_KNOWN_PATH except ImportError: # Fallback for older versions of a2a-sdk. AGENT_CARD_WELL_KNOWN_PATH /.well-known/agent.json也就是说prime_agent实际请求的 agent card 地址是http://localhost:8001/a2a/check_prime_agent/.well-known/agent.json这个路径正是远程 A2A 服务器暴露 Agent 描述信息能力、URL、输入输出模式等的约定位置。根 Agent 通过这张卡片才能发现远程 Agent 的存在及其能力这是 A2A 协议实现服务发现的核心机制。其他可选参数源码可见的扩展点从 RemoteA2aAgent.init签名 可以看出该类的构造参数还包括timeout默认 600 秒、a2a_client_factory自定义 A2A 客户端工厂、auth_scheme/auth_credential/credential_key对远程 Agent 调用的认证配置、full_history_when_stateless无状态 Agent 是否每次携带完整会话历史等。本示例使用默认配置即可但在生产环境对接需要鉴权的远程 Agent 时这些参数是重要的扩展入口。远程服务端实现check_prime_agent 的完整剖析远程 A2A 服务端位于 contributing/samples/a2a/a2a_basic/remote_a2a/check_prime_agent/包含三个文件agent.pyAgent 实现、agent.jsonAgent 卡片、__init__.py。素数判断算法与工具agent.py 中定义了一个异步函数工具check_primeasync def check_prime(nums: list[int]) - str: Check if a given list of numbers are prime. Args: nums: The list of numbers to check. Returns: A str indicating which number is prime. primes set() for number in nums: number int(number) if number 1: continue is_prime True for i in range(2, int(number**0.5) 1): if number % i 0: is_prime False break if is_prime: primes.add(number) return ( No prime numbers found. if not primes else f{, .join(str(num) for num in primes)} are prime numbers. )算法要点支持批量输入nums: list[int]一次性判断多个数字去重后返回素数集合试除法优化只需遍历到int(number ** 0.5)将时间复杂度控制在 O(√n)对示例场景足够高效边界处理number 1直接跳过1 及以下不是素数返回值语义化无素数时返回No prime numbers found.否则返回2, 3, 5 are prime numbers.这类可读文本。远程 Agent 的配置同一文件中远程服务端定义了自己的 Agentfrom google.adk import Agent root_agent Agent( namecheck_prime_agent, descriptioncheck prime agent that can check whether numbers are prime., instruction You check whether numbers are prime. When checking prime numbers, call the check_prime tool with a list of integers. Be sure to pass in a list of integers. You should never pass in a string. You should not rely on the previous history on prime results. , tools[ check_prime, ], generate_content_configtypes.GenerateContentConfig( safety_settings[ types.SafetySetting( # avoid false alarm about rolling dice. categorytypes.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, thresholdtypes.HarmBlockThreshold.OFF, ), ] ), )注意这里有两个值得学习的细节参数类型强约束instruction 反复强调必须以整数列表传入绝不能传字符串因为check_prime内部虽然做了int(number)强转但 LLM 若以字符串形式传参会导致批量逻辑出错。这与主 Agent 中roll_agent的 instruction 风格一致——通过 prompt 约束工具调用契约是 ADK 示例中的常见最佳实践。不依赖历史声明instruction 中明确不要依赖之前关于素数结果的对话历史这确保了每次判断都基于最新传入的参数重新计算避免模型凭记忆作答。agent card 文件agent.json远程服务的 Agent 卡片位于 contributing/samples/a2a/a2a_basic/remote_a2a/check_prime_agent/agent.json{ capabilities: {}, defaultInputModes: [ text/plain ], defaultOutputModes: [ application/json ], description: An agent specialized in checking whether numbers are prime. It can efficiently determine the primality of individual numbers or lists of numbers., name: check_prime_agent, skills: [ { id: prime_checking, name: Prime Number Checking, description: Check if numbers in a list are prime using efficient mathematical algorithms, tags: [ mathematical, computation, prime, numbers ] } ], url: http://localhost:8001/a2a/check_prime_agent, version: 1.0.0 }字段含义name/descriptionAgent 的标识与描述供 A2A 客户端即主 Agent 侧的RemoteA2aAgent发现和理解url远程 Agent 的 RPC 端点地址是整张卡片中最关键的字段下文部署一节详述skills声明 Agent 具备的能力此处为prime_checking带tags便于检索匹配defaultInputModes/defaultOutputModes声明默认输入为纯文本、默认输出为 JSONcapabilities当前为空对象表示未声明额外能力如流式、状态持久化等。根 Agent 编排委托逻辑与少样本工具根 Agent 负责把用户请求路由到正确的子 Agent并支持链式操作。其定义同样在 contributing/samples/a2a/a2a_basic/agent.py 中example_tool ExampleTool([ { input: { role: user, parts: [{text: Roll a 6-sided die.}], }, output: [ {role: model, parts: [{text: I rolled a 4 for you.}]} ], }, { input: { role: user, parts: [{text: Is 7 a prime number?}], }, output: [{ role: model, parts: [{text: Yes, 7 is a prime number.}], }], }, { input: { role: user, parts: [{text: Roll a 10-sided die and check if its prime.}], }, output: [ { role: model, parts: [{text: I rolled an 8 for you.}], }, { role: model, parts: [{text: 8 is not a prime number.}], }, ], }, ]) root_agent Agent( nameroot_agent, instruction You are a helpful assistant that can roll dice and check if numbers are prime. You delegate rolling dice tasks to the roll_agent and prime checking tasks to the prime_agent. Follow these steps: 1. If the user asks to roll a die, delegate to the roll_agent. 2. If the user asks to check primes, delegate to the prime_agent. 3. If the user asks to roll a die and then check if the result is prime, call roll_agent first, then pass the result to prime_agent. Always clarify the results before proceeding. , global_instruction( You are DicePrimeBot, ready to roll dice and check prime numbers. ), sub_agents[roll_agent, prime_agent], tools[example_tool], generate_content_configtypes.GenerateContentConfig( safety_settings[ types.SafetySetting( # avoid false alarm about rolling dice. categorytypes.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, thresholdtypes.HarmBlockThreshold.OFF, ), ] ), )编排机制拆解sub_agents声明sub_agents[roll_agent, prime_agent]同时注册了本地子 Agent 和远程 A2A Agent。对根 Agent 而言两者都是可转交transfer的下游节点LLM 根据用户意图自主选择。instruction中的显式路由规则三步规则覆盖了三种请求形态掷骰子 / 判素数 / 组合任务特别是第 3 步明确了组合任务的调用顺序——先roll_agent后prime_agent再把结果串联。global_instruction角色设定You are DicePrimeBot, ready to roll dice and check prime numbers.为根 Agent 注入统一人格与能力边界与instruction的任务流程分工明确。ExampleTool少样本引导ExampleTool是 ADK 内置工具实现位于 example_tool.py它并不执行真实操作而是在每次 LLM 请求时把示例追加为指令引导模型模仿输入-输出模式。其process_llm_request会基于当前用户输入从示例列表中挑选最匹配的通过example_util.build_example_si构建系统指令。示例中三个问答对分别对应掷骰子判素数组合操作让模型在零训练的情况下也能稳定复现正确的回复格式与调用路径。环境搭建与运行两条命令启动整套系统运行该示例需要两个独立进程。前提是当前环境已安装 ADK 及其 A2A 相关依赖。第 1 步启动远程 A2A 服务在第一个终端执行# Start the remote a2a server that serves the check prime agent on port 8001 adk api_server --a2a --port 8001 contributing/samples/a2a/a2a_basic/remote_a2a--a2a标志表示以 A2A 模式启动 API 服务器--port 8001指定远程服务监听端口目录参数指向remote_a2a该目录下的check_prime_agent即被发布的服务 Agent。启动后服务会在http://localhost:8001/a2a/check_prime_agent暴露 A2A RPC 端点并在/.well-known/agent.json提供 Agent 卡片。第 2 步运行主 AgentWeb 服务器在另一个终端执行# In a separate terminal, run the adk web server adk web contributing/samples/a2aadk web启动 ADK 的本地 Web 开发服务器默认端口 8000并加载contributing/samples/a2a目录下的示例两个进程分别占用8000本地 Web UI与8001远程 A2A 服务端口互不冲突。交互示例两个服务都就绪后即可通过 Web UI 与根 Agent 对话。README 给出了三组典型交互覆盖全部编排路径简单掷骰子走 roll_agentUser: Roll a 6-sided die Bot: I rolled a 4 for you.素数判断走 prime_agent 远程调用User: Is 7 a prime number? Bot: Yes, 7 is a prime number.组合操作先 roll_agent 再 prime_agentUser: Roll a 10-sided die and check if its prime Bot: I rolled an 8 for you. Bot: 8 is not a prime number.第三个例子最能体现多智能体协作价值根 Agent 自动完成掷骰子 → 取结果 → 传给远程素数服务 → 汇总回复的全链路对用户而言只是发出了一句自然语言请求。代码结构速览contributing/samples/a2a/a2a_basic/ ├── agent.py # 主 Agentroll_agent / prime_agent / root_agent / ExampleTool ├── __init__.py # 导入 agent 模块 └── remote_a2a/ └── check_prime_agent/ ├── agent.py # 远程素数服务的 Agent 与 check_prime 工具 ├── agent.json # A2A Agent 卡片含 url 端点声明 └── __init__.py从仓库结构看contributing/samples/a2a目录下还提供了同主题的进阶变体例如 a2a_authA2A 认证、a2a_human_in_loop人类介入、a2a_state_forwarding状态转发、a2a_root 等可作为本示例的延伸学习材料。部署到其他环境必须更新 agent card 的 url 字段当把远程 A2A Agent 部署到不同环境如 Cloud Run、不同主机/端口时必须同步更新 agent.json 中的url字段本地开发{ url: http://localhost:8001/a2a/check_prime_agent, ... }Cloud Run 示例{ url: https://your-service-abc123-uc.a.run.app/a2a/check_prime_agent, ... }自定义主机/端口示例{ url: https://your-domain.com:9000/a2a/check_prime_agent, ... }重要提示url字段必须指向远程 A2A Agent 实际部署并可访问的 RPC 端点。同时主 Agent 侧构造RemoteA2aAgent时传入的 agent card URL 也必须与该服务器实际运行位置保持一致。也就是说存在两处一致性要求agent.json中的url指向真实部署地址主 Agent 侧RemoteA2aAgent(agent_card...)能通过/.well-known/agent.json拉取到这张卡片。两条链路任一处失配都会导致根 Agent 无法发现或无法调用远程服务。故障排查指南连接问题确保本地 ADK Web 服务器运行在 8000 端口确保远程 A2A 服务器运行在 8001 端口检查是否有防火墙阻止 localhost 连接验证 agent.json 中的url字段与实际部署位置一致验证传给RemoteA2aAgent构造函数的 agent card URL 与正在运行的 A2A 服务器相匹配。Agent 无响应同时检查本地 ADK Web 服务器8000与远程 A2A 服务器8001两侧的日志检查 Agent 的 instruction 是否清晰、无歧义尤其是工具调用的参数类型约束再次核对 agent.json 中的 RPC URL 是否正确且可访问。从实现层面补充一个定位技巧RemoteA2aAgent启动时会先解析 agent card源码中的_resolve_agent_card路径见 remote_a2a_agent.py。若卡片拉取失败或字段非法请求根本不会进入远程 Agent 的执行阶段因此排查时可以用curl http://localhost:8001/a2a/check_prime_agent/.well-known/agent.json之类的命令先验证卡片端点本身是否可达、返回是否完整。扩展方向README 为本示例列出了五条扩展路径结合源码可进一步落地增加更多数学运算仿照check_prime添加因式分解、平方根等工具与子 Agent创建更多远程 Agent复用remote_a2a/check_prime_agent的目录结构agent.py agent.json新建服务并在根 Agent 中注册新的RemoteA2aAgent实现更复杂的委托逻辑扩展root_agent的instruction路由规则支持多级、条件式任务分发添加持久化状态管理远程 Agent 可在agent.json的capabilities中声明状态支持配合RemoteA2aAgent的会话状态管理机制使用集成外部 API 或数据库把外部调用封装为函数工具挂载到对应子 Agent。小结A2A Basic 示例以最小可运行的形式展示了 ADK 多智能体协作的三层能力本地子 Agent 集成roll_agent 函数工具、远程 A2A Agent 集成RemoteA2aAgent agent card 服务发现、根 Agent 智能编排instruction 路由 ExampleTool 少样本引导。其中agent.json的url字段是连接本地与远程世界的桥梁也是部署迁移时最需要留意的配置项。掌握了这个示例你就具备了在 ADK 中搭建跨服务多智能体系统的基础骨架。【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考