Usar comunicação de agente para agente

Campos A2A

A tabela a seguir descreve os campos na seção a2a: do arquivo agent.yaml:

Use A2A como Ferramentas LLM

O método app.a2a_tools() retorna dois objetos LangChain StructuredTool: discover_available_agents e invoke_a2a_agent. Esses objetos permitem que o LLM descobrir e chamar outros agentes de forma autônoma. Vincule-os às suas ferramentas existentes para que o modelo possa rotear para outros agentes por conta própria.

O método app.a2a_tools() retorna uma lista vazia quando a2a.enabled está false em seu arquivo agent.yaml, portanto é seguro incluir em todos os agentes.

O exemplo a seguir vincula as ferramentas A2A às ferramentas existentes:

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

Quando vinculado ao LLM, o modelo chama as duas ferramentas pelo nome. As ferramentas expõem as seguintes assinaturas de função:

• discover_available_agents() retorna uma lista JSON de agentes visíveis para o agente de chamada. Cada entrada inclui agent_id, name, description, skills e capabilities. O agente de chamada é filtrado a partir de seus próprios resultados.

• invoke_a2a_agent(agent_id, message, custom_headers="") invoca o agente de destino e retorna o resultado como JSON. Se você fornecer custom_headers, passe-a como uma string JSON, por exemplo '{"x-tenant": "acme"}'.

Ligar diretamente para o cliente A2A

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

O exemplo a seguir descobre os agentes por habilidade e invoca a primeira correspondência:

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)

O atributo runtime.a2a retorna None quando A2A não está disponível. Para saber quando isso ocorre, consulte Considerações. Sempre verifique esse caso antes de chamar os métodos do cliente , como mostrado no exemplo a seguir:

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

Classe AgentToAgent

AgentToAgent é um cliente HTTP que se comunica com os endpoints OE /a2a/discover e /a2a/invoke. No código do agente , use a instância pré-autenticada de runtime.a2a em vez de construir essa classe diretamente.

O exemplo a seguir mostra o construtor e seus parâmetros:

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

A tabela a seguir descreve os parâmetros do construtor AgentToAgent:

Método find_agents()

O método discover_agents() retorna agentes que o chamador tem permissão para ver. Quando você passa vários valores em um único parâmetro de filtro, um agente corresponde se satisfizer qualquer um desses valores. Por exemplo, skills=["a", "b"] retorna agentes anunciando qualquer habilidade. Quando você passa vários parâmetros de filtro, um agente deve satisfazer todos eles. Os agentes que não são habilitados para A2A e excluem o chamador por meio de allowed_callers nunca são retornados.

O exemplo a seguir mostra este método e seus parâmetros:

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]

A tabela seguinte descreve os parâmetros para o método discover_agents():

Método invoke_agent()

O método invoke_agent() invoca um agente de destino e espera o resultado. O OE valida o token, verifica o controle de acesso, cria uma execução secundária e pesquisa até que o agente termine, suspenda para revisão humana ou expire.

O exemplo a seguir mostra este método e seus parâmetros:

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

A tabela seguinte descreve os parâmetros para o método invoke_agent():

Classes de dados de resposta

A tabela a seguir lista os tipos de resposta, que são todas classes de dados imutáveis:

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)

Passe cabeçalhos para um agente chamado

Para anexar cabeçalhos a um agente chamado, passe um dict usando o parâmetro custom_headers de invoke_agent(). Ao utilizar a ferramenta LLM invoke_a2a_agent, passe cabeçalhos como uma string JSON.

O exemplo a seguir passa cabeçalhos personalizados usando o cliente direto :

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

O exemplo a seguir passa cabeçalhos personalizados usando a ferramenta LLM:

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

As chaves de cabeçalho não devem usar o prefixo reservado a2a- insensível a maiúsculas e minúsculas. A OE rejeita solicitações com uma resposta 400 Bad Request se alguma chave começar com a2a-. O prefixo a2a- é reservado para roteamento de plataforma e cabeçalhos de identidade como a2a-token e a2a-caller-workspace.

Ler cabeçalhos de um chamador

Para ler cabeçalhos personalizados encaminhados por um agente de chamada, use o método get_current_custom_headers() da classeagent_engine_runner_shared.context. O exemplo a seguir chama este método:

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() retorna um dict[str, str]. O método remove todos os cabeçalhos de plataforma a2a- antes de retornar, para que o código do agente nunca veja tokens A2A ou metadados de roteamento. Se o chamador não enviar cabeçalhos personalizados, o método retornará um dicionário vazio.

Quando runtime.a2a não retorna nenhum

runtime.a2a retorna um cliente somente quando todas as seguintes condições forem verdadeiras:

1. O código do agente está sendo executado na sandbox do agente . A2A não está disponível para código executado na sandbox da ferramenta.

2. Uma URL OE está presente no contexto de execução.

3. Um a2a-token está presente nos cabeçalhos recebidos. O OE injeta esse token automaticamente em cada despacho quando o A2A é configurado.

Quando qualquer condição não é atendida, runtime.a2a retorna None. Ao usar app.a2a_tools(), as ferramentas retornam um JSON de erro estruturado em vez de gerar uma exceção.

Vida útil do token

Por padrão, o token A2A expira após 5 minutos. Para fluxos típicos de solicitação e resposta, isso é suficiente. Os agentes que funcionam por mais tempo do que a vida útil do token podem receber uma resposta 401 nas chamadas A2A. A atualização automática do token ainda não foi implementada. Trate uma resposta 401 de uma chamada de longa duração como uma falha transitória.

Filtro de autodescoberta

A ferramenta LLM discover_available_agents filtra o espaço de trabalho do próprio agente de chamada dos resultados para que o modelo não chame a si mesmo. Se você usar o método bruto do cliente discover_agents(), o OE poderá incluir seu próprio agente nos resultados. Filtre você mesmo, se necessário.

Escopo Intra-Projeto

Na versão atual, a descoberta e a invocação de A2A são limitadas aos agentes no mesmo projeto. O roteamento entre projetos ainda não foi implementado.

Observabilidade

A plataforma registra e vincula todas as chamadas A2A automaticamente. Você não precisa adicionar instrumentação. O destino é executado como uma execução filho vinculada ao chamador por meio de parent_execution_id. Um root_session_id compartilhado agrupa a árvore completa de chamadas entre agentes na sessão do usuário de origem.