1. 从三个 Agent 各自为战说起A2A 协议到底解决什么问题如果你已经用 LangGraph 搭过单 Agent大概率经历过这个阶段一个 ReAct 循环加几个 Tool跑得挺顺。但当业务要求变成数据分析 Agent 出完结论后自动触发报告 Agent 生成 PDF再交给邮件 Agent 发送事情就开始失控了。三个 Agent 可能分别用 LangGraph、裸 Python 脚本、甚至第三方 SaaS 实现它们之间没有共同语言你只能写一堆 HTTP 胶水代码硬拧在一起。A2A 协议Agent-to-Agent Protocol就是为这个场景设计的通信标准。它不要求你换框架也不绑定某个厂商核心只做一件事约定 Agent 之间怎么交换任务和结果。技术底座非常务实——HTTP/1.1 或 HTTP/2 做传输SSEServer-Sent Events做流式推送JSON-RPC 2.0 做消息格式。这意味着任何支持 HTTP 的 Agent无论用什么语言写的都能接入。它适合谁正在做多 Agent 协作、需要跨团队或跨系统编排异构 Agent 的开发者。如果你只是单 Agent 调工具A2A 暂时用不上但一旦 Agent 数量从 1 变成 N通信协议就是决定你是分布式智能还是分布式混乱的关键。这篇会给出可复制的 Agent Card 配置骨架、消息端点配置以及在 LangGraph 里做跨 Agent 任务委派和响应验证的完整动作。2. 前置准备TaoToken 接入与 LangGraph 环境在动手写 A2A 通信之前先把模型调用这条链路打通。多 Agent 协作里每个 Agent 都要调模型如果每个 Agent 各自维护一套 Key 和计费排查问题会非常痛苦。我习惯用 TaoToken 统一管理模型接入它的 API 兼容 OpenAI 格式LangGraph 里的 ChatModel 可以直接指向它。官网入口在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台创建 API Key。API 基地址是 https://taotoken.net/api 注意这个地址不带 UTM 参数配置时直接用。环境依赖装这几个就够pip install langgraph langchain-openai fastapi uvicorn httpx sse-starlette pydantic模型客户端配置如下把 Key 换成你在控制台生成的那串import os from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-4o-mini, base_urlhttps://taotoken.net/api, api_keyos.environ[TAOTOKEN_API_KEY], temperature0 )这里有个细节base_url末尾不要加/v1TaoToken 的兼容层会自动处理路径。如果你在 LangGraph 的多个节点里都要用模型建议把llm做成单例注入避免每个 Agent 重复初始化连接池。提示多 Agent 场景下建议给每个 Agent 分配独立的 API Key 或至少独立的调用标签方便在控制台按 Agent 维度看用量。Key 管理入口在 https://taotoken.net/console/api-keys 。3. Agent Card 与消息端点配置骨架A2A 协议里每个 Agent 都要暴露一个 Agent Card放在/.well-known/agent.json路径下。它相当于 Agent 的身份证加能力广告Client Agent 通过标准 HTTP GET 就能拿到里面写清楚我是谁、我会什么、怎么找我、怎么认证。先定义 Pydantic 模型保证 Card 结构可校验from pydantic import BaseModel, Field from typing import List, Optional class Skill(BaseModel): id: str name: str description: str input_modes: List[str] Field(default_factorylambda: [text]) output_modes: List[str] Field(default_factorylambda: [text]) class AgentCard(BaseModel): name: str description: str url: str version: str 1.0.0 capabilities: dict Field(default_factorylambda: {streaming: True}) skills: List[Skill] authentication: Optional[dict] None一个报表生成 Agent 的 Card 实例report_card AgentCard( name报表生成助手, description接收结构化数据生成 PDF 报表并返回下载链接, urlhttp://127.0.0.1:8001, capabilities{streaming: True, pushNotifications: False}, skills[ Skill( idgenerate_pdf_report, name生成 PDF 报表, description输入 JSON 数据输出 PDF 文件 Artifact, input_modes[text, file], output_modes[file] ) ], authentication{schemes: [Bearer]} )消息端点用 FastAPI 实现核心是 JSON-RPC 2.0 格式的tasks/send方法。下面是一个最小可运行的 Remote Agent 骨架from fastapi import FastAPI, Request from fastapi.responses import JSONResponse import uuid, asyncio app FastAPI() TASKS {} app.get(/.well-known/agent.json) async def agent_card(): return report_card.model_dump() app.post(/a2a) async def a2a_endpoint(request: Request): body await request.json() method body.get(method) params body.get(params, {}) req_id body.get(id) if method tasks/send: task_id str(uuid.uuid4()) TASKS[task_id] {state: submitted, artifact: None} asyncio.create_task(run_task(task_id, params)) return JSONResponse({ jsonrpc: 2.0, id: req_id, result: {id: task_id, state: submitted} }) if method tasks/get: task TASKS.get(params[id]) return JSONResponse({ jsonrpc: 2.0, id: req_id, result: task }) return JSONResponse({ jsonrpc: 2.0, id: req_id, error: {code: -32601, message: Method not found} }) async def run_task(task_id: str, params: dict): TASKS[task_id][state] working await asyncio.sleep(2) # 模拟耗时处理 TASKS[task_id][state] completed TASKS[task_id][artifact] { parts: [{type: text, text: f报表已生成: report_{task_id[:8]}.pdf}] }启动命令uvicorn remote_agent:app --host 127.0.0.1 --port 8001这套骨架的关键点在于任务提交后立即返回task_id而不是同步等结果。这是 A2A 和普通 REST 接口的本质区别——它把 Agent 通信从函数调用升级成了任务邮局。4. LangGraph 中做跨 Agent 委派与响应验证现在把 Remote Agent 接进 LangGraph。核心思路是把 A2A 调用封装成一个独立节点Node 内部处理 Task 创建、状态轮询、异常收敛对外只暴露输入 State、输出 State的契约。先写 A2A 客户端import httpx, asyncio class A2AClient: def __init__(self, base_url: str, token: str None): self.base_url base_url.rstrip(/) self.headers {Content-Type: application/json} if token: self.headers[Authorization] fBearer {token} async def discover(self) - dict: async with httpx.AsyncClient() as c: r await c.get(f{self.base_url}/.well-known/agent.json) r.raise_for_status() return r.json() async def send_task(self, text: str) - str: payload { jsonrpc: 2.0, id: 1, method: tasks/send, params: {message: {role: user, parts: [{type: text, text: text}]}} } async with httpx.AsyncClient() as c: r await c.post(f{self.base_url}/a2a, jsonpayload, headersself.headers, timeout30) r.raise_for_status() return r.json()[result][id] async def wait_for_completion(self, task_id: str, timeout: float 60.0) - dict: deadline asyncio.get_event_loop().time() timeout while asyncio.get_event_loop().time() deadline: payload {jsonrpc: 2.0, id: 2, method: tasks/get, params: {id: task_id}} async with httpx.AsyncClient() as c: r await c.post(f{self.base_url}/a2a, jsonpayload, headersself.headers, timeout10) task r.json()[result] if task[state] in (completed, failed, canceled): return task await asyncio.sleep(1) raise TimeoutError(f任务 {task_id} 超时未完成)接入 LangGraph 的 State 定义和远程节点from typing import TypedDict, List from langgraph.graph import StateGraph, END class GraphState(TypedDict): topic: str research: str report_url: str status: str errors: List[dict] async def report_node(state: GraphState) - GraphState: client A2AClient(http://127.0.0.1:8001) new_state dict(state) try: card await client.discover() new_state[status] f已发现 Agent: {card[name]} task_id await client.send_task(state[research]) result await client.wait_for_completion(task_id) if result[state] completed: new_state[report_url] result[artifact][parts][0][text] new_state[status] 报表生成完成 else: new_state[errors] state.get(errors, []) [ {node: report_node, task_id: task_id, reason: result.get(error, unknown)} ] except Exception as e: new_state[errors] state.get(errors, []) [ {node: report_node, reason: str(e)} ] return new_state组装图并执行验证builder StateGraph(GraphState) builder.add_node(report, report_node) builder.set_entry_point(report) builder.add_edge(report, END) graph builder.compile() async def main(): result await graph.ainvoke({ topic: Q3 销售分析, research: Q3 销售额 1200 万同比增长 15%, report_url: , status: , errors: [] }) print(状态:, result[status]) print(报表:, result[report_url]) print(错误:, result[errors]) asyncio.run(main())预期输出状态: 报表生成完成 报表: 报表已生成: report_a1b2c3d4.pdf 错误: []到这里跨 Agent 委派和响应验证就跑通了。你可以把report_node换成任意远程 Agent只要对方暴露了符合 A2A 的 Agent Card 和tasks/send端点图里就能直接编排。5. 本篇常见错误排查Agent Card 返回 404检查路径是否严格是/.well-known/agent.json注意well-known前面有个点。FastAPI 路由里写成/.well-known/agent.json而不是/well-known/agent.json。tasks/send 返回 JSON-RPC 错误码 -32601说明 method 名称不匹配。A2A 标准方法名是tasks/send和tasks/get不要写成task.send或sendTask。任务一直卡在 working 状态先确认 Remote Agent 的run_task是否真的被调度了。用asyncio.create_task时如果事件循环被阻塞任务不会执行。另外检查wait_for_completion的超时时间是否太短。LangGraph 节点报 TimeoutError默认 60 秒对复杂任务不够。把wait_for_completion的 timeout 调大或者在 Remote Agent 侧实现 SSE 流式推送Client 侧改用subscribe_task监听避免轮询。认证失败 401检查 Agent Card 里声明的authentication.schemes和 Client 实际发送的 Header 是否一致。如果 Card 写的是 BearerClient 就必须带Authorization: Bearer token。跨 Agent 数据格式对不上Remote Agent 返回的 Artifact 结构和本地 State Schema 往往不一致。建议在 Node 里加一层 Adapter显式做字段映射不要直接硬塞进 State。6. 继续深入的方向跑通上面的骨架后下一步可以往三个方向走。一是把轮询改成 SSE 流式监听Remote Agent 用sse-starlette推送状态变更Client 侧实时消费长任务体验会好很多。二是引入 OAuth2 或 mTLS把 Agent 之间的认证从简单 Bearer 升级到短期 Token 加 Scope 细分。三是把多个 Remote Agent 编排成串行或并行图比如研究 Agent 完成后同时触发报表 Agent 和通知 Agent。模型调用这块如果你要在多个 Agent 节点里共享配置可以直接用 TaoToken 的模型对话入口 https://taotoken.net/models 做联调确认模型响应正常后再接进图里。长期跑编码类 Agent 的话Coding Plan https://taotoken.net/coding-plan 在用量上会更划算。接入文档在 https://taotoken.net/doc 里面有完整的 API 参数说明和错误码对照。A2A 的价值不在于协议本身多复杂而在于它让异构 Agent 之间的协作有了共同语言。你不需要一次性把所有 Agent 都改造成 A2A从最关键的那条跨系统链路开始把 Agent Card 和 Task 状态机跑通剩下的就是复制粘贴的事了。