エージェント間通信の使用

A2A フィールド

次の表は、agent.yamlファイルの a2a: セクション内のフィールドを説明したものです。

LM ツールとして A2A を使用

app.a2a_tools() メソッドは、2 つの Lgachein StructuredTool オブジェクト（discover_available_agents と invoke_a2a_agent ）を返します。これらのオブジェクトを使用すると、LM は他のエージェントを単独で検出し、呼び出すことができます。既存のツールと一緒にこれらをバインドすることで、モデルは単独で他のエージェントにルーティングできます。

agent.yamlファイルで a2a.enabled が false の場合、app.a2a_tools() メソッドでは空のリストが返されるため、すべてのエージェントに を含めても安全です。

次の例では、A2A ツールを既存のツールと並行してバインドします。

from agent_engine_sdk_langgraph import App
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph
from langgraph.prebuilt import ToolNode
app = App(app_name="router-agent")
@app.entrypoint
def build_agent():
    llm = app.llm(ChatOpenAI(model="gpt-5.4"))
    a2a = app.a2a_tools()
    llm_with_tools = llm.bind_tools(
        app.get_tool_schemas() + a2a
    )
    tool_node = ToolNode(app.get_tools() + a2a)
    graph = StateGraph(...)
    return graph.compile(checkpointer=app.checkpointer())

LM にバインドされると、モデルは 2 つのツールを名前で呼び出します。これらのツールは次の関数シグネチャを公開します。

• discover_available_agents() は、呼び出しエージェントに表示されるエージェントのJSONリストを返します。各エントリには、agent_id、name、description、skills、capabilities が含まれます。呼び出しエージェントは、それ自体の結果からフィルタリングされます。

• invoke_a2a_agent(agent_id, message, custom_headers="") は、ターゲットエージェントを呼び出し、結果をJSONとして返します。 custom_headers を指定する場合は、それらをJSON文字列（例: '{"x-tenant": "acme"}'）として渡します。

A2A クライアントを直接呼び出す

決定的なルーティングを使用する場合、クライアントを公開する属性である runtime.a2a を介して AgentToAgentクライアントに直接アクセスできます。

次の例では、能力によってエージェントを検出し、最初の一致を呼び出します。

agents = app._runtime.a2a.discover_agents(
    skills=["policy-lookup"],
    limit=10,
)
if agents:
    resp = app._runtime.a2a.invoke_agent(
        agent_id=agents[0].agent_id,
        message="Look up policy POL-123",
        skill="policy-lookup",
        timeout=120,
    )
    if resp.status == "completed":
        print(resp.result)

runtime.a2aNone属性は、A2 A が利用できない場合に を返します。これがいつ発生するかについては、「 考慮事項 」を参照してください。次の例に示すように、クライアントメソッドを呼び出す前に、必ずこのケースを確認してください。

a2a = app._runtime.a2a
if a2a is None:
    ...

AgentToAgent クラス

AgentToAgent は、OU の /a2a/discover および /a2a/invoke エンドポイントと通信するHTTPクライアントです。エージェントコードでは、このクラスを直接構築するのではなく、runtime.a2a の事前認証済みインスタンスを使用します。

次の例は、 コンストラクターとそのパラメーターを示しています。

AgentToAgent(oe_url: str, auth_token: str = "")

以下の表では、AgentToAgent コンストラクター パラメーターを説明しています。

Discovery_ Agents() メソッド

discover_agents() メソッドは、呼び出し元が表示を許可されているエージェントを返します。 1 つのフィルター パラメーターに複数の値を渡すと、エージェントはそれらの値のいずれか 1 つを満たす場合は一致します。例、skills=["a", "b"] はいずれかの特権を紹介するエージェントを返します。複数のフィルターパラメーターを渡す場合、エージェントはそれらのすべてを満たす必要があります。 A2A が有効になっておらず、allowed_callers を通じて呼び出し元を除外しているエージェントは返されません。

次の例は、このメソッドとそのパラメーターを示しています。

discover_agents(
    project_id: str = "",
    skills: list[str] | None = None,
    capabilities: list[str] | None = None,
    input_modes: list[str] | None = None,
    limit: int = 50,
) -> list[DiscoveredAgent]

次の表では、discover_agents() メソッドのパラメータを説明しています。

invoice_agent() メソッド

invoke_agent() メソッドは、ターゲットエージェントを呼び出し、結果を待ちます。 OA は、トークンを検証し、アクセス制御をチェックし、子実行を作成し、エージェントが完了するまでポーリングします。

次の例は、このメソッドとそのパラメーターを示しています。

invoke_agent(
    agent_id: str,
    message: str,
    skill: str = "",
    timeout: float | None = None,
    custom_headers: dict[str, str] | None = None,
) -> AgentResponse

次の表では、invoke_agent() メソッドのパラメータを説明しています。

応答データ クラス

次の表は、すべて不変のデータクラスである応答タイプを示しています。

resp = app._runtime.a2a.invoke_agent(
    agent_id="ws-billing-001",
    message="Process the payment",
)
if resp.status == "completed":
    handle(resp.result)
elif resp.status == "input-required":
    notify_user(
        "The billing agent needs human approval before continuing."
    )
else:
    log.error("A2A call failed: %s", resp.error)

呼び出されたエージェントにヘッダーを渡す

呼び出されるエージェントにヘッダーをアタッチするには、invoke_agent() の custom_headers パラメータを使用して dict を渡します。 invoke_a2a_agent LM ツールを使用する場合は、ヘッダーをJSON string として渡します。

次の例では、ダイレクトクライアントを使用してカスタム ヘッダーを渡します。

app._runtime.a2a.invoke_agent(
    agent_id="ws-billing-001",
    message="Charge the customer",
    custom_headers={
        "x-tenant": "acme",
        "oauth-token": "<delegated-token>",
    },
)

次の例では、 LM ツールを使用してカスタム ヘッダーを渡します。

invoke_a2a_agent(
    agent_id="ws-billing-001",
    message="Charge the customer",
    custom_headers='{"x-tenant": "acme"}',
)

ヘッダーキーでは、予約された a2a- 大文字と小文字を区別しない プレフィックスは使用できません。 UE は、いずれかのキーが a2a- で始まる場合、400 Bad Request 応答でリクエストを拒否します。 a2a- プレフィックスは、a2a-tokenや a2a-caller-workspace などのプラットフォーム ルーティングと ID ヘッダー用に予約されています。

呼び出し元からの読み取りヘッダー

呼び出しエージェントによって転送されたカスタム ヘッダーを読み取るには、agent_engine_runner_shared.contextクラスの get_current_custom_headers() メソッドを使用します。次の例では、このメソッドを呼び出します。

from agent_engine_runner_shared.context import get_current_custom_headers
headers = get_current_custom_headers()
tenant = headers.get("x-tenant")

get_current_custom_headers() は dict[str, str] を返します。メソッドは返される前にすべての a2a- プラットフォームのヘッダーを処理するため、エージェントコードでは A2A トークンやルーティングメタデータが表示されることはありません。呼び出し元がカスタム ヘッダーを送信しない場合、メソッドは空の辞書を返します。

runtime.a2 の場合、 は なし を返します

runtime.a2a では、次の条件が すべて 当てはまる場合にのみクライアントが返されます。

1. エージェントコードはエージェントサンドボックスで実行中。 A2A は、 ツール サンドボックスで実行されるコードからは使用できません。

2. UE URL が実行コンテキストに存在する。

3. 受信ヘッダーに a2a-token が存在する。 UE は、A2A が設定されている場合、すべてのディスパッチでこのトークンを自動的に挿入します。

いずれの条件も満たされない場合、runtime.a2a は None を返します。 app.a2a_tools() を使用する場合、例外が発生する代わりにツールは構造化エラーJSONを返します。

トークンの有効期間

A2A トークンは、デフォルトで 5 後に期限切れになります。一般的なリクエストと応答のフローでは、これが十分です。トークンの有効期間よりも長く実行されるエージェントは、A2A 呼び出しで 401 応答を受け取る場合があります。自動トークン更新はまだ実装されていません。長時間実行される呼び出しからの 401 応答を一時的な障害として扱います。

自己検出フィルター

discover_available_agents LM ツールは、呼び出しエージェントの独自のワークスペースの結果をフィルタリングして、モデルが自分自身を呼び出さないようにします。未加工の discover_agents()クライアントメソッドを使用する場合、OA は結果に独自のエージェントを含める可能性があります。必要に応じて、自分でフィルタリングします。

プロジェクト内スコープ

現在のリリースでは、 A2A の検出と呼び出しは、 同じプロジェクト内のエージェントに制限されています。プロジェクト間のルーティングはまだ実装されていません。

観察可能性

プラットフォームはすべての A2A 呼び出しを自動的にログに記録し、リンクします。ツールを追加する必要はありません。ターゲットは、parent_execution_id を介して呼び出し元にリンクされた子実行として実行されます。共有 root_session_id は、元のユーザー セッションの下に完全なクロスエージェント呼び出しツリーをグループ化します。