ag2-a2a

A2A（エージェント間）連携

AG2 の Agent を A2A プロトコル 経由で公開します。A2A は標準的な JSON-RPC／REST／gRPC 契約と、検出可能な AgentCard を提供するため、他のシステムからネットワーク経由でエージェントを呼び出せます。同じモジュールでは、A2AConfig をエージェントのモデル設定として使い、AG2 の Agent からリモート A2A エージェントを利用することもできます。

使用する場面

• Python からだけでなく、AG2 以外の A2A クライアントからも呼び出せる、標準的なネットワークサービスとして AG2 エージェントを利用したい場合（相互運用）。
• エージェントのスキル、トランスポート、認証、機能を記述する AgentCard を /.well-known/agent-card.json に公開して、エージェントを検出可能にしたい場合。
• トランスポート（JSON-RPC、REST、gRPC）を選択したい場合（1 つのサーバーから複数を同時に使うこともできます）。
• 認証方式（ベアラー／API キー／OAuth2／OIDC／mTLS）、プッシュ通知、マルチテナンシーをカードに明示したい場合。
• 反対に、AG2 エージェントからリモート A2A エージェントを LLM プロバイダーのように呼び出したい場合は、呼び出し側エージェントの config として A2AConfig を使います。

エージェントの前にウェブフロントエンド（React/CopilotKit）だけが必要な場合は、代わりに AG-UI スキルを使ってください。2 つの AG2 エージェントが 1 つのプロセス／クラスター内で通信するだけであれば、標準的なマルチエージェントパターンである AG2 ネットワークを使ってください。

インストール

pip install "ag2[a2a]"

必須です。a2a エクストラには a2a-sdk（HTTP サーバーを含む）が含まれます。gRPC には追加で a2a-sdk[grpc] エクストラが必要です。コマンドを実行できない場合は、正確な pip install コマンドを伝えてください。

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

60 秒でできるレシピ — A2A 経由でエージェントを提供する

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

エージェントカードは 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 エージェントから提供中のエージェントを呼び出す

A2AConfig は ModelConfig です。リモートカードの 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	タスクをストリーミングします。代わりに polling_intervals ごとに get_task をポーリングするには、False に設定します。
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(...) を参照）。

エージェントカード — 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 を省略すると、エージェントのシステムプロンプトが使用されます。
• 必ず宣言される項目：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 ミドルウェア）。

1 つのサーバーから複数のトランスポートを利用する

1 つの A2AServer は、トランスポート間で単一のタスクストアを共有するため、同じエージェントを複数の方法で同時に公開できます。

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=...) → 自分で start()／await する grpc.aio.Server（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 エージェントを利用する側）では、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 には、httpx.ASGITransport を介してサーバーアプリに直接ディスパッチする ASGI ヘルパーが含まれています。ソケット、ポート、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() を使うと、テストで実際の gRPC サーバーをバインドするための空き TCP ポートを見つけられます。

このスキルのすべてのスニペットを実行するテストは、assets/test_snippets.py にあります。

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

よくある落とし穴

• a2a エクストラが不足している — pip install "ag2[a2a]" を実行してください。これがないと、from ag2.a2a import A2AServer で「追加の依存関係が不足しています」というエラーが発生します。gRPC にはさらに a2a-sdk[grpc] エクストラが必要です。
• FastAPI は不要です — build_jsonrpc／build_rest はすでに Starlette ASGI アプリを返します。uvicorn で直接サーブするか、マウントしてください。エージェントを独自に作成した FastAPI ルートでラップしないでください。
• url= は公開 URL であり、バインドアドレスではありません — この URL は AgentCard のインターフェース項目に書き込まれるため、クライアントからアクセスできる必要があります。gRPC の場合、bind はリスナー、grpc_url はクライアントが接続する 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/