에이전트 간 통신 사용

A2A 필드

다음 표에서는 agent.yaml 파일 의 a2a: 섹션에 있는 필드에 대해 설명합니다.

A2A를 LLM 도구로 사용

app.a2a_tools() 메서드는 두 개의 LangChain StructuredTool 객체 discover_available_agents 및 invoke_a2a_agent을 반환합니다. 이러한 객체를 통해 LLM은 자동으로 다른 에이전트를 검색하고 호출할 수 있습니다. 모델이 자체적으로 다른 에이전트로 라우팅할 수 있도록 기존 도구와 함께 바인딩합니다.

app.a2a_tools() 메서드는 agent.yaml 파일 에서 a2a.enabled이 false인 경우 빈 목록을 반환하므로 모든 에이전트에 포함하는 것이 안전합니다.

다음 예시 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())

LLM에 바인딩되면 모델은 두 도구를 이름으로 호출합니다. 이 도구는 다음과 같은 함수 시그니처를 노출합니다.

• 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.a2a 속성은 A2A를 사용할 수 없을 때 None을 반환합니다. 이러한 상황이 발생하는 경우를 학습 고려 사항을 참조하세요. 다음 예시 와 같이 클라이언트 메서드를 호출하기 전에 항상 이 대소문자를 확인하세요.

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

AgentToAgent 클래스

AgentToAgent 는 OE /a2a/discover 및 /a2a/invoke 엔드포인트와 통신하는 HTTP 클라이언트 입니다. 에이전트 코드에서 이 클래스를 직접 구성하는 대신 runtime.a2a의 사전 인증된 인스턴스 사용합니다.

다음 예시 생성자와 해당 매개변수를 보여줍니다.

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

다음 표에서는 AgentToAgent 생성자 매개변수에 대해 설명합니다:

discover_agents() 메서드

discover_agents() 메서드는 호출자가 볼 수 있도록 허용된 에이전트를 반환합니다. 단일 필터하다 매개변수에 여러 값을 전달할 때 에이전트 해당 값 중 하나를 충족하면 일치합니다. 예시 를 들어, 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() 메서드의 매개변수에 대해 설명합니다.

invoke_agent() 메서드

invoke_agent() 메서드는 대상 에이전트 호출하고 결과를 기다립니다. OE는 토큰의 유효성을 검사하고, 액세스 제어를 확인하고, 하위 실행을 생성하고, 에이전트 완료되거나, 사람의 검토 위해 일시 중단되거나, 시간이 초과될 때까지 폴링합니다.

다음 예시 이 메서드와 해당 매개변수를 보여줍니다.

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 LLM 도구를 사용하는 경우, 헤더를 JSON 문자열로 전달합니다.

다음 예시 직접 클라이언트 사용하여 사용자 지정 헤더를 전달합니다.

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

다음 예시 LLM 도구를 사용하여 사용자 지정 헤더를 전달합니다.

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

헤더 키는 대소문자를 구분하지 않는 예약된 a2a- 접두사를 사용해서는 안 됩니다. OE는 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 토큰이나 라우팅 메타데이터 표시되지 않습니다. 호출자가 사용자 지정 헤더를 보내지 않으면 메서드는 빈 사전을 반환합니다.

런타임.a2반환 없음

runtime.a2a 다음 조건이 모두 참인 경우에만 클라이언트 반환합니다.

1. 에이전트 코드가 에이전트 샌드박스에서 실행 입니다. 도구 샌드박스에서 실행되는 코드에서는 A2A를 사용할 수 없습니다.

2. OE URL 은 실행 컨텍스트에 있습니다.

3. 수신 헤더에 a2a-token이(가) 있습니다. OE는 A2A가 구성될 때마다 디스패치할 때마다 이 토큰을 자동으로 삽입합니다.

조건이 충족되지 않으면 runtime.a2a이 None을 반환합니다. app.a2a_tools()를 사용할 때 도구는 예외를 발생시키는 대신 구조화된 오류 JSON 반환합니다.

토큰 수명

A2A 토큰은 기본값 으로 5분 후에 만료됩니다. 일반적인 요청 및 응답 흐름의 경우 이 정도면 충분합니다. 토큰 수명보다 오래 실행 에이전트는 A2A 호출에 대해 401 응답을 받을 수 있습니다. 자동 토큰 새로 고침은 아직 구현되지 않았습니다. 장기 실행 호출의 401 응답을 일시적 실패로 처리합니다.

자체 검색 필터

discover_available_agents LLM 도구는 모델이 자체적으로 호출하지 않도록 결과에서 호출 에이전트의 자체 작업 공간을 필터링합니다. 원시 discover_agents() 클라이언트 메서드를 사용하는 경우, OE는 결과에 자체 에이전트 포함할 수 있습니다. 필요한 경우 직접 필터링합니다.

프로젝트 내 범위

현재 출시하다 에서는2A 검색 및 호출이 동일한 프로젝트 의 에이전트로 제한됩니다. 프로젝트 간 라우팅은 아직 구현되지 않았습니다.

관찰 가능성

이 플랫폼은 모든 A2A 호출을 자동으로 기록하고 연결합니다. 계측을 추가할 필요가 없습니다. 대상은 parent_execution_id를 통해 호출자에 연결된 하위 실행으로 실행됩니다. 공유 root_session_id는 원래 사용자 세션 아래에 전체 교차 에이전트 호출 트리를 그룹화합니다.