Use Agent-to-Agent Communication

A2A Fields

The following table describes the fields in the a2a: section of the agent.yaml file:

Use A2A as LLM Tools

The app.a2a_tools() method returns two LangChain StructuredTool objects: discover_available_agents and invoke_a2a_agent. These objects let the LLM discover and call other agents autonomously. Bind them alongside your existing tools so the model can route to other agents on its own.

The app.a2a_tools() method returns an empty list when a2a.enabled is false in your agent.yaml file, so it is safe to include in all agents.

The following example binds A2A tools alongside existing tools:

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())

When bound to the LLM, the model calls the two tools by name. The tools expose the following function signatures:

• discover_available_agents() returns a JSON list of agents visible to the calling agent. Each entry includes agent_id, name, description, skills, and capabilities. The calling agent is filtered from its own results.

• invoke_a2a_agent(agent_id, message, custom_headers="") invokes the target agent and returns the result as JSON. If you provide custom_headers, pass them as a JSON string, for example '{"x-tenant": "acme"}'.

Call the A2A Client Directly

When you want to use deterministic routing, you can access the AgentToAgent client directly through runtime.a2a, an attribute that exposes the client.

The following example discovers agents by skill and invokes the first match:

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)

The runtime.a2a attribute returns None when A2A is not available. To learn when this occurs, see Considerations. Always check for this case before calling client methods, as shown in the following example:

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

AgentToAgent Class

AgentToAgent is an HTTP client that communicates with the OE /a2a/discover and /a2a/invoke endpoints. In agent code, use the pre-authenticated instance from runtime.a2a rather than constructing this class directly.

The following example shows the constructor and its parameters:

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

The following table describes the AgentToAgent constructor parameters:

discover_agents() Method

The discover_agents() method returns agents that the caller is permitted to see. When you pass multiple values in a single filter parameter, an agent matches if it satisfies any one of those values. For example, skills=["a", "b"] returns agents advertising either skill. When you pass multiple filter parameters, an agent must satisfy all of them. Agents that are not A2A-enabled and exclude the caller through allowed_callers are never returned.

The following example shows this method and its parameters:

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]

The following table describes the parameters for the discover_agents() method:

invoke_agent() Method

The invoke_agent() method invokes a target agent and waits for the result. The OE validates the token, checks access control, creates a child execution, and polls until the agent finishes, suspends for human review, or times out.

The following example shows this method and its parameters:

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

The following table describes the parameters for the invoke_agent() method:

Response Data Classes

The following table lists the response types, which are all are immutable dataclasses:

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)

Pass Headers to a Called Agent

To attach headers to a called agent, pass a dict by using the custom_headers parameter of invoke_agent(). When using the invoke_a2a_agent LLM tool, pass headers as a JSON string.

The following example passes custom headers by using the direct client:

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

The following example passes custom headers by using the LLM tool:

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

Header keys must not use the reserved a2a- case-insensitive prefix. The OE rejects requests with a 400 Bad Request response if any key starts with a2a-. The a2a- prefix is reserved for platform routing and identity headers such as a2a-token and a2a-caller-workspace.

Read Headers from a Caller

To read custom headers forwarded by a calling agent, use the get_current_custom_headers() method from the agent_engine_runner_shared.context class. The following example calls this method:

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() returns a dict[str, str]. The method strips all a2a- platform headers before returning, so your agent code never sees A2A tokens or routing metadata. If the caller sends no custom headers, the method returns an empty dictionary.

When runtime.a2a Returns None

runtime.a2a returns a client only when all of the following conditions are true:

1. The agent code is running in the agent sandbox. A2A is not available from code that runs in the tool sandbox.

2. An OE URL is present in the execution context.

3. An a2a-token is present in the incoming headers. The OE injects this token automatically on every dispatch when A2A is configured.

When any condition is not met, runtime.a2a returns None. When using app.a2a_tools(), the tools return a structured error JSON instead of raising an exception.

Token Lifetime

The A2A token expires after 5 minutes by default. For typical request-and-response flows this is sufficient. Agents that run longer than the token lifetime may receive a 401 response on A2A calls. Automatic token refresh is not yet implemented. Treat a 401 response from a long-running call as a transient failure.

Self-Discovery Filter

The discover_available_agents LLM tool filters the calling agent's own workspace from results so the model does not call itself. If you use the raw discover_agents() client method, the OE may include your own agent in results. Filter it out yourself if needed.

Intra-Project Scope

In the current release, A2A discovery and invocation are limited to agents in the same project. Cross-project routing is not yet implemented.

Observability

The platform logs and links every A2A call automatically. You do not need to add instrumentation. The target runs as a child execution linked to the caller through parent_execution_id. A shared root_session_id groups the full cross-agent call tree under the originating user session.