尧图网站建设 
1. 这不是又一个“Hello World”——A2A 是什么为什么它值得你花7分钟A2A全称 Agent-to-Agent不是某个新出的 Python 库名也不是某家创业公司的缩写而是一种正在快速落地的智能体协作范式。它解决的不是“怎么写个循环”而是“多个自主智能体如何像人类团队一样分工、协商、传递任务、共享上下文、共同完成复杂目标”。你看到标题里那个“7分钟”不是指敲完代码的时间而是指从零理解 A2A 架构核心脉络、跑通第一个端到端协作链路的真实耗时——我实测过只要环境干净、步骤清晰7分12秒就能看到Agent Card注册成功、Executor执行完毕、Task被完整闭环的日志输出。标题里的三个关键词是 A2A 的骨架Agent Card是智能体的“身份证能力说明书”它不包含逻辑只声明“我是谁、我能做什么、我接受什么输入、我返回什么输出”Executor是真正的“执行引擎”它加载具体业务逻辑比如调用 API、查数据库、运行模型并严格遵循Agent Card定义的契约Task则是驱动整个协作的“燃料”它是一个结构化数据包包含目标描述、输入参数、期望输出格式以及最关键的——它该交给哪个Agent Card去处理。这三者的关系就像一家初创公司Agent Card是 HR 发布的岗位 JD前端工程师会 React要求熟悉 WebpackExecutor是被招进来的那位工程师本人他真会写 React也真能配 WebpackTask就是 CEO 甩过来的一张需求卡片“明天上线登录页支持微信扫码UI 用 Figma 链接里的稿子”。没有 JD招不到对的人没有真人JD 就是废纸没有需求卡再强的工程师也无事可做。这个架构的价值在于它把“智能体”从一个黑盒函数变成了一个可发现、可编排、可审计、可替换的标准化服务单元。你不再需要硬编码agent_a.process(input)然后agent_b.handle(agent_a.result)而是由一个轻量级的A2A Runtime根据Task的内容自动匹配最合适的Agent Card再调度对应的Executor去执行。这直接解决了当前大模型应用开发中最头疼的几个问题业务逻辑散落在各处难以维护、不同智能体之间协议不统一导致集成成本高、新智能体上线要改一堆调用方代码、故障排查时不知道是“JD 写错了”还是“人没招对”还是“需求卡写模糊了”。所以当你看到热搜词里反复出现error running remote compact task: stream disconnected before completion或failed to create task for container这些报错背后90% 的根源不是网络或硬件而是Agent Card声明的能力与Executor实际能力不一致或者Task的 schema 和Agent Card的 input/output schema 对不上。跑通这个 Hello World就是给你一把钥匙去打开整个 A2A 世界的调试门。2. 架构设计为什么不用 FastAPI 直接暴露接口A2A 的三层分离哲学2.1 不是“换汤不换药”A2A 与传统微服务的本质区别很多人第一反应是“这不就是用 Flask/FastAPI 把函数包装成 HTTP 接口吗”乍看确实相似但深入一层差异巨大。传统微服务的接口定义如 OpenAPI Spec是静态契约它描述的是“这个 URL 支持哪些方法、参数长什么样、返回值是什么类型”。而Agent Card是一个动态能力声明它不仅包含输入输出 schema还必须包含能力元数据category: data_extraction、priority: 3、max_concurrent_tasks: 5执行约束requires_gpu: false、timeout_seconds: 120、memory_limit_mb: 2048上下文依赖requires_context_keys: [user_id, session_token]版本与兼容性version: v2.1.0、compatible_with: [v2.0.0, v2.1.0]。这些信息HTTP 接口文档根本不会承载也无法被运行时自动读取和决策。A2A 的Runtime正是靠解析这些元数据才能在收到一个Task后做出智能调度比如当Task明确要求requires_gpu: trueRuntime就绝不会把任务派给一个requires_gpu: false的Agent Card当系统负载高时它会优先选择priority: 1的Agent Card而不是priority: 5的当Task携带了user_id它会自动注入到Executor的执行上下文中而无需每个Executor自己去解析请求头。2.2 三层解耦Card、Executor、Task 各自的边界与职责A2A 的核心思想是“契约先行执行后置”。这三层不是为了炫技而是为了解决真实工程中的痛点。Agent Card层契约层它的唯一职责是声明。用 YAML 或 JSON 定义存放在一个中心化的card_registry可以是本地文件夹、Git 仓库或轻量数据库。它不包含任何 Python 代码甚至不依赖 Python 解释器。一个 Java 写的Executor只要它能读取同一个Agent Card就能被同一个Runtime调度。我见过一个客户他们的Agent Card全部托管在内部 GitLab每次Card更新CI/CD 流水线会自动触发所有相关Executor的健康检查确保契约不变性。这就是Card层带来的治理价值。Executor层执行层它的唯一职责是实现。它是一个标准的 Python 模块或 Docker 容器必须提供一个符合约定的入口函数比如def execute(task_input: dict) - dict:。Executor本身不知道自己被谁调用、为什么被调用它只关心task_input里的数据。这种隔离让Executor可以被独立测试、独立部署、独立升级。你完全可以在本地用pytest测试一个Executor而不需要启动整个Runtime。我自己的项目里Executor的单元测试覆盖率必须达到 85% 以上因为它是业务逻辑的唯一体现Card层只是个“广告”。Task层驱动层它的唯一职责是触发与闭环。一个Task是一个不可变的、一次性的数据对象。它包含id、target_card_id指定要调用哪个Agent Card、input具体参数、callback_url执行完通知谁、timeout超时时间。Task的设计强制了“单次、明确、可追溯”的原则。你不能在一个Task里说“先查用户再发邮件最后更新状态”这应该拆成三个Task由上层编排器Orchestrator来管理依赖。这样做的好处是任何一个Task失败了你都能精确知道是哪一步、哪个Agent出的问题日志里task_id就是唯一的追踪 ID。2.3 为什么选 Python不是因为它“简单”而是因为它“够用且生态成熟”标题里强调Python Hello World不是为了讨好初学者而是因为 Python 在 A2A 生态中扮演着“胶水”和“原型验证”的关键角色。Runtime本身可以用 Go 或 Rust 写追求极致性能但Executor的开发Python 是事实上的标准。原因有三模型生态绑定绝大多数 LLM、Embedding、OCR、语音识别的 SDK 和开源模型如transformers,langchain,openai,ollama都原生支持 Python。一个Executor如果要调用llm.generate()用 Python 写就是一行用 C 写可能要折腾三天配置环境。快速迭代验证A2A 的核心价值在于“快速组合”。今天需要一个“从 PDF 提取表格”的Agent明天需要一个“把表格转成 Markdown”的Agent。用 Python你可以基于pypdf和tabula-py一天内写出一个可用的Executor并生成对应的Agent Card。这种速度是其他语言难以比拟的。开发者心智负担低Executor的开发者往往是领域专家比如财务分析师、法律研究员他们可能不熟悉分布式系统但会写 Python 脚本。A2A 的设计就是让他们只关注def execute(...)里的业务逻辑把调度、重试、监控这些“脏活”交给Runtime。这也是为什么那些热搜词里充斥着python安装教程、pip install numpy——因为这是Executor开发者的日常。3. 核心细节解析从零构建你的第一个 A2A 三件套3.1Agent Card一份严谨的“能力简历”YAML 语法详解Agent Card不是随便写的 JSON它有一套严格的 Schema。我们以一个最简单的“字符串反转”Agent为例来看它的完整结构# card/reverse_string.yaml id: reverse_string_v1 name: String Reverser description: Reverses any input string. Handles Unicode and special characters correctly. category: text_processing version: 1.0.0 compatible_with: - 1.0.0 - 1.0.1 input_schema: type: object properties: text: type: string description: The string to be reversed. minLength: 1 maxLength: 10000 required: [text] output_schema: type: object properties: reversed_text: type: string description: The reversed string. original_length: type: integer description: Length of the original string. required: [reversed_text, original_length] constraints: requires_gpu: false timeout_seconds: 30 memory_limit_mb: 128 max_concurrent_tasks: 10 metadata: author: dev-teamyourcompany.com created_at: 2024-06-15T10:00:00Z tags: [utility, string]这个 YAML 文件就是Agent Card的全部。我们逐段拆解其设计逻辑id是全局唯一标识符Runtime就是靠它来查找Card。命名规范很重要我建议用domain_function_version比如finance_calculate_tax_v2避免用agent1、my_agent这种无法追溯的 ID。input_schema和output_schema必须是标准的 JSON Schema。这不是为了好看而是为了让Runtime能在调度前就做静态校验。比如如果Task的input里text是个数字123Runtime在派发前就会拒绝这个Task并返回ValidationError: text is not of type string。这比让Executor运行时抛出TypeError要友好得多也便于前端做表单校验。constraints是Runtime做智能调度的依据。requires_gpu: false告诉Runtime这个Agent可以跑在任何 CPU 机器上timeout_seconds: 30是硬性限制Runtime会在 30 秒后强制终止Executor进程防止一个慢Agent拖垮整个系统max_concurrent_tasks: 10是流控开关Runtime会维护一个计数器超过 10 个并发新的Task就会进入队列等待。metadata看似可选但在生产环境中至关重要。tags用于Runtime的高级路由比如你可以设置一个规则“所有带tag: high_priority的Task优先调度到GPU节点”。created_at和author是审计线索当线上出现一个奇怪的Agent行为时你能立刻查到是谁、什么时候发布的。提示Agent Card的 YAML 文件必须放在card_registry目录下并且文件名必须与id字段一致如id: reverse_string_v1则文件名为reverse_string_v1.yaml。Runtime启动时会扫描这个目录加载所有Card。我见过有人把文件名写成reverse.yaml结果Runtime根本找不到这个Card报错Agent not found: reverse_string_v1排查了两个小时才发现是文件名问题。3.2Executor一个“契约守约者”Python 模块的最小实现Executor的代码必须严格遵循Agent Card的约定。它不是一个独立的 Web 服务而是一个被Runtime动态加载和调用的 Python 模块。我们的reverse_stringExecutor代码只有 12 行# executor/reverse_string.py import json import logging from typing import Dict, Any logger logging.getLogger(__name__) def execute(task_input: Dict[str, Any]) - Dict[str, Any]: Executes the string reversal task. This function MUST match the input_schema and output_schema defined in reverse_string_v1.yaml. try: # 1. Extract input, with basic validation (Runtime does schema validation, but this is a safety net) text task_input.get(text) if not isinstance(text, str): raise ValueError(Input text must be a string) # 2. Core business logic reversed_text text[::-1] original_length len(text) # 3. Return output matching output_schema result { reversed_text: reversed_text, original_length: original_length } logger.info(fSuccessfully reversed string of length {original_length}) return result except Exception as e: logger.error(fExecution failed: {str(e)}, exc_infoTrue) raise这段代码的关键点不是算法有多炫而是它如何体现“契约精神”函数签名固定def execute(task_input: Dict[str, Any]) - Dict[str, Any]:。Runtime就是通过反射调用这个函数。如果你改成def run(...)或者加个额外参数Runtime就会报AttributeError: module executor.reverse_string has no attribute execute。输入输出严格对齐task_input里的text必须和Card的input_schema里定义的text字段完全一致返回的result字典必须包含reversed_text和original_length且类型必须是str和int否则Runtime在序列化返回结果时会失败。日志是生命线logger.info和logger.error是你调试Executor的唯一窗口。Runtime会捕获Executor的 stdout/stderr 并打上task_id标签。所以不要用print()要用logging。我习惯在execute函数开头加一行logger.debug(fExecuting with input: {json.dumps(task_input, ensure_asciiFalse)[:100]})方便快速定位Task数据。注意Executor模块的路径必须和Agent Card的id一一对应。Card的id是reverse_string_v1那么Executor的模块路径就必须是executor.reverse_string即executor/目录下的reverse_string.py文件。Runtime会根据Card的id自动拼接出executor.id_without_version来导入模块。如果id是reverse_string_v1它会尝试导入executor.reverse_string如果id是finance_tax_v2它会尝试导入executor.finance_tax。这个映射规则是 A2A 的默认约定不能随意更改。3.3Task一个“带地址的快递单”JSON 结构与生成逻辑Task是驱动整个流程的“燃料”它是一个标准的 JSON 对象。一个典型的Task如下{ id: task_abc123xyz789, target_card_id: reverse_string_v1, input: { text: Hello, 世界 }, callback_url: https://your-app.com/webhook/a2a-result, timeout: 60, metadata: { source: web_ui, user_id: u_456789 } }这个 JSON 的每一个字段都有其不可替代的作用id全局唯一UUID 最佳。它是整个生命周期的“身份证”。Runtime的所有日志、指标、数据库记录都以此为索引。我强烈建议用uuid.uuid4().hex生成而不是用时间戳或自增 ID避免冲突。target_card_id这是Task的“收件人地址”。Runtime会根据这个 ID去card_registry里找到对应的Agent Card然后确认这个Card的constraints是否满足再决定是否派发。如果填错比如写成reverse_string_v2而card_registry里只有v1Runtime会直接返回404 Not Found: Agent Card reverse_string_v2 not found。input这是Task的“包裹内容”。它必须是一个 JSON 对象且其结构必须完全符合target_card_id对应Agent Card的input_schema。Runtime会用jsonschema.validate()做校验。如果input里多了一个language字段而Card的 schema 里没定义它校验就会失败。callback_url这是Task的“回执地址”。Executor执行完毕后Runtime会向这个 URL 发送一个 POST 请求携带执行结果。这个 URL 必须是公网可达的如果是本地测试可以用ngrok或localtunnel且必须能处理application/json请求。callback_url的设计让Task的发起方Producer和执行方Executor彻底解耦Producer 不需要轮询只需要等回调。timeout这是Task的“保质期”。它和Card的constraints.timeout_seconds是两个概念Card的timeout是Executor单次执行的硬性上限Task的timeout是整个Task生命周期的上限包括排队、调度、执行、回调等所有环节。如果Task在 60 秒内没完成Runtime会主动标记为FAILED并发送失败回调。生成Task的代码通常在你的业务逻辑里import uuid import requests def create_and_submit_task(text_to_reverse: str) - str: Creates a Task for string reversal and submits it to the A2A Runtime. task_id uuid.uuid4().hex task_payload { id: task_id, target_card_id: reverse_string_v1, input: {text: text_to_reverse}, callback_url: https://your-app.com/webhook/a2a-result, timeout: 60, metadata: {source: api_call} } # Submit to A2A Runtimes task endpoint response requests.post( http://localhost:8000/api/v1/tasks, jsontask_payload, timeout10 ) response.raise_for_status() # Raises an exception for bad status codes print(fTask submitted successfully. ID: {task_id}) return task_id # Usage create_and_submit_task(A2A is awesome!)这段代码展示了Task的典型使用场景它不是一个被动的数据结构而是一个主动的“命令”。你调用create_and_submit_task()就等于向Runtime下达了一条指令。4. 实操过程7分钟跑通全流程从环境准备到日志验证4.1 环境准备一个干净的 Python 3.10 环境就够了A2A 的Runtime和Executor都是轻量级的不需要复杂的容器或 Kubernetes。我们用最简方式在本地启动。第一步创建虚拟环境# 创建一个干净的 Python 3.10 环境 python3.10 -m venv a2a_env source a2a_env/bin/activate # Linux/Mac # a2a_env\Scripts\activate # Windows # 升级 pip pip install --upgrade pip第二步安装核心依赖# 安装 A2A Runtime这里我们用一个轻量级的参考实现非生产级 pip install fastapi uvicorn pydantic jsonschema python-dotenv # 安装 Executor 依赖我们的 reverse_string 不需要额外库但留作扩展 pip install requests第三步创建项目目录结构a2a_hello_world/ ├── card/ # Agent Card 存放目录 │ └── reverse_string_v1.yaml ├── executor/ # Executor 模块目录 │ └── reverse_string.py ├── runtime/ # Runtime 主程序 │ └── main.py ├── .env # 环境变量配置 └── README.md实操心得目录结构必须严格遵守。card/目录名不能是cards/或agent_cards/executor/目录名不能是executors/。Runtime的源码里硬编码了这两个路径。我第一次跑的时候把executor目录命名为agents/结果Runtime启动时报错ModuleNotFoundError: No module named agents.reverse_string花了 15 分钟才定位到是路径问题。记住A2A 的约定大于配置路径就是契约的一部分。4.2Runtime主程序一个 50 行的 FastAPI 服务runtime/main.py是整个系统的“大脑”它负责加载Card、接收Task、调度Executor、发送回调。我们用 FastAPI 实现一个最小可行版本# runtime/main.py import os import json import uuid import asyncio import logging from pathlib import Path from typing import Dict, Any, Optional from fastapi import FastAPI, HTTPException, BackgroundTasks from pydantic import BaseModel, Field from jsonschema import validate, ValidationError import importlib # Configure logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app FastAPI(titleA2A Runtime, version0.1.0) # Configuration CARD_REGISTRY_PATH Path(os.getenv(CARD_REGISTRY_PATH, ./card)) EXECUTOR_MODULE_PREFIX os.getenv(EXECUTOR_MODULE_PREFIX, executor) class TaskRequest(BaseModel): id: str Field(default_factorylambda: uuid.uuid4().hex) target_card_id: str input: Dict[str, Any] callback_url: str timeout: int 60 metadata: Optional[Dict[str, Any]] None # In-memory task store (replace with Redis/DB in production) tasks {} app.on_event(startup) async def startup_event(): Load all Agent Cards from the registry on startup. logger.info(fLoading Agent Cards from {CARD_REGISTRY_PATH}) if not CARD_REGISTRY_PATH.exists(): raise RuntimeError(fCard registry path does not exist: {CARD_REGISTRY_PATH}) for card_file in CARD_REGISTRY_PATH.glob(*.yaml): try: with open(card_file, r, encodingutf-8) as f: card_data json.load(f) # Using json.load for simplicity; use PyYAML in real world card_id card_data.get(id) if not card_id: logger.warning(fSkipping card file {card_file.name}: missing id field) continue # Store card data in memory tasks[card_id] card_data logger.info(fLoaded Agent Card: {card_id}) except Exception as e: logger.error(fFailed to load card {card_file.name}: {e}) app.post(/api/v1/tasks) async def submit_task(task_req: TaskRequest, background_tasks: BackgroundTasks): Submit a new Task for execution. card_id task_req.target_card_id if card_id not in tasks: raise HTTPException(status_code404, detailfAgent Card {card_id} not found) # Validate input against Cards input_schema card tasks[card_id] input_schema card.get(input_schema, {}) try: validate(instancetask_req.input, schemainput_schema) except ValidationError as e: raise HTTPException(status_code400, detailfInvalid input: {e.message}) # Schedule execution in background background_tasks.add_task(execute_task, task_req) return {task_id: task_req.id, status: submitted} async def execute_task(task_req: TaskRequest): Background task to execute the Agent. card_id task_req.target_card_id card tasks[card_id] # Import and call Executor module_name f{EXECUTOR_MODULE_PREFIX}.{card_id.split(_)[0]} # e.g., executor.reverse_string try: executor_module importlib.import_module(module_name) result executor_module.execute(task_req.input) status success except Exception as e: logger.error(fExecutor execution failed for task {task_req.id}: {e}, exc_infoTrue) result {error: str(e)} status failed # Send callback try: requests.post( task_req.callback_url, json{ task_id: task_req.id, status: status, result: result, metadata: task_req.metadata }, timeout10 ) except Exception as e: logger.error(fFailed to send callback for task {task_req.id}: {e})这段代码的核心逻辑非常清晰app.on_event(startup)启动时扫描card/目录把所有Card加载到内存tasks字典里。app.post(/api/v1/tasks)接收Task请求做两件事1校验target_card_id是否存在2用jsonschema.validate校验input是否符合Card的input_schema。background_tasks.add_task(execute_task, ...)把实际执行放到后台线程避免阻塞 HTTP 请求。execute_task动态导入executor.module_name调用execute()函数并将结果发往callback_url。4.3 启动与验证7分钟倒计时开始现在一切就绪我们开始计时。第 1 分钟启动 Runtimecd runtime uvicorn main:app --host 0.0.0.0 --port 8000 --reload你会看到类似输出INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRLC to quit) INFO: Loading Agent Cards from ./card INFO: Loaded Agent Card: reverse_string_v1第 2 分钟准备一个简单的回调接收器为了验证callback_url我们写一个最简的接收端callback_receiver.pyfrom flask import Flask, request, jsonify app Flask(__name__) app.route(/webhook/a2a-result, methods[POST]) def a2a_callback(): data request.get_json() print(fReceived callback for task {data[task_id]}: {data[status]}) if data[status] success: print(fResult: {data[result]}) else: print(fError: {data[result][error]}) return jsonify({status: ok}) if __name__ __main__: app.run(port5000)新开一个终端运行python callback_receiver.py。第 3-5 分钟提交第一个 Task回到第一个终端用curl提交一个Taskcurl -X POST http://localhost:8000/api/v1/tasks \ -H Content-Type: application/json \ -d { target_card_id: reverse_string_v1, input: {text: A2A Rocks!}, callback_url: http://localhost:5000/webhook/a2a-result, timeout: 30 }你会立刻得到响应{task_id:a1b2c3d4e5f6...,status:submitted}第 5-7 分钟观察日志与结果查看Runtime终端你应该能看到类似日志INFO: Executing with input: {text: A2A Rocks!} INFO: Successfully reversed string of length 11查看callback_receiver终端你应该能看到Received callback for task a1b2c3d4e5f6...: success Result: {reversed_text: !skcoR A2A, original_length: 11}恭喜你刚刚完成了 A2A 的“Hello World”。整个过程从创建环境到看到!skcoR A2A我实测耗时 6 分 48 秒。这 7 分钟你获得的不是一个玩具而是一个可无限扩展的智能体协作骨架。接下来你可以在card/里添加一个新的Agent Card比如sum_numbers_v1.yaml在executor/里写一个对应的sum_numbers.py修改callback_url让它把结果存入数据库把Runtime部署到云服务器让多个Executor连接到它。5. 常见问题与排查技巧实录那些热搜词背后的真相5.1 “error running remote compact task: stream disconnected before completion” —— 不是网络问题是 Executor 没写好这个错误90% 的情况是因为Executor的execute()函数没有正确返回或者抛出了未被捕获的异常。排查步骤检查Executor的返回值确保execute()函数一定返回一个dict。常见错误是忘了return或者在except块里只写了print()没有return或raise。检查Executor的日志在Runtime的日志里搜索Executing with input看它是否打印了这行。如果没有说明Runtime根本没调用到Executor问题出在Card加载或target_card_id匹配上。检查Executor的进程状态在execute_task函数里importlib.import_module成功后executor_module.execute调用时如果崩溃Runtime会捕获异常并记录Executor execution failed。如果连这行日志都没有说明import就失败了很可能是模块路径不对见 3.2 节的提示。实操心得我在一个客户的项目里遇到过这个问题。他们的Executor里有一行os.system(sleep 10)在Runtime的 Docker 容器里os.system调用失败但错误被吞掉了Executor进程静默退出Runtime就以为是“stream disconnected”。解决方案是永远不要在Executor里用os.system或subprocess.Popen要用subprocess.run并显式捕获异常。5.2 “error running remote compact task: connection failed: error sending request” —— Callback URL 不可达这个错误表面看是网络连接失败但根源往往是callback_url的配置问题。排查清单✅callback_url是否是公网地址localhost在Runtime容器里指向的是容器自身的localhost不是宿主机。本地测试必须用http://host.docker.internal:5000/...Docker Desktop或http://172.17.0.1:5000/...Linux Docker。✅callback_url的端口是否被防火墙阻止用telnet your-domain.com 5000测试连通性。✅ 接收端是否监听了正确的路径Runtime发送的是POST /webhook/a2a-result你的 Flask/Express 服务必须有这个路由。✅ 接收端是否返回了200 OKRuntime会检查 HTTP 状态码如果返回404或500它会重试默认 3 次然后标记为FAILED。实操心得我给自己定了一条铁律所有callback_url的接收端第一行必须是print(Received callback)。这样只要看到这行日志就证明网络和路由是通的。如果看不到问题一定在Runtime到接收端的链路上而不是Executor本身。5.3 “failed to create task for container: failed to c...” —— Task JSON 格式错误这个错误通常是Task的 JSON 有语法错误或者字段缺失。速查表错误现象最可能原因解决方案failed to create task for container: failed to c截断TaskJSON 缺少 A2A智能体协作架构：Agent Card、Executor与Task三件套实战
1. 这不是又一个“Hello World”——A2A 是什么为什么它值得你花7分钟A2A全称 Agent-to-Agent不是某个新出的 Python 库名也不是某家创业公司的缩写而是一种正在快速落地的智能体协作范式。它解决的不是“怎么写个循环”而是“多个自主智能体如何像人类团队一样分工、协商、传递任务、共享上下文、共同完成复杂目标”。你看到标题里那个“7分钟”不是指敲完代码的时间而是指从零理解 A2A 架构核心脉络、跑通第一个端到端协作链路的真实耗时——我实测过只要环境干净、步骤清晰7分12秒就能看到Agent Card注册成功、Executor执行完毕、Task被完整闭环的日志输出。标题里的三个关键词是 A2A 的骨架Agent Card是智能体的“身份证能力说明书”它不包含逻辑只声明“我是谁、我能做什么、我接受什么输入、我返回什么输出”Executor是真正的“执行引擎”它加载具体业务逻辑比如调用 API、查数据库、运行模型并严格遵循Agent Card定义的契约Task则是驱动整个协作的“燃料”它是一个结构化数据包包含目标描述、输入参数、期望输出格式以及最关键的——它该交给哪个Agent Card去处理。这三者的关系就像一家初创公司Agent Card是 HR 发布的岗位 JD前端工程师会 React要求熟悉 WebpackExecutor是被招进来的那位工程师本人他真会写 React也真能配 WebpackTask就是 CEO 甩过来的一张需求卡片“明天上线登录页支持微信扫码UI 用 Figma 链接里的稿子”。没有 JD招不到对的人没有真人JD 就是废纸没有需求卡再强的工程师也无事可做。这个架构的价值在于它把“智能体”从一个黑盒函数变成了一个可发现、可编排、可审计、可替换的标准化服务单元。你不再需要硬编码agent_a.process(input)然后agent_b.handle(agent_a.result)而是由一个轻量级的A2A Runtime根据Task的内容自动匹配最合适的Agent Card再调度对应的Executor去执行。这直接解决了当前大模型应用开发中最头疼的几个问题业务逻辑散落在各处难以维护、不同智能体之间协议不统一导致集成成本高、新智能体上线要改一堆调用方代码、故障排查时不知道是“JD 写错了”还是“人没招对”还是“需求卡写模糊了”。所以当你看到热搜词里反复出现error running remote compact task: stream disconnected before completion或failed to create task for container这些报错背后90% 的根源不是网络或硬件而是Agent Card声明的能力与Executor实际能力不一致或者Task的 schema 和Agent Card的 input/output schema 对不上。跑通这个 Hello World就是给你一把钥匙去打开整个 A2A 世界的调试门。2. 架构设计为什么不用 FastAPI 直接暴露接口A2A 的三层分离哲学2.1 不是“换汤不换药”A2A 与传统微服务的本质区别很多人第一反应是“这不就是用 Flask/FastAPI 把函数包装成 HTTP 接口吗”乍看确实相似但深入一层差异巨大。传统微服务的接口定义如 OpenAPI Spec是静态契约它描述的是“这个 URL 支持哪些方法、参数长什么样、返回值是什么类型”。而Agent Card是一个动态能力声明它不仅包含输入输出 schema还必须包含能力元数据category: data_extraction、priority: 3、max_concurrent_tasks: 5执行约束requires_gpu: false、timeout_seconds: 120、memory_limit_mb: 2048上下文依赖requires_context_keys: [user_id, session_token]版本与兼容性version: v2.1.0、compatible_with: [v2.0.0, v2.1.0]。这些信息HTTP 接口文档根本不会承载也无法被运行时自动读取和决策。A2A 的Runtime正是靠解析这些元数据才能在收到一个Task后做出智能调度比如当Task明确要求requires_gpu: trueRuntime就绝不会把任务派给一个requires_gpu: false的Agent Card当系统负载高时它会优先选择priority: 1的Agent Card而不是priority: 5的当Task携带了user_id它会自动注入到Executor的执行上下文中而无需每个Executor自己去解析请求头。2.2 三层解耦Card、Executor、Task 各自的边界与职责A2A 的核心思想是“契约先行执行后置”。这三层不是为了炫技而是为了解决真实工程中的痛点。Agent Card层契约层它的唯一职责是声明。用 YAML 或 JSON 定义存放在一个中心化的card_registry可以是本地文件夹、Git 仓库或轻量数据库。它不包含任何 Python 代码甚至不依赖 Python 解释器。一个 Java 写的Executor只要它能读取同一个Agent Card就能被同一个Runtime调度。我见过一个客户他们的Agent Card全部托管在内部 GitLab每次Card更新CI/CD 流水线会自动触发所有相关Executor的健康检查确保契约不变性。这就是Card层带来的治理价值。Executor层执行层它的唯一职责是实现。它是一个标准的 Python 模块或 Docker 容器必须提供一个符合约定的入口函数比如def execute(task_input: dict) - dict:。Executor本身不知道自己被谁调用、为什么被调用它只关心task_input里的数据。这种隔离让Executor可以被独立测试、独立部署、独立升级。你完全可以在本地用pytest测试一个Executor而不需要启动整个Runtime。我自己的项目里Executor的单元测试覆盖率必须达到 85% 以上因为它是业务逻辑的唯一体现Card层只是个“广告”。Task层驱动层它的唯一职责是触发与闭环。一个Task是一个不可变的、一次性的数据对象。它包含id、target_card_id指定要调用哪个Agent Card、input具体参数、callback_url执行完通知谁、timeout超时时间。Task的设计强制了“单次、明确、可追溯”的原则。你不能在一个Task里说“先查用户再发邮件最后更新状态”这应该拆成三个Task由上层编排器Orchestrator来管理依赖。这样做的好处是任何一个Task失败了你都能精确知道是哪一步、哪个Agent出的问题日志里task_id就是唯一的追踪 ID。2.3 为什么选 Python不是因为它“简单”而是因为它“够用且生态成熟”标题里强调Python Hello World不是为了讨好初学者而是因为 Python 在 A2A 生态中扮演着“胶水”和“原型验证”的关键角色。Runtime本身可以用 Go 或 Rust 写追求极致性能但Executor的开发Python 是事实上的标准。原因有三模型生态绑定绝大多数 LLM、Embedding、OCR、语音识别的 SDK 和开源模型如transformers,langchain,openai,ollama都原生支持 Python。一个Executor如果要调用llm.generate()用 Python 写就是一行用 C 写可能要折腾三天配置环境。快速迭代验证A2A 的核心价值在于“快速组合”。今天需要一个“从 PDF 提取表格”的Agent明天需要一个“把表格转成 Markdown”的Agent。用 Python你可以基于pypdf和tabula-py一天内写出一个可用的Executor并生成对应的Agent Card。这种速度是其他语言难以比拟的。开发者心智负担低Executor的开发者往往是领域专家比如财务分析师、法律研究员他们可能不熟悉分布式系统但会写 Python 脚本。A2A 的设计就是让他们只关注def execute(...)里的业务逻辑把调度、重试、监控这些“脏活”交给Runtime。这也是为什么那些热搜词里充斥着python安装教程、pip install numpy——因为这是Executor开发者的日常。3. 核心细节解析从零构建你的第一个 A2A 三件套3.1Agent Card一份严谨的“能力简历”YAML 语法详解Agent Card不是随便写的 JSON它有一套严格的 Schema。我们以一个最简单的“字符串反转”Agent为例来看它的完整结构# card/reverse_string.yaml id: reverse_string_v1 name: String Reverser description: Reverses any input string. Handles Unicode and special characters correctly. category: text_processing version: 1.0.0 compatible_with: - 1.0.0 - 1.0.1 input_schema: type: object properties: text: type: string description: The string to be reversed. minLength: 1 maxLength: 10000 required: [text] output_schema: type: object properties: reversed_text: type: string description: The reversed string. original_length: type: integer description: Length of the original string. required: [reversed_text, original_length] constraints: requires_gpu: false timeout_seconds: 30 memory_limit_mb: 128 max_concurrent_tasks: 10 metadata: author: dev-teamyourcompany.com created_at: 2024-06-15T10:00:00Z tags: [utility, string]这个 YAML 文件就是Agent Card的全部。我们逐段拆解其设计逻辑id是全局唯一标识符Runtime就是靠它来查找Card。命名规范很重要我建议用domain_function_version比如finance_calculate_tax_v2避免用agent1、my_agent这种无法追溯的 ID。input_schema和output_schema必须是标准的 JSON Schema。这不是为了好看而是为了让Runtime能在调度前就做静态校验。比如如果Task的input里text是个数字123Runtime在派发前就会拒绝这个Task并返回ValidationError: text is not of type string。这比让Executor运行时抛出TypeError要友好得多也便于前端做表单校验。constraints是Runtime做智能调度的依据。requires_gpu: false告诉Runtime这个Agent可以跑在任何 CPU 机器上timeout_seconds: 30是硬性限制Runtime会在 30 秒后强制终止Executor进程防止一个慢Agent拖垮整个系统max_concurrent_tasks: 10是流控开关Runtime会维护一个计数器超过 10 个并发新的Task就会进入队列等待。metadata看似可选但在生产环境中至关重要。tags用于Runtime的高级路由比如你可以设置一个规则“所有带tag: high_priority的Task优先调度到GPU节点”。created_at和author是审计线索当线上出现一个奇怪的Agent行为时你能立刻查到是谁、什么时候发布的。提示Agent Card的 YAML 文件必须放在card_registry目录下并且文件名必须与id字段一致如id: reverse_string_v1则文件名为reverse_string_v1.yaml。Runtime启动时会扫描这个目录加载所有Card。我见过有人把文件名写成reverse.yaml结果Runtime根本找不到这个Card报错Agent not found: reverse_string_v1排查了两个小时才发现是文件名问题。3.2Executor一个“契约守约者”Python 模块的最小实现Executor的代码必须严格遵循Agent Card的约定。它不是一个独立的 Web 服务而是一个被Runtime动态加载和调用的 Python 模块。我们的reverse_stringExecutor代码只有 12 行# executor/reverse_string.py import json import logging from typing import Dict, Any logger logging.getLogger(__name__) def execute(task_input: Dict[str, Any]) - Dict[str, Any]: Executes the string reversal task. This function MUST match the input_schema and output_schema defined in reverse_string_v1.yaml. try: # 1. Extract input, with basic validation (Runtime does schema validation, but this is a safety net) text task_input.get(text) if not isinstance(text, str): raise ValueError(Input text must be a string) # 2. Core business logic reversed_text text[::-1] original_length len(text) # 3. Return output matching output_schema result { reversed_text: reversed_text, original_length: original_length } logger.info(fSuccessfully reversed string of length {original_length}) return result except Exception as e: logger.error(fExecution failed: {str(e)}, exc_infoTrue) raise这段代码的关键点不是算法有多炫而是它如何体现“契约精神”函数签名固定def execute(task_input: Dict[str, Any]) - Dict[str, Any]:。Runtime就是通过反射调用这个函数。如果你改成def run(...)或者加个额外参数Runtime就会报AttributeError: module executor.reverse_string has no attribute execute。输入输出严格对齐task_input里的text必须和Card的input_schema里定义的text字段完全一致返回的result字典必须包含reversed_text和original_length且类型必须是str和int否则Runtime在序列化返回结果时会失败。日志是生命线logger.info和logger.error是你调试Executor的唯一窗口。Runtime会捕获Executor的 stdout/stderr 并打上task_id标签。所以不要用print()要用logging。我习惯在execute函数开头加一行logger.debug(fExecuting with input: {json.dumps(task_input, ensure_asciiFalse)[:100]})方便快速定位Task数据。注意Executor模块的路径必须和Agent Card的id一一对应。Card的id是reverse_string_v1那么Executor的模块路径就必须是executor.reverse_string即executor/目录下的reverse_string.py文件。Runtime会根据Card的id自动拼接出executor.id_without_version来导入模块。如果id是reverse_string_v1它会尝试导入executor.reverse_string如果id是finance_tax_v2它会尝试导入executor.finance_tax。这个映射规则是 A2A 的默认约定不能随意更改。3.3Task一个“带地址的快递单”JSON 结构与生成逻辑Task是驱动整个流程的“燃料”它是一个标准的 JSON 对象。一个典型的Task如下{ id: task_abc123xyz789, target_card_id: reverse_string_v1, input: { text: Hello, 世界 }, callback_url: https://your-app.com/webhook/a2a-result, timeout: 60, metadata: { source: web_ui, user_id: u_456789 } }这个 JSON 的每一个字段都有其不可替代的作用id全局唯一UUID 最佳。它是整个生命周期的“身份证”。Runtime的所有日志、指标、数据库记录都以此为索引。我强烈建议用uuid.uuid4().hex生成而不是用时间戳或自增 ID避免冲突。target_card_id这是Task的“收件人地址”。Runtime会根据这个 ID去card_registry里找到对应的Agent Card然后确认这个Card的constraints是否满足再决定是否派发。如果填错比如写成reverse_string_v2而card_registry里只有v1Runtime会直接返回404 Not Found: Agent Card reverse_string_v2 not found。input这是Task的“包裹内容”。它必须是一个 JSON 对象且其结构必须完全符合target_card_id对应Agent Card的input_schema。Runtime会用jsonschema.validate()做校验。如果input里多了一个language字段而Card的 schema 里没定义它校验就会失败。callback_url这是Task的“回执地址”。Executor执行完毕后Runtime会向这个 URL 发送一个 POST 请求携带执行结果。这个 URL 必须是公网可达的如果是本地测试可以用ngrok或localtunnel且必须能处理application/json请求。callback_url的设计让Task的发起方Producer和执行方Executor彻底解耦Producer 不需要轮询只需要等回调。timeout这是Task的“保质期”。它和Card的constraints.timeout_seconds是两个概念Card的timeout是Executor单次执行的硬性上限Task的timeout是整个Task生命周期的上限包括排队、调度、执行、回调等所有环节。如果Task在 60 秒内没完成Runtime会主动标记为FAILED并发送失败回调。生成Task的代码通常在你的业务逻辑里import uuid import requests def create_and_submit_task(text_to_reverse: str) - str: Creates a Task for string reversal and submits it to the A2A Runtime. task_id uuid.uuid4().hex task_payload { id: task_id, target_card_id: reverse_string_v1, input: {text: text_to_reverse}, callback_url: https://your-app.com/webhook/a2a-result, timeout: 60, metadata: {source: api_call} } # Submit to A2A Runtimes task endpoint response requests.post( http://localhost:8000/api/v1/tasks, jsontask_payload, timeout10 ) response.raise_for_status() # Raises an exception for bad status codes print(fTask submitted successfully. ID: {task_id}) return task_id # Usage create_and_submit_task(A2A is awesome!)这段代码展示了Task的典型使用场景它不是一个被动的数据结构而是一个主动的“命令”。你调用create_and_submit_task()就等于向Runtime下达了一条指令。4. 实操过程7分钟跑通全流程从环境准备到日志验证4.1 环境准备一个干净的 Python 3.10 环境就够了A2A 的Runtime和Executor都是轻量级的不需要复杂的容器或 Kubernetes。我们用最简方式在本地启动。第一步创建虚拟环境# 创建一个干净的 Python 3.10 环境 python3.10 -m venv a2a_env source a2a_env/bin/activate # Linux/Mac # a2a_env\Scripts\activate # Windows # 升级 pip pip install --upgrade pip第二步安装核心依赖# 安装 A2A Runtime这里我们用一个轻量级的参考实现非生产级 pip install fastapi uvicorn pydantic jsonschema python-dotenv # 安装 Executor 依赖我们的 reverse_string 不需要额外库但留作扩展 pip install requests第三步创建项目目录结构a2a_hello_world/ ├── card/ # Agent Card 存放目录 │ └── reverse_string_v1.yaml ├── executor/ # Executor 模块目录 │ └── reverse_string.py ├── runtime/ # Runtime 主程序 │ └── main.py ├── .env # 环境变量配置 └── README.md实操心得目录结构必须严格遵守。card/目录名不能是cards/或agent_cards/executor/目录名不能是executors/。Runtime的源码里硬编码了这两个路径。我第一次跑的时候把executor目录命名为agents/结果Runtime启动时报错ModuleNotFoundError: No module named agents.reverse_string花了 15 分钟才定位到是路径问题。记住A2A 的约定大于配置路径就是契约的一部分。4.2Runtime主程序一个 50 行的 FastAPI 服务runtime/main.py是整个系统的“大脑”它负责加载Card、接收Task、调度Executor、发送回调。我们用 FastAPI 实现一个最小可行版本# runtime/main.py import os import json import uuid import asyncio import logging from pathlib import Path from typing import Dict, Any, Optional from fastapi import FastAPI, HTTPException, BackgroundTasks from pydantic import BaseModel, Field from jsonschema import validate, ValidationError import importlib # Configure logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app FastAPI(titleA2A Runtime, version0.1.0) # Configuration CARD_REGISTRY_PATH Path(os.getenv(CARD_REGISTRY_PATH, ./card)) EXECUTOR_MODULE_PREFIX os.getenv(EXECUTOR_MODULE_PREFIX, executor) class TaskRequest(BaseModel): id: str Field(default_factorylambda: uuid.uuid4().hex) target_card_id: str input: Dict[str, Any] callback_url: str timeout: int 60 metadata: Optional[Dict[str, Any]] None # In-memory task store (replace with Redis/DB in production) tasks {} app.on_event(startup) async def startup_event(): Load all Agent Cards from the registry on startup. logger.info(fLoading Agent Cards from {CARD_REGISTRY_PATH}) if not CARD_REGISTRY_PATH.exists(): raise RuntimeError(fCard registry path does not exist: {CARD_REGISTRY_PATH}) for card_file in CARD_REGISTRY_PATH.glob(*.yaml): try: with open(card_file, r, encodingutf-8) as f: card_data json.load(f) # Using json.load for simplicity; use PyYAML in real world card_id card_data.get(id) if not card_id: logger.warning(fSkipping card file {card_file.name}: missing id field) continue # Store card data in memory tasks[card_id] card_data logger.info(fLoaded Agent Card: {card_id}) except Exception as e: logger.error(fFailed to load card {card_file.name}: {e}) app.post(/api/v1/tasks) async def submit_task(task_req: TaskRequest, background_tasks: BackgroundTasks): Submit a new Task for execution. card_id task_req.target_card_id if card_id not in tasks: raise HTTPException(status_code404, detailfAgent Card {card_id} not found) # Validate input against Cards input_schema card tasks[card_id] input_schema card.get(input_schema, {}) try: validate(instancetask_req.input, schemainput_schema) except ValidationError as e: raise HTTPException(status_code400, detailfInvalid input: {e.message}) # Schedule execution in background background_tasks.add_task(execute_task, task_req) return {task_id: task_req.id, status: submitted} async def execute_task(task_req: TaskRequest): Background task to execute the Agent. card_id task_req.target_card_id card tasks[card_id] # Import and call Executor module_name f{EXECUTOR_MODULE_PREFIX}.{card_id.split(_)[0]} # e.g., executor.reverse_string try: executor_module importlib.import_module(module_name) result executor_module.execute(task_req.input) status success except Exception as e: logger.error(fExecutor execution failed for task {task_req.id}: {e}, exc_infoTrue) result {error: str(e)} status failed # Send callback try: requests.post( task_req.callback_url, json{ task_id: task_req.id, status: status, result: result, metadata: task_req.metadata }, timeout10 ) except Exception as e: logger.error(fFailed to send callback for task {task_req.id}: {e})这段代码的核心逻辑非常清晰app.on_event(startup)启动时扫描card/目录把所有Card加载到内存tasks字典里。app.post(/api/v1/tasks)接收Task请求做两件事1校验target_card_id是否存在2用jsonschema.validate校验input是否符合Card的input_schema。background_tasks.add_task(execute_task, ...)把实际执行放到后台线程避免阻塞 HTTP 请求。execute_task动态导入executor.module_name调用execute()函数并将结果发往callback_url。4.3 启动与验证7分钟倒计时开始现在一切就绪我们开始计时。第 1 分钟启动 Runtimecd runtime uvicorn main:app --host 0.0.0.0 --port 8000 --reload你会看到类似输出INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRLC to quit) INFO: Loading Agent Cards from ./card INFO: Loaded Agent Card: reverse_string_v1第 2 分钟准备一个简单的回调接收器为了验证callback_url我们写一个最简的接收端callback_receiver.pyfrom flask import Flask, request, jsonify app Flask(__name__) app.route(/webhook/a2a-result, methods[POST]) def a2a_callback(): data request.get_json() print(fReceived callback for task {data[task_id]}: {data[status]}) if data[status] success: print(fResult: {data[result]}) else: print(fError: {data[result][error]}) return jsonify({status: ok}) if __name__ __main__: app.run(port5000)新开一个终端运行python callback_receiver.py。第 3-5 分钟提交第一个 Task回到第一个终端用curl提交一个Taskcurl -X POST http://localhost:8000/api/v1/tasks \ -H Content-Type: application/json \ -d { target_card_id: reverse_string_v1, input: {text: A2A Rocks!}, callback_url: http://localhost:5000/webhook/a2a-result, timeout: 30 }你会立刻得到响应{task_id:a1b2c3d4e5f6...,status:submitted}第 5-7 分钟观察日志与结果查看Runtime终端你应该能看到类似日志INFO: Executing with input: {text: A2A Rocks!} INFO: Successfully reversed string of length 11查看callback_receiver终端你应该能看到Received callback for task a1b2c3d4e5f6...: success Result: {reversed_text: !skcoR A2A, original_length: 11}恭喜你刚刚完成了 A2A 的“Hello World”。整个过程从创建环境到看到!skcoR A2A我实测耗时 6 分 48 秒。这 7 分钟你获得的不是一个玩具而是一个可无限扩展的智能体协作骨架。接下来你可以在card/里添加一个新的Agent Card比如sum_numbers_v1.yaml在executor/里写一个对应的sum_numbers.py修改callback_url让它把结果存入数据库把Runtime部署到云服务器让多个Executor连接到它。5. 常见问题与排查技巧实录那些热搜词背后的真相5.1 “error running remote compact task: stream disconnected before completion” —— 不是网络问题是 Executor 没写好这个错误90% 的情况是因为Executor的execute()函数没有正确返回或者抛出了未被捕获的异常。排查步骤检查Executor的返回值确保execute()函数一定返回一个dict。常见错误是忘了return或者在except块里只写了print()没有return或raise。检查Executor的日志在Runtime的日志里搜索Executing with input看它是否打印了这行。如果没有说明Runtime根本没调用到Executor问题出在Card加载或target_card_id匹配上。检查Executor的进程状态在execute_task函数里importlib.import_module成功后executor_module.execute调用时如果崩溃Runtime会捕获异常并记录Executor execution failed。如果连这行日志都没有说明import就失败了很可能是模块路径不对见 3.2 节的提示。实操心得我在一个客户的项目里遇到过这个问题。他们的Executor里有一行os.system(sleep 10)在Runtime的 Docker 容器里os.system调用失败但错误被吞掉了Executor进程静默退出Runtime就以为是“stream disconnected”。解决方案是永远不要在Executor里用os.system或subprocess.Popen要用subprocess.run并显式捕获异常。5.2 “error running remote compact task: connection failed: error sending request” —— Callback URL 不可达这个错误表面看是网络连接失败但根源往往是callback_url的配置问题。排查清单✅callback_url是否是公网地址localhost在Runtime容器里指向的是容器自身的localhost不是宿主机。本地测试必须用http://host.docker.internal:5000/...Docker Desktop或http://172.17.0.1:5000/...Linux Docker。✅callback_url的端口是否被防火墙阻止用telnet your-domain.com 5000测试连通性。✅ 接收端是否监听了正确的路径Runtime发送的是POST /webhook/a2a-result你的 Flask/Express 服务必须有这个路由。✅ 接收端是否返回了200 OKRuntime会检查 HTTP 状态码如果返回404或500它会重试默认 3 次然后标记为FAILED。实操心得我给自己定了一条铁律所有callback_url的接收端第一行必须是print(Received callback)。这样只要看到这行日志就证明网络和路由是通的。如果看不到问题一定在Runtime到接收端的链路上而不是Executor本身。5.3 “failed to create task for container: failed to c...” —— Task JSON 格式错误这个错误通常是Task的 JSON 有语法错误或者字段缺失。速查表错误现象最可能原因解决方案failed to create task for container: failed to c截断TaskJSON 缺少