CrewAI A2A生产部署

读完这篇你能带走什么
：用 CrewAI 把你的 Agent 接入 A2A 协议，加上认证和容错，直接部署到生产环境。
适用场景：多 Agent 协作系统上线 | 不适用：单 Agent + MCP 工具调用

上个月我们团队把一个单体 Agent 拆成 3 个专业 Agent，用 A2A 协议串联。上线第一天就踩了两个坑：远程 Agent 超时导致整个链路卡死，废弃 API 写法在 v2.0 直接报错。

本文把我们踩过的坑整理成 4 段可直接用的代码，从最基本的 A2A 客户端到带认证的生产级部署。

A2A 是什么，为什么值得用

先说结论：A2A 解决的问题是——让不同框架、不同机器、不同组织写的 Agent 能互相协作。

协议	管什么	典型场景
MCP	Agent ↔ 工具	Agent 调数据库、调 API
A2A	Agent ↔ Agent	多个 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：

python
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 种认证方式：

python
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 硬编码在代码里，用环境变量：

bash
export ANALYST_TOKEN="prod-token-xxx"

python
import os

auth=BearerTokenAuth(
token=os.environ["ANALYST_TOKEN"]
)

第三段代码：容错与主备切换

这是上线前最容易忽略的环节。我们踩的坑：主 Agent 宕机，整个任务链卡死 10 分钟才发现。

解决方案就一个参数：

python
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 的行为逻辑：

场景	行为
主 Agent 正常	直接委托给主 Agent
主 Agent 挂了	LLM 自动切到备用 Agent
全部挂了	LLM 被告知全部不可用，自行处理
部分挂了	LLM 只用可用的 Agent

这是 CrewAI A2A 相比其他框架最大的优势——不需要写额外的熔断逻辑，框架层面就帮你处理了。

第四段代码：发布自己的 Agent Card

前面的代码都是调用别人的 Agent。如果你要让自己的 Agent 被别人调用，需要发布 Agent Card：

python
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，内容类似：

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 会直接报错：

废弃写法	正确写法
`A2AConfig(...)`	客户端用 `A2AClientConfig`，服务端用 `A2AServerConfig`
`transport_protocol="GRPC"`	`transport=ClientTransportConfig(preferred="GRPC")`
`signatures=[...]`	`signing_config=AgentCardSigningConfig(...)`

如果你从旧教程抄代码，大概率会踩这个坑。

超时建议

场景	建议 timeout	建议 max_turns
简单查询（搜索/计算）	30-60 秒	3-5 轮
数据分析	120 秒	5-10 轮
代码生成/审查	180-300 秒	10-15 轮

max_turns 防止无限循环非常重要。我们线上有个 Agent 因为 max_turns 没设，跟远程 Agent 来回了 47 轮才超时。

任务状态更新

默认用流式接收结果，适合大多数场景。长任务（超过 2 分钟）建议用轮询模式：

python
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 就是这样一行代码——简单，但救命。