ag2-a2a

A2A（Agent-to-Agent）集成

通过 A2A 协议公开 AG2 Agent——该协议提供标准的 JSON-RPC / REST / gRPC 契约以及可发现的 AgentCard——让其他系统能够通过网络调用你的 Agent。同一模块也支持 AG2 Agent 使用 A2AConfig 作为其模型配置来调用远程 A2A Agent。

适用场景

• 你希望将 AG2 Agent 作为标准网络服务提供，让非 AG2 A2A 客户端也能调用（实现互操作），而不只是供 Python 调用。
• 你需要一个可发现的 Agent：在 /.well-known/agent-card.json 发布 AgentCard，以描述 Agent 的技能、传输方式、身份验证和能力。
• 你希望能够选择传输方式（JSON-RPC、REST 或 gRPC），也可以让一个服务同时使用多种传输方式。
• 你需要在卡片中声明身份验证方案（Bearer / API 密钥 / OAuth2 / OIDC / mTLS）、推送通知或多租户信息。
• 反过来，如果你希望一个 AG2 Agent 像调用大模型提供商一样调用远程 A2A Agent，请将 A2AConfig 用作调用方 Agent 的 config。

如果你只需要在 Agent 前端使用 Web 界面（React/CopilotKit），请改用 AG-UI 技能。如果两个 AG2 Agent 只需在同一进程或集群内通信，AG2 网络是标准的多 Agent 模式。

安装

pip install "ag2[a2a]"

必需。a2a extra 会安装 a2a-sdk（含 HTTP 服务端）。gRPC 还需要 a2a-sdk[grpc] extra。如果你无法运行命令，请说明确切的 pip install 命令。

公共 API（from ag2.a2a import ...）：A2AConfig、A2AServer、build_card。

60 秒快速上手——通过 A2A 提供 Agent 服务

A2AServer.build_jsonrpc(...) 会返回一个可直接提供服务的 Starlette ASGI 应用。使用 uvicorn 直接运行：

from ag2 import Agent
from ag2.a2a import A2AServer
from ag2.config import OpenAIConfig

agent = Agent(
    name="weather_bot",
    prompt="You answer questions about the weather.",
    config=OpenAIConfig(model="gpt-4o-mini"),
)

server = A2AServer(agent)
# returns a Starlette ASGI app; `url` is the public URL written into the card
app = server.build_jsonrpc(url="http://localhost:8000")

uvicorn serve_a2a:app --host 0.0.0.0 --port 8000

随后，Agent 卡片将在 http://localhost:8000/.well-known/agent-card.json 提供，JSON-RPC 请求则在 / 提供。

如需与其他路由组合，可将返回的应用挂载到父级 Starlette（或 FastAPI）应用下：

from starlette.applications import Starlette
from starlette.routing import Mount

a2a_app = server.build_jsonrpc(url="http://localhost:8000/a2a")
parent = Starlette(routes=[Mount("/a2a", app=a2a_app)])

A2A 没有中间件规范——请自行将 CORS / 身份验证 / 跟踪功能添加到返回的 Starlette 应用。

从另一个 AG2 Agent 调用已提供服务的 Agent

A2AConfig 是一种 ModelConfig：将 Agent 指向远程卡片 URL，然后像调用任何 Agent 一样使用 ask()。客户端会从 {card_url}/.well-known/agent-card.json 获取卡片，选择一种传输方式，然后执行任务。

from ag2 import Agent
from ag2.a2a import A2AConfig

remote = Agent(
    "remote",
    config=A2AConfig(card_url="http://localhost:8000"),
)
reply = await remote.ask("What is the weather in Paris?")
print(await reply.content())

A2AConfig 的关键选项（均为关键字参数，含默认值）：

选项	默认值	用途
card_url	（必需）	基础 URL；卡片从 {card_url}/.well-known/agent-card.json 获取。
prefer	None	当卡片列出多种传输方式时，强制使用其中一种："jsonrpc"、"rest" 或 "grpc"。None 表示自动选择。
streaming	True	流式传输任务；设为 False 时，改为每隔 polling_interval 轮询一次 get_task。
headers	None	每次请求都发送的额外 HTTP 请求头（例如身份验证令牌）。
timeout	60.0	每次请求的超时时间（秒）。
polling_interval	0.5	关闭流式传输或不支持流式传输时的轮询间隔（秒）。
input_required_timeout	None	TASK_STATE_INPUT_REQUIRED 时等待 HITL 钩子的时间上限；None = 永久等待。
tenant	None	将每个请求限定到服务端租户（可通过 context.variables["a2a:tenant"] 按调用覆盖）。
history_length	None	服务端提示：将回传的 Task.history 截断为最近 N 条消息。
httpx_client_factory	None	提供自定义 httpx.AsyncClient 工厂（供下文的进程内测试辅助工具使用）。
preset_card	None	使用已获取的 AgentCard，跳过发现请求往返（参见 A2AConfig.from_card(...)）。

Agent 卡片 — build_card

A2AServer 会自动构建卡片，但你可以直接调用 build_card 自定义卡片，并通过 card= 将结果传给 build_* 方法。

from a2a.types import AgentProvider, AgentSkill

from ag2.a2a import build_card

card = build_card(
    agent,
    url="http://localhost:8000",
    transports=("jsonrpc",),          # one AgentInterface per binding
    version="2.1.0",
    description="Customer support agent",
    push_notifications=True,
    skills=[
        AgentSkill(
            id="refunds",
            name="Process refunds",
            description="Issue refunds for orders",
            tags=["billing"],
        ),
    ],
    provider=AgentProvider(organization="Acme", url="https://acme.example"),
    documentation_url="https://acme.example/docs",
)

值得了解的卡片行为：

• 技能：如果省略 skills，build_card 会自动检测 agent.tools 上的任何 SkillsToolkit，并发布其本地技能；如果没有，则会回退到根据 agent.name / 系统提示词生成的单个技能，从而确保卡片符合规范。
• 默认描述：省略 description 时，会使用 Agent 的系统提示词。
• 始终声明 urn:ag2:client-tools:v1 扩展，并将其设为 required=False（允许 AG2 客户端运行客户端侧工具；对于非 AG2 客户端，则回退为纯文本）。
• transports= 接受 ("jsonrpc", "rest", "grpc") 中的任意值。对于 "grpc"，还必须传入 grpc_url=。

身份验证 — 声明安全方案

使用 ag2.a2a.security 中的工厂辅助函数构建方案，将它们组合到 require(...) 要求中，并将 security=[...] 传给 build_card。卡片的 security_schemes 会根据你引用的方案自动生成，无需重复声明。

from ag2.a2a import build_card
from ag2.a2a.security import api_key_scheme, bearer_scheme, require

bearer = bearer_scheme(name="bearer")                               # Authorization: Bearer <JWT>
apikey = api_key_scheme(name="apikey", key_name="X-API-Key", location="header")

card = build_card(
    agent,
    url="http://localhost:8000",
    # list is OR-ed: present a bearer token OR an API key
    security=[require(bearer), require(apikey)],
)

单个 require(...) 中的方案采用 AND 关系（必须全部提供）；多个 require(...) 条目采用 OR 关系（满足任意一项即可）。OAuth2 / OIDC 作用域通过 .with_scopes(...) 添加：

from a2a.types import AuthorizationCodeOAuthFlow, OAuthFlows

from ag2.a2a.security import bearer_scheme, oauth2_scheme, require

flows = OAuthFlows(
    authorization_code=AuthorizationCodeOAuthFlow(
        authorization_url="https://auth.example/authorize",
        token_url="https://auth.example/token",
        scopes={"read": "Read access", "write": "Write access"},
    ),
)
oauth = oauth2_scheme(name="oauth", flows=flows)
bearer = bearer_scheme(name="bearer")

card = build_card(
    agent,
    url="http://localhost:8000",
    security=[require(bearer, oauth.with_scopes("read", "write"))],  # AND, scoped
)

可用的方案工厂函数：bearer_scheme、http_auth_scheme（basic/digest/...）、api_key_scheme、oauth2_scheme、open_id_connect_scheme、mtls_scheme。卡片只会声明方案——A2A 不提供中间件，因此你必须在返回的 Starlette/gRPC 对象上执行身份验证（例如，使用能验证请求头的 Starlette 中间件）。

从一个服务提供多种传输方式

一个 A2AServer 会在各种传输方式间共享同一个任务存储，因此可以同时以多种方式公开同一个 Agent：

server = A2AServer(agent)
jsonrpc_app = server.build_jsonrpc(url="http://localhost:8000")
rest_app    = server.build_rest(url="http://localhost:8001", path_prefix="/v1")
grpc_srv    = server.build_grpc(bind="0.0.0.0:50051", grpc_url="localhost:50051")  # needs ag2[a2a] grpc extra

• build_jsonrpc(url=...) / build_rest(url=..., path_prefix=...) → Starlette ASGI 应用（使用 uvicorn 提供服务）。
• build_grpc(bind=..., grpc_url=...) → 一个 grpc.aio.Server，需要由你自行调用 start() / await（bind = 监听器地址，grpc_url = 卡片中的公开 URL；仅支持不安全绑定）。

若要将卡片整合为包含多个绑定的单一卡片，请调用 build_card(agent, url=..., transports=("jsonrpc","rest","grpc"), grpc_url=...)，并通过 card= 将其传给每个构建器。

推送通知

向服务传入 PushNotificationConfigStore；卡片的 capabilities.push_notifications 会自动启用。

from a2a.server.tasks import InMemoryPushNotificationConfigStore

from ag2.a2a import A2AServer

server = A2AServer(agent, push_config_store=InMemoryPushNotificationConfigStore())
app = server.build_jsonrpc(url="http://localhost:8000")

客户端侧（使用 AG2 Agent 时），可通过 ag2.a2a.push 中的辅助函数为任务注册 webhook：

from ag2.a2a import A2AConfig
from ag2.a2a.push import A2APushConfig, create_push_notification_config

config = A2AConfig(card_url="http://localhost:8000")
await create_push_notification_config(
    config,
    task_id="task-123",
    push_config=A2APushConfig(url="https://my-app.example/webhook", token="secret"),
)

配套辅助函数：get_push_notification_config、list_push_notification_configs、delete_push_notification_config。任务检查辅助函数位于 ag2.a2a.tasks：get_task、list_tasks（返回带有分页元数据的 ListedTasks）、cancel_task。

这些客户端辅助函数需要一个正在运行且可访问的 A2A 服务（会进行真实的网络往返），因此下方的进程内测试不会调用它们。

测试 — 进程内运行，无需网络或 API 密钥

ag2.a2a.testing 提供 ASGI 辅助函数，通过 httpx.ASGITransport 直接向服务应用分发请求——无需套接字、端口或 SSE 代理。将工厂函数接入 A2AConfig.httpx_client_factory，即可完成完整往返：

import asyncio

from ag2 import Agent
from ag2.a2a import A2AConfig, A2AServer
from ag2.a2a.testing import make_test_client_factory
from ag2.testing import TestConfig  # mock the served agent's LLM

async def main():
    served = Agent(
        name="weather_bot",
        prompt="You answer questions about the weather.",
        config=TestConfig("It is sunny in Paris."),  # no real LLM call
    )
    server = A2AServer(served)

    factory = make_test_client_factory(server, url="http://test")
    remote = Agent("remote", config=A2AConfig(card_url="http://test", httpx_client_factory=factory))

    reply = await remote.ask("What is the weather in Paris?")
    print(await reply.content())  # -> "It is sunny in Paris."

asyncio.run(main())

对于 REST，请使用 make_test_rest_client_factory 和 prefer="rest"：

from ag2.a2a.testing import make_test_rest_client_factory

factory = make_test_rest_client_factory(server, url="http://test")
remote = Agent("remote", config=A2AConfig(card_url="http://test", prefer="rest", httpx_client_factory=factory))

对于 gRPC（没有对应的进程内 ASGI 实现），testing.pick_free_port() 会找到一个空闲 TCP 端口，以便在测试中绑定真实的 gRPC 服务。

此技能中每个代码片段的可运行测试位于 assets/test_snippets.py：

.venv/bin/python skills/ag2-a2a/assets/test_snippets.py

常见问题

• 缺少 a2a extra——pip install "ag2[a2a]"；如果没有它，from ag2.a2a import A2AServer 会引发“缺少额外依赖项”错误。gRPC 还需要 a2a-sdk[grpc] extra。
• 无需 FastAPI——build_jsonrpc/build_rest 已会返回 Starlette ASGI 应用；可直接用 uvicorn 提供服务，也可以将其挂载。不要用手写的 FastAPI 路由封装 Agent。
• url= 是公开 URL，而非绑定地址——它会写入 AgentCard 接口条目，因此客户端必须能够访问。对于 gRPC，bind 是监听地址，客户端连接的地址则是 grpc_url。
• 仅声明认证，不会强制执行——security=[...] 只会在卡片上声明认证方案。A2A 没有中间件规范；请自行在返回的 Starlette/gRPC 对象上强制执行认证。
• 无状态执行器——每个请求都是独立的一轮；AG2 客户端每次调用都会发送完整历史记录。无需粘性会话，但不要假定服务器会在不同轮次之间保存每项任务的记忆。
• build_grpc 需要 grpc_url——如果忘记传入 grpc_url=，build_card(..., transports=("grpc",)) 会引发 ValueError。

深入了解（权威来源）

• ag2/a2a/server.py——A2AServer，以及 build_jsonrpc / build_rest / build_grpc 构建器。
• ag2/a2a/card.py——build_card，以及技能/安全/接口解析。
• ag2/a2a/config.py——A2AConfig、from_card 和所有客户端选项。
• ag2/a2a/security.py——方案工厂和 require。
• ag2/a2a/push.py / tasks.py——客户端推送和任务辅助工具。
• ag2/a2a/testing.py——进程内测试工厂。
• ag2/a2a/events.py——用于观测 AG2 流中线路事件的 A2AEvent 系列。
• A2A 协议：https://a2a-protocol.org/