Agent 到 Agent（A2A）协议

A2A 代理委派

Info

要将 A2A 代理部署到生产环境？请参见 AMP 上的 A2A，了解分布式状态、企业级认证、gRPC 传输和横向扩展。

CrewAI 将 A2A 协议 视为一等的委派原语，使代理能够委派任务、请求信息、与远程代理协作，也可以作为符合 A2A 标准的服务器代理运行。 在客户端模式下，代理会根据任务需求自主决定本地执行还是远程委派。

工作原理

当代理配置了 A2A 能力时：

1. 代理分析每个任务
2. 它会决定：
   • 直接利用自身能力处理任务
   • 委派给远程 A2A 代理进行专门处理
3. 如果进行委派，代理会通过协议与远程 A2A 代理通信
4. 结果会返回到 CrewAI 工作流

Note

A2A 委派需要 a2a-sdk 包。可使用 uv add 'crewai[a2a]' 或 pip install 'crewai[a2a]' 安装。

基础配置

Warning

crewai.a2a.config.A2AConfig 已弃用，并将在 v2.0.0 中移除。连接远程代理请使用 A2AClientConfig，将代理暴露为服务器请使用 A2AServerConfig。

通过设置 a2a 参数来为代理配置 A2A 委派：

from crewai import Agent, Crew, Task
from crewai.a2a import A2AClientConfig

agent = Agent(
    role="Research Coordinator",
    goal="Coordinate research tasks efficiently",
    backstory="Expert at delegating to specialized research agents",
    llm="gpt-4o",
    a2a=A2AClientConfig(
        endpoint="https://example.com/.well-known/agent-card.json",
        timeout=120,
        max_turns=10
    )
)

task = Task(
    description="Research the latest developments in quantum computing",
    expected_output="A comprehensive research report",
    agent=agent
)

crew = Crew(agents=[agent], tasks=[task], verbose=True)
result = crew.kickoff()

客户端配置选项

A2AClientConfig 类接受以下参数：

endpoint str required

A2A 代理的端点 URL（通常指向 .well-known/agent-card.json）

auth AuthScheme default: None

A2A 代理的认证方案。支持 Bearer token、OAuth2、API key 和 HTTP 认证。

timeout int default: 120

请求超时时间（秒）

max_turns int default: 10

与 A2A 代理进行对话的最大轮数

response_model type[BaseModel] default: None

可选的 Pydantic 模型，用于向 A2A 代理请求结构化输出。A2A 协议并不强制这一点，因此 A2A 代理不一定需要遵守该请求。

fail_fast bool default: True

如果代理连接失败，是否立即抛出错误。设为 False 时，代理会继续使用可用代理，并向 LLM 告知哪些代理不可用。

trust_remote_completion_status bool default: False

当为 True 时，在远程 A2A 代理报告完成时直接返回其结果。当为 False 时，允许服务器代理审查结果并可能继续对话。

updates UpdateConfig default: StreamingConfig()

接收任务状态的更新机制。可选：StreamingConfig、PollingConfig 或 PushNotificationConfig。

accepted_output_modes list[str] default: ["application/json"]

客户端可以在响应中接受的媒体类型。

extensions list[str] default: []

客户端支持的 A2A 协议扩展 URI。

client_extensions list[A2AExtension] default: []

客户端侧的处理钩子，用于工具注入、提示词增强和响应修改。

transport ClientTransportConfig default: ClientTransportConfig()

传输配置，包括首选传输、用于协商的支持传输，以及协议特定设置（gRPC 消息大小、keepalive 等）。

transport_protocol Literal['JSONRPC', 'GRPC', 'HTTP+JSON'] default: None

已弃用：请改用 transport=ClientTransportConfig(preferred=...)。

supported_transports list[str] default: None

已弃用：请改用 transport=ClientTransportConfig(supported=...)。

认证

对于需要认证的 A2A 代理，请使用以下任一认证方案：

Bearer Token

from crewai.a2a import A2AClientConfig
from crewai.a2a.auth import BearerTokenAuth

agent = Agent(
    role="Secure Coordinator",
    goal="Coordinate tasks with secured agents",
    backstory="Manages secure agent communications",
    llm="gpt-4o",
    a2a=A2AClientConfig(
        endpoint="https://secure-agent.example.com/.well-known/agent-card.json",
        auth=BearerTokenAuth(token="your-bearer-token"),
        timeout=120
    )
)

API Key

from crewai.a2a import A2AClientConfig
from crewai.a2a.auth import APIKeyAuth

agent = Agent(
    role="API Coordinator",
    goal="Coordinate with API-based agents",
    backstory="Manages API-authenticated communications",
    llm="gpt-4o",
    a2a=A2AClientConfig(
        endpoint="https://api-agent.example.com/.well-known/agent-card.json",
        auth=APIKeyAuth(
            api_key="your-api-key",
            location="header",  # or "query" or "cookie"
            name="X-API-Key"
        ),
        timeout=120
    )
)

OAuth2

from crewai.a2a import A2AClientConfig
from crewai.a2a.auth import OAuth2ClientCredentials

agent = Agent(
    role="OAuth Coordinator",
    goal="Coordinate with OAuth-secured agents",
    backstory="Manages OAuth-authenticated communications",
    llm="gpt-4o",
    a2a=A2AClientConfig(
        endpoint="https://oauth-agent.example.com/.well-known/agent-card.json",
        auth=OAuth2ClientCredentials(
            token_url="https://auth.example.com/oauth/token",
            client_id="your-client-id",
            client_secret="your-client-secret",
            scopes=["read", "write"]
        ),
        timeout=120
    )
)

HTTP Basic

from crewai.a2a import A2AClientConfig
from crewai.a2a.auth import HTTPBasicAuth

agent = Agent(
    role="Basic Auth Coordinator",
    goal="Coordinate with basic auth agents",
    backstory="Manages basic authentication communications",
    llm="gpt-4o",
    a2a=A2AClientConfig(
        endpoint="https://basic-agent.example.com/.well-known/agent-card.json",
        auth=HTTPBasicAuth(
            username="your-username",
            password="your-password"
        ),
        timeout=120
    )
)

多个 A2A 代理

通过传入列表来配置多个 A2A 代理进行委派：

from crewai.a2a import A2AClientConfig
from crewai.a2a.auth import BearerTokenAuth

agent = Agent(
    role="Multi-Agent Coordinator",
    goal="Coordinate with multiple specialized agents",
    backstory="Expert at delegating to the right specialist",
    llm="gpt-4o",
    a2a=[
        A2AClientConfig(
            endpoint="https://research.example.com/.well-known/agent-card.json",
            timeout=120
        ),
        A2AClientConfig(
            endpoint="https://data.example.com/.well-known/agent-card.json",
            auth=BearerTokenAuth(token="data-token"),
            timeout=90
        )
    ]
)

LLM 会根据任务需求自动选择要委派给哪个 A2A 代理。

错误处理

使用 fail_fast 参数控制如何处理代理连接失败：

from crewai.a2a import A2AClientConfig

# Fail immediately on connection errors (default)
agent = Agent(
    role="Research Coordinator",
    goal="Coordinate research tasks",
    backstory="Expert at delegation",
    llm="gpt-4o",
    a2a=A2AClientConfig(
        endpoint="https://research.example.com/.well-known/agent-card.json",
        fail_fast=True
    )
)

# Continue with available agents
agent = Agent(
    role="Multi-Agent Coordinator",
    goal="Coordinate with multiple agents",
    backstory="Expert at working with available resources",
    llm="gpt-4o",
    a2a=[
        A2AClientConfig(
            endpoint="https://primary.example.com/.well-known/agent-card.json",
            fail_fast=False
        ),
        A2AClientConfig(
            endpoint="https://backup.example.com/.well-known/agent-card.json",
            fail_fast=False
        )
    ]
)

当 fail_fast=False 时：

• 如果部分代理失败，LLM 会被告知哪些代理不可用，并可将任务委派给仍可工作的代理
• 如果所有代理都失败，LLM 会收到关于不可用代理的提示，并直接处理该任务
• 连接错误会被捕获并包含在上下文中，以便更好地决策

更新机制

控制你的代理如何从远程 A2A 代理接收任务状态更新：

流式（默认）

from crewai.a2a import A2AClientConfig
from crewai.a2a.updates import StreamingConfig

agent = Agent(
    role="Research Coordinator",
    goal="Coordinate research tasks",
    backstory="Expert at delegation",
    llm="gpt-4o",
    a2a=A2AClientConfig(
        endpoint="https://research.example.com/.well-known/agent-card.json",
        updates=StreamingConfig()
    )
)

轮询

from crewai.a2a import A2AClientConfig
from crewai.a2a.updates import PollingConfig

agent = Agent(
    role="Research Coordinator",
    goal="Coordinate research tasks",
    backstory="Expert at delegation",
    llm="gpt-4o",
    a2a=A2AClientConfig(
        endpoint="https://research.example.com/.well-known/agent-card.json",
        updates=PollingConfig(
            interval=2.0,
            timeout=300.0,
            max_polls=100
        )
    )
)

推送通知

from crewai.a2a import A2AClientConfig
from crewai.a2a.updates import PushNotificationConfig

agent = Agent(
    role="Research Coordinator",
    goal="Coordinate research tasks",
    backstory="Expert at delegation",
    llm="gpt-4o",
    a2a=A2AClientConfig(
        endpoint="https://research.example.com/.well-known/agent-card.json",
        updates=PushNotificationConfig(
            url="{base_url}/a2a/callback",
            token="your-validation-token",
            timeout=300.0
        )
    )
)

将代理暴露为 A2A 服务器

你可以将 CrewAI 代理暴露为符合 A2A 标准的服务器，让其他 A2A 客户端将任务委派给它们。

服务器配置

为你的代理添加 A2AServerConfig 即可启用服务器能力：

from crewai import Agent
from crewai.a2a import A2AServerConfig

agent = Agent(
    role="Data Analyst",
    goal="Analyze datasets and provide insights",
    backstory="Expert data scientist with statistical analysis skills",
    llm="gpt-4o",
    a2a=A2AServerConfig(url="https://your-server.com")
)

服务器配置选项

name str default: None

代理的可读名称。如果未提供，默认为代理的 role。

description str default: None

可读描述。如果未提供，默认为代理的 goal 和 backstory。

version str default: 1.0.0

代理卡片的版本字符串。

skills list[AgentSkill] default: []

代理技能列表。如果未提供，则会根据代理工具自动生成。

capabilities AgentCapabilities default: AgentCapabilities(streaming=True, push_notifications=False)

代理支持的可选能力声明。

default_input_modes list[str] default: ["text/plain", "application/json"]

支持的输入 MIME 类型。

default_output_modes list[str] default: ["text/plain", "application/json"]

支持的输出 MIME 类型。

url str default: None

首选端点 URL。如果设置，将覆盖传给 to_agent_card() 的 URL。

protocol_version str default: 0.3.0

此代理支持的 A2A 协议版本。

provider AgentProvider default: None

有关该代理服务提供方的信息。

documentation_url str default: None

该代理文档的 URL。

icon_url str default: None

该代理图标的 URL。

additional_interfaces list[AgentInterface] default: []

其他受支持的接口（传输方式和 URL 组合）。

security list[dict[str, list[str]]] default: []

所有代理交互所需的安全要求对象。

security_schemes dict[str, SecurityScheme] default: {}

可用于授权请求的安全方案。

supports_authenticated_extended_card bool default: False

该代理是否向已认证用户提供扩展卡片。

extended_skills list[AgentSkill] default: []

仅对已认证用户可见的扩展代理卡片附加技能。

signing_config AgentCardSigningConfig default: None

使用 JWS 对 AgentCard 进行签名的配置。支持 RS256、ES256、PS256 及相关算法。

server_extensions list[ServerExtension] default: []

带有 on_request/on_response 钩子的服务器端 A2A 协议扩展，可修改代理行为。

push_notifications ServerPushNotificationConfig default: None

发出推送通知的配置，包括 HMAC-SHA256 签名密钥。

transport ServerTransportConfig default: ServerTransportConfig()

传输配置，包括首选传输、gRPC 服务器设置、JSON-RPC 路径和 HTTP+JSON 设置。

auth ServerAuthScheme default: None

接收 A2A 请求的认证方案。默认使用 AUTH_TOKEN 环境变量的 SimpleTokenAuth。

preferred_transport Literal['JSONRPC', 'GRPC', 'HTTP+JSON'] default: None

已弃用：请改用 transport=ServerTransportConfig(preferred=...)。

signatures list[AgentCardSignature] default: None

已弃用：请改用 signing_config=AgentCardSigningConfig(...)。

组合客户端和服务器

代理可以同时充当客户端和服务器，只需同时提供两种配置：

from crewai import Agent
from crewai.a2a import A2AClientConfig, A2AServerConfig

agent = Agent(
    role="Research Coordinator",
    goal="Coordinate research and serve analysis requests",
    backstory="Expert at delegation and analysis",
    llm="gpt-4o",
    a2a=[
        A2AClientConfig(
            endpoint="https://specialist.example.com/.well-known/agent-card.json",
            timeout=120
        ),
        A2AServerConfig(url="https://your-server.com")
    ]
)

文件输入和结构化输出

A2A 支持双向传递文件和请求结构化输出。

客户端侧：当向远程 A2A 代理委派时，任务 input_files 中的文件会作为 FilePart 随外发消息发送。如果在 A2AClientConfig 上设置了 response_model，Pydantic 模型的 JSON schema 会嵌入到消息元数据中，请求远程代理返回结构化输出。

服务端侧：传入的 FilePart 会被提取并作为 input_files 传递给代理的任务。如果客户端包含了 JSON schema，服务器会据此创建响应模型并应用到任务上。当代理返回结构化数据时，响应会以 DataPart 而不是纯文本发送回去。

最佳实践

设置合适的超时

根据预期的 A2A 代理响应时间配置超时。耗时更长的任务可能需要更高的超时值。

限制对话轮数

使用 max_turns 防止过多来回交互。代理会在达到上限前自动结束对话。

使用弹性的错误处理

在有多个代理的生产环境中，将 fail_fast 设为 False，以优雅地处理连接失败并保持工作流连续性。

保护你的凭证

将认证 token 和凭证存储为环境变量，而不是写在代码里。

监控委派决策

使用 verbose 模式观察 LLM 何时选择委派，何时直接处理任务。

支持的认证方式

• Bearer Token - 基于 token 的简单认证
• OAuth2 Client Credentials - 用于机器对机器通信的 OAuth2 流程
• OAuth2 Authorization Code - 需要用户授权的 OAuth2 流程
• API Key - 基于密钥的认证（header、query 参数或 cookie）
• HTTP Basic - 用户名 / 密码认证
• HTTP Digest - 摘要认证（需要 httpx-auth 包）

了解更多

如需了解更多关于 A2A 协议和参考实现的信息：

• A2A 协议文档
• A2A 示例实现
• A2A Python SDK