夜雨聆风读完这篇你能带走什么：用 CrewAI 把你的 Agent 接入 A2A 协议，加上认证和容错，直接部署到生产环境。
适用场景：多 Agent 协作系统上线 | 不适用：单 Agent + MCP 工具调用
上个月我们团队把一个单体 Agent 拆成 3 个专业 Agent，用 A2A 协议串联。上线第一天就踩了两个坑：远程 Agent 超时导致整个链路卡死，废弃 API 写法在 v2.0 直接报错。
本文把我们踩过的坑整理成 4 段可直接用的代码，从最基本的 A2A 客户端到带认证的生产级部署。
A2A 是什么，为什么值得用
先说结论：A2A 解决的问题是——让不同框架、不同机器、不同组织写的 Agent 能互相协作。
一个常见误解是把 MCP 当成多 Agent 方案。MCP 是工具协议，一个 Agent 调多个工具。A2A 是 Agent 间协议，多个 Agent 各有各的能力，互相委托任务。
2026 年 4 月，所有主流框架——CrewAI、LangGraph、OpenAI Agents SDK、Google ADK——都已原生支持 A2A。这意味着你不用自己造轮子，直接用框架的 API 就行。
我们选 CrewAI 是因为三个原因：
- API 最简洁
：3 行代码就能启动 A2A 客户端 - 容错内置
： fail_fast=False一行搞定主备切换 - 认证齐全
：Bearer Token / API Key / OAuth2 / Basic 都有
第一段代码：最简单的 A2A 委托
这是最小可运行示例。一个 Agent 收到任务后，把子任务委托给远程 Agent：
from crewai import Agent, Crew, Task
from crewai.a2a import A2AClientConfig
# 本地 Agent：任务分发器
dispatcher = Agent(
role="任务分发器",
goal="把分析任务分给专业Agent",
backstory="资深数据工程师",
llm="gpt-4o",
a2a=A2AClientConfig(
endpoint=(
"https://analyst.example.com/"
".well-known/agent-card.json"
),
timeout=120,
max_turns=10
)
)
task = Task(
description="分析Q1销售数据趋势",
agent=dispatcher,
expected_output="趋势分析报告"
)
crew = Crew(
agents=[dispatcher],
tasks=[task],
verbose=True
)
result = crew.kickoff()
print(result)
关键点：endpoint 指向远程 Agent 的 .well-known/agent-card.json，这是 A2A 协议的发现机制。远程 Agent 需要先发布自己的 Agent Card（相当于一张名片），其他 Agent 才能找到它。
第二段代码：生产环境必须加的认证
开发环境裸跑没问题，但上生产之前必须加认证。CrewAI 支持 4 种认证方式：
from crewai import Agent
from crewai.a2a import A2AClientConfig
from crewai.a2a.auth import (
BearerTokenAuth,
APIKeyAuth,
OAuth2ClientCredentials,
HTTPBasicAuth
)
# 方式1：Bearer Token（最常用）
agent_bearer = Agent(
role="数据分析",
llm="gpt-4o",
a2a=A2AClientConfig(
endpoint=(
"https://analyst.example.com/"
".well-known/agent-card.json"
),
auth=BearerTokenAuth(
token="prod-token-xxx"
)
)
)
# 方式2：API Key（自定义 Header）
agent_apikey = Agent(
role="数据分析",
llm="gpt-4o",
a2a=A2AClientConfig(
endpoint=(
"https://analyst.example.com/"
".well-known/agent-card.json"
),
auth=APIKeyAuth(
api_key="your-key",
location="header",
name="X-API-Key"
)
)
)
# 方式3：OAuth2（企业环境推荐）
agent_oauth = Agent(
role="数据分析",
llm="gpt-4o",
a2a=A2AClientConfig(
endpoint=(
"https://analyst.example.com/"
".well-known/agent-card.json"
),
auth=OAuth2ClientCredentials(
token_url=(
"https://auth.example.com"
"/oauth/token"
),
client_id="my-client",
client_secret="my-secret",
scopes=["read", "write"]
)
)
)
# 方式4：HTTP Basic（简单内网场景）
agent_basic = Agent(
role="数据分析",
llm="gpt-4o",
a2a=A2AClientConfig(
endpoint=(
"https://analyst.example.com/"
".well-known/agent-card.json"
),
auth=HTTPBasicAuth(
username="agent-user",
password="agent-pass"
)
)
)
我们线上用 OAuth2。注意：绝对不要把 token 硬编码在代码里，用环境变量：
export ANALYST_TOKEN="prod-token-xxx"
import os
auth=BearerTokenAuth(
token=os.environ["ANALYST_TOKEN"]
)
第三段代码：容错与主备切换
这是上线前最容易忽略的环节。我们踩的坑：主 Agent 宕机，整个任务链卡死 10 分钟才发现。
解决方案就一个参数：
from crewai import Agent, Crew, Task
from crewai.a2a import A2AClientConfig
# 主 Agent + 备用 Agent，自动切换
dispatcher = Agent(
role="任务分发器",
goal="分发分析任务",
backstory="资深数据工程师",
llm="gpt-4o",
a2a=[
A2AClientConfig(
endpoint=(
"https://primary.example.com/"
".well-known/agent-card.json"
),
timeout=60,
fail_fast=False # 关键！
),
A2AClientConfig(
endpoint=(
"https://backup.example.com/"
".well-known/agent-card.json"
),
timeout=60,
fail_fast=False
)
]
)
task = Task(
description="分析Q1销售数据",
agent=dispatcher,
expected_output="分析报告"
)
crew = Crew(
agents=[dispatcher],
tasks=[task],
verbose=True
)
fail_fast=False 的行为逻辑：
这是 CrewAI A2A 相比其他框架最大的优势——不需要写额外的熔断逻辑，框架层面就帮你处理了。
第四段代码：发布自己的 Agent Card
前面的代码都是调用别人的 Agent。如果你要让自己的 Agent 被别人调用，需要发布 Agent Card：
from crewai import Agent
from crewai.a2a import A2AServerConfig
analyst = Agent(
role="数据分析师",
goal="分析数据并输出报告",
backstory="专业数据分析师",
llm="gpt-4o",
a2a=A2AServerConfig(
url="https://analyst.your-domain.com"
)
)
Agent Card 会自动发布到 https://analyst.your-domain.com/.well-known/agent-card.json，内容类似：
{
"name": "数据分析师",
"url": "https://analyst.your-domain.com",
"version": "1.0.0",
"skills": [
{
"id": "data-analysis",
"name": "数据分析",
"description": "分析数据并输出报告"
}
]
}
其他团队的 Agent 通过这个 URL 就能发现并调用你的 Agent。
上线前的避坑清单
废弃 API 别用
CrewAI A2A 经历过一次 API 重构，下面这些写法在 v2.0 会直接报错：
如果你从旧教程抄代码，大概率会踩这个坑。
超时建议
max_turns 防止无限循环非常重要。我们线上有个 Agent 因为 max_turns 没设，跟远程 Agent 来回了 47 轮才超时。
任务状态更新
默认用流式接收结果，适合大多数场景。长任务（超过 2 分钟）建议用轮询模式：
from crewai.a2a.config import (
A2AClientConfig,
PollingConfig
)
agent = Agent(
role="数据分析师",
llm="gpt-4o",
a2a=A2AClientConfig(
endpoint=(
"https://analyst.example.com/"
".well-known/agent-card.json"
),
updates=PollingConfig(
interval=2.0,
timeout=300.0
)
)
)
你明天就能做的行动清单
- 装依赖
： pip install 'crewai[a2a]'，跑通第一段代码 - 加认证
：把生产环境的 API Key 通过环境变量注入，别硬编码 - 开容错
：所有 A2A 客户端加 fail_fast=False，配置备用 endpoint - 设超时
：根据任务类型设置合理的 timeout和max_turns，防止无限循环
多 Agent 系统上线不是写完代码就完事——认证、容错、超时，三个缺一个都会在生产环境翻车。我们踩过的坑，希望你别再踩一遍。
好的架构不是没有故障，而是故障发生时系统能自己恢复。 fail_fast=False就是这样一行代码——简单，但救命。
