Utilice la comunicación de agente a agente.

Campos A2A

La siguiente tabla describe los campos de la sección a2a: del archivo agent.yaml:

Utilice A2A como herramienta LLM

El método app.a2a_tools() devuelve dos objetos LangChain StructuredTool: discover_available_agents y invoke_a2a_agent. Estos objetos permiten que el LLM descubra y llame a otros agentes de forma autónoma. Intégrelos con sus herramientas existentes para que el modelo pueda enrutar a otros agentes por sí mismo.

El método app.a2a_tools() devuelve una lista vacía cuando a2a.enabled es false en su archivo agent.yaml, por lo que es seguro incluirlo en todos los agentes.

El siguiente ejemplo vincula las herramientas A2A con las herramientas 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())

Cuando se vincula al LLM, el modelo llama a las dos herramientas por su nombre. Las herramientas exponen las siguientes firmas de función:

• discover_available_agents() Devuelve una lista JSON de agentes visibles para el agente que realiza la llamada. Cada entrada incluye agent_id, name, description, skills y capabilities. El agente que realiza la llamada se filtra a partir de sus propios resultados.

• invoke_a2a_agent(agent_id, message, custom_headers="") Invoca al agente de destino y devuelve el resultado como JSON. Si proporciona custom_headers, páselo como una cadena JSON, por ejemplo '{"x-tenant": "acme"}'.

Llame directamente al cliente de A2A

Cuando quieras utilizar el enrutamiento determinista, puedes acceder al cliente AgentToAgent directamente a través de runtime.a2a, un atributo que expone el cliente.

El siguiente ejemplo descubre agentes por habilidad e invoca la primera coincidencia:

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)

El atributo runtime.a2a devuelve None cuando A2A no está disponible. Para saber cuándo ocurre esto, consulte Consideraciones. Compruebe siempre este caso antes de llamar a los métodos del cliente, como se muestra en el siguiente ejemplo:

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

Clase Agente a Agente

AgentToAgent Es un cliente HTTP que se comunica con los puntos finales /a2a/discover y /a2a/invoke de OE. En el código del agente, utilice la instancia preautenticada de runtime.a2a en lugar de construir esta clase directamente.

El siguiente ejemplo muestra el constructor y sus parámetros:

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

La siguiente tabla describe los parámetros del constructor AgentToAgent:

Método discover_agents()

El método discover_agents() devuelve los agentes que el usuario tiene permiso para ver. Al pasar varios valores en un único parámetro de filtro, un agente coincide si cumple con alguno de esos valores. Por ejemplo, skills=["a", "b"] devuelve los agentes que anuncian cualquiera de las habilidades. Al pasar varios parámetros de filtro, un agente debe cumplir con todos ellos. Los agentes que no tienen la función A2A habilitada y excluyen al usuario mediante allowed_callers nunca se devuelven.

El siguiente ejemplo muestra este método y sus 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]

La siguiente tabla describe los parámetros para el método discover_agents():

Método invoke_agent()

El método invoke_agent() invoca a un agente objetivo y espera el resultado. El OE valida el token, comprueba el control de acceso, crea una ejecución secundaria y realiza sondeos hasta que el agente finaliza, se suspende para revisión humana o se agota el tiempo de espera.

El siguiente ejemplo muestra este método y sus parámetros:

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

La siguiente tabla describe los parámetros para el método invoke_agent():

Clases de datos de respuesta

La siguiente tabla enumera los tipos de respuesta, que son todos clases de datos inmutables:

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)

Pasar encabezados a un agente llamado

Para adjuntar encabezados a un agente llamado, pase un dict usando el parámetro custom_headers de invoke_agent(). Cuando use la herramienta LLM invoke_a2a_agent, pase los encabezados como una cadena JSON.

El siguiente ejemplo pasa encabezados personalizados mediante el uso del cliente directo:

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

El siguiente ejemplo pasa encabezados personalizados mediante la herramienta LLM:

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

Las claves de encabezado no deben usar el prefijo reservado a2a- que no distingue entre mayúsculas y minúsculas. El OE rechaza las solicitudes con una respuesta 400 Bad Request si alguna clave comienza con a2a-. El prefijo a2a- está reservado para el enrutamiento de la plataforma y los encabezados de identidad, como a2a-token y a2a-caller-workspace.

Leer los encabezados de una persona que realiza la llamada

Para leer los encabezados personalizados reenviados por un agente que realiza la llamada, utilice el método get_current_custom_headers() de la clase agent_engine_runner_shared.context. El siguiente ejemplo llama a 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() Devuelve un dict[str, str]. El método elimina todos los encabezados de plataforma a2a- antes de devolverlo, por lo que el código de su agente nunca ve tokens A2A ni metadatos de enrutamiento. Si quien realiza la llamada no envía encabezados personalizados, el método devuelve un diccionario vacío.

Cuando runtime.a2a devuelve None

runtime.a2a Devuelve un cliente solo cuando se cumplen todas las siguientes condiciones:

1. El código del agente se está ejecutando en el entorno aislado del agente. A2A no está disponible desde el código que se ejecuta en el entorno aislado de la herramienta.

2. En el contexto de ejecución hay una URL de OE presente.

3. En los encabezados entrantes se encuentra presente un a2a-token. El OE inyecta este token automáticamente en cada envío cuando se configura A2A.

Cuando no se cumple alguna condición, runtime.a2a devuelve None. Al usar app.a2a_tools(), las herramientas devuelven un JSON de error estructurado en lugar de generar una excepción.

Duración del token

El token A2A caduca después de 5 minutos por defecto. Para flujos típicos de solicitud y respuesta, esto es suficiente. Los agentes que se ejecutan durante más tiempo que la vida útil del token pueden recibir una respuesta 401 en llamadas A2A. La actualización automática del token aún no está implementada. Considere una respuesta 401 de una llamada de larga duración como un fallo transitorio.

Filtro de autodescubrimiento

La herramienta LLM discover_available_agents filtra el espacio de trabajo del agente que realiza la llamada para evitar que el modelo se llame a sí mismo. Si utiliza el método de cliente discover_agents() sin procesar, el OE podría incluir su propio agente en los resultados. Filtre este espacio manualmente si es necesario.

Alcance dentro del proyecto

En la versión actual, el descubrimiento y la invocación de A2A están limitados a agentes del mismo proyecto. El enrutamiento entre proyectos aún no está implementado.

Observabilidad

La plataforma registra y vincula automáticamente cada llamada A2A. No es necesario añadir instrumentación. El objetivo se ejecuta como una ejecución secundaria vinculada al emisor a través de parent_execution_id. Un root_session_id compartido agrupa todo el árbol de llamadas entre agentes bajo la sesión del usuario de origen.