Agent Server의 A2A 엔드포인트

엔벨로프는 패밀리가 아니라 메서드에 따라 다릅니다: SendMessage는 작업을 result.task로 감싸지만, GetTask와 모든 v0.3 메서드는 result에 직접 반환합니다. ListTasks는 result.tasks를 반환합니다.

지원 메서드

정확히 네 개의 v0.3 이름이 수용됩니다: message/send, message/stream, tasks/get, tasks/cancel. 그 외의 것 — agent/getAuthenticatedExtendedCard, tasks/resubscribe를 포함해 — 은 -32601 Method not found를 반환합니다.

JSON-RPC 바인딩만 사용할 수 있습니다. gRPC와 HTTP+JSON은 구현되지 않았습니다.

응답의 작업 기록 (Task history in responses)

컨텍스트는 많은 작업을 보유합니다. 기본적으로 SendMessage, GetTask, ListTasks는 요청한 작업뿐 아니라 전체 컨텍스트의 기록을 반환합니다. 컨텍스트의 두 번째 작업은 첫 번째 작업의 메시지를 다시 재생하므로, 모든 기록 항목을 렌더링하는 클라이언트는 이전 턴을 다시 보여줍니다 — 이전 도구 결과와 A2UI 페이로드까지요.

historyScope를 task로 설정하면 요청한 작업에 속한 메시지만 돌려받을 수 있습니다. 기본값은 context로 유지되므로 기존 통합에는 영향이 없습니다.

옵션이 어디에 가는지는 메서드에 따라 다릅니다. SendMessage는 configuration에서 읽습니다:

{
  "jsonrpc": "2.0",
  "id": "1",
  "method": "SendMessage",
  "params": {
    "message": {
      "role": "ROLE_USER",
      "parts": [{"text": "Second request"}],
      "messageId": "message-2",
      "contextId": "8b1f0e5c-9a3d-4f27-b0c8-2e6a5d4c1b7a"
    },
    "configuration": {"historyScope": "task"}
  }
}

GetTask와 ListTasks는 params에서 직접 읽습니다:

{"jsonrpc": "2.0", "id": "2", "method": "GetTask",
 "params": {"id": "<taskId>", "historyScope": "task"}}

클라이언트가 요청 본문에 필드를 추가할 수 없으면 대신 헤더를 보내세요. 요청의 명시적 값이 헤더보다 우선합니다.

LangGraph-A2A-History-Scope: task

에이전트 카드는 capabilities.extensions 아래에 이를 광고하므로 가정하지 않고 지원 여부를 감지할 수 있습니다:

{
  "uri": "https://langchain.com/a2a/extensions/history-scope/v1",
  "description": "Choose task-only or full-context response history",
  "required": false,
  "params": {
    "header": "LangGraph-A2A-History-Scope",
    "values": ["context", "task"],
    "default": "context",
    "methods": ["SendMessage", "GetTask", "ListTasks"]
  }
}

알아두면 좋은 세 가지 제한:

• 스트리밍은 두 기록 옵션을 모두 무시합니다. SendStreamingMessage는 historyScope도 historyLength도 읽지 않으며, 보내도 오류를 반환하지 않습니다 — 그러니 SSE에서는 둘 다에 의존하지 마세요.
• historyLength는 10으로 제한됩니다. 더 큰 값은 historyLength cannot exceed 10과 함께 -32602를 반환합니다.
• 범위는 historyLength보다 먼저 적용되므로 해당 작업의 마지막 N개 메시지를 얻습니다.

인식할 수 없는 값은 historyScope must be 'context' or 'task'와 함께 -32602를 반환합니다. historyscope처럼 대소문자가 틀린 키는 오류가 아닙니다 — 무시되고 조용히 전체 컨텍스트 기록을 받게 되므로, 필터링이 작동하지 않는 것 같으면 철자를 확인하세요.

에이전트 카드 발견 (Agent card discovery)

각 어시스턴트는 자신의 기능을 설명하고 다른 에이전트가 연결하는 데 필요한 정보를 제공하는 A2A 에이전트 카드를 자동으로 노출합니다. 다음을 사용해 모든 어시스턴트의 에이전트 카드를 검색할 수 있습니다:

GET /.well-known/agent-card.json?assistant_id={assistant_id}

에이전트 카드에는 어시스턴트의 이름, 설명, 사용 가능한 스킬, 지원되는 입출력 모드, 통신을 위한 A2A 엔드포인트 URL이 포함됩니다.

선택적 기능 (Optional capabilities)

이들은 Assistants API의 metadata.a2a를 통해 어시스턴트별로 구성됩니다. langgraph.json은 어시스턴트 메타데이터를 설정할 수 없으므로 배포 후 어시스턴트를 패치하세요.

입출력 모드 선언 (Declare input and output modes)

{
  "metadata": {
    "a2a": {
      "input_modes": ["text/plain", "application/pdf"],
      "output_modes": ["text/plain", "application/pdf"]
    }
  }
}

값은 카드의 defaultInputModes/defaultOutputModes와 생성된 스킬의 모드 모두에 반영됩니다. 이는 광고일 뿐입니다 — 선언되지 않은 모드도 여전히 수용됩니다. 교체는 필드별로 이루어지므로 둘 다 덮어쓰려면 둘 다 보내야 하고, 빈 목록은 거부된다는 점을 유의하세요.

파일 파트 (File parts)

FilePart는 양방향으로 작동합니다. 인바운드 파일, 이미지, 오디오, 비디오 파트는 LangChain 콘텐츠 블록이 됩니다. 아웃바운드 콘텐츠 블록은 작업 기록과 최종 스트리밍 산출물에서 FilePart로 다시 매핑됩니다. MIME 타입, URI, 파일 이름은 변함없이 통과하고, 인라인 데이터는 표준 base64로 재인코딩됩니다.

A2UI v0.9

어시스턴트별로 옵트인:

{ "metadata": { "a2a": { "a2ui": true } } }

그러면 카드가 확장을 광고하고 정식 MIME 타입을 두 모드 목록에 모두 추가합니다:

"capabilities": {
  "extensions": [
    {
      "uri": "https://a2ui.org/a2a-extension/a2ui/v0.9",
      "description": "Ability to render A2UI v0.9",
      "required": false,
      "params": {
        "v0.9": {
          "supportedCatalogIds": [],
          "acceptsInlineCatalogs": false
        }
      }
    }
  ]
}

클라이언트는 message.extensions에 URI를 나열해 활성화합니다. 페이로드는 v0.9 스키마에 대해 양방향으로 검증되며, A2UI 파트는 message/send, tasks/get, 최종 message/stream 산출물에서 보존됩니다. 응답은 metadata.mimeType: "application/a2ui+json"을 지닙니다. application/json+a2ui는 입력의 별칭으로 수용됩니다.

도구 결과 필터링 (Filter tool results)

기본적으로 상관된 모든 도구 결과가 DataPart로 게시됩니다. 일부만 게시하려면 배포에서 도구 이름의 허용 목록을 설정하세요:

A2A_ALLOWED_TOOL_CALL_RESULTS=generative_ui_tool,another_tool

설정하지 않으면 모든 도구 결과가 게시됩니다. 필터는 작업 기록과 스트리밍 모두에 적용됩니다.

요건 (Requirements)

pip install "langgraph-api>=0.13.0"

A2UI v0.9와 historyScope는 0.14.0 릴리스 후보가 잘린 이후 추가되었으므로 0.15.0에서 제공됩니다. 그 버전은 아직 안정 릴리스로 게시되지 않았습니다 — historyScope에 의존하기 전에 에이전트 카드의 capabilities.extensions를 확인하세요.

그래프의 상태에 A2A 텍스트와 파일 파트를 수용하려면 messages 키가 포함되어야 합니다. 입력 스키마에 messages 필드가 없는 어시스턴트는 설명적인 오류와 함께 거부됩니다.

A2A 호환 에이전트 만들기 (Creating an A2A-compatible agent)

이 예시는 OpenAI API로 수신 메시지를 처리하고 대화 상태를 유지하는 A2A 호환 에이전트를 만듭니다. 에이전트는 메시지 기반 상태 구조를 정의하고 A2A 프로토콜의 메시지 형식을 처리합니다.

A2A "text" 파트와 호환되려면 에이전트의 state에 messages 키가 있어야 합니다.

A2A 프로토콜은 대화 연속성을 유지하기 위해 두 식별자를 사용합니다:

• contextId: 메시지를 대화 스레드(세션 ID처럼)로 그룹화
• taskId: 해당 대화 내에서 각 개별 요청을 식별

첫 번째 메시지에서는 둘 다 생략하세요 — 에이전트가 생성해 반환합니다. 대화의 모든 후속 메시지에서는 이전 응답의 contextId를 보내고 taskId는 생략해서, 각 턴이 같은 대화 안에서 새 작업을 열게 합니다. taskId는 여전히 실행 중인 작업(입력을 기다리는 것 같은)에 추가할 때만 보내세요.

LangSmith 트레이싱: LangSmith Deployment A2A 엔드포인트는 LangSmith 트레이싱을 위해 A2A contextId를 자동으로 thread_id로 변환하여 대화의 모든 메시지를 단일 스레드로 그룹화합니다.

예를 들어:

"""LangGraph A2A conversational agent.

Supports the A2A protocol with messages input for conversational interactions.

"""

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Any, Dict, List, TypedDict

from langgraph.graph import StateGraph
from langgraph.runtime import Runtime
from openai import AsyncOpenAI


class Context(TypedDict):
    """Context parameters for the agent."""
    my_configurable_param: str


@dataclass
class State:
    """Input state for the agent.

    Defines the initial structure for A2A conversational messages.
    """
    messages: List[Dict[str, Any]]


async def call_model(state: State, runtime: Runtime[Context]) -> Dict[str, Any]:
    """Process conversational messages and returns output using OpenAI."""
    # Initialize OpenAI client
    client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    # Process the incoming messages
    latest_message = state.messages[-1] if state.messages else {}
    user_content = latest_message.get("content", "No message content")

    # Create messages for OpenAI API
    openai_messages = [
        {
            "role": "system",
            "content": "You are a helpful conversational agent. Keep responses brief and engaging."
        },
        {
            "role": "user",
            "content": user_content
        }
    ]

    try:
        # Make OpenAI API call
        response = await client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=openai_messages,
            max_tokens=100,
            temperature=0.7
        )

        ai_response = response.choices[0].message.content

    except Exception as e:
        ai_response = f"I received your message but had trouble processing it. Error: {str(e)[:50]}..."

    # Create a response message
    response_message = {
        "role": "assistant",
        "content": ai_response
    }

    return {
        "messages": state.messages + [response_message]
    }


# Define the graph
graph = (
    StateGraph(State, context_schema=Context)
    .add_node(call_model)
    .add_edge("__start__", "call_model")
    .compile()
)

에이전트 간 통신 (Agent-to-agent communication)

에이전트가 langgraph dev로 로컬에서 실행되거나 프로덕션에 배포되면, A2A 프로토콜을 사용해 에이전트 간 통신을 촉진할 수 있습니다.

이 예시는 두 에이전트가 서로의 A2A 엔드포인트에 JSON-RPC 메시지를 보내 통신하는 방법을 보여줍니다. 스크립트는 각 에이전트가 상대의 응답을 처리하고 대화를 이어가는 다중 턴 대화를 시뮬레이션합니다.

#!/usr/bin/env python3
"""Agent-to-Agent conversation simulation using the LangGraph A2A endpoint."""

import asyncio
import aiohttp
import os
import uuid


def extract_text(result: dict) -> str:
    """Best-effort extraction of response text from an A2A result."""
    if "error" in result:
        raise RuntimeError(f"A2A error {result['error']['code']}: {result['error']['message']}")

    for art in result.get("result", {}).get("artifacts", []) or []:
        for part in art.get("parts", []) or []:
            if part.get("kind") == "text" and part.get("text"):
                return part["text"]

    msg = (result.get("result", {}).get("status", {}) or {}).get("message", {}) or {}
    for part in msg.get("parts", []) or []:
        if part.get("kind") == "text" and part.get("text"):
            return part["text"]

    return "(no text found)"


async def send_message(session, port, assistant_id, text, context_id=None):
    """Send an A2A message. Returns (response_text, returned_context_id)."""
    url = f"http://127.0.0.1:{port}/a2a/{assistant_id}"

    message = {
        "role": "user",
        "parts": [{"kind": "text", "text": text}],
        "messageId": str(uuid.uuid4()),
    }

    # A2A multi-turn continuity: reuse contextId across turns and agents.
    # Do not reuse taskId — each turn starts a new task within the same context.
    if context_id:
        message["contextId"] = context_id

    payload = {
        "jsonrpc": "2.0",
        "id": str(uuid.uuid4()),
        "method": "message/send",
        "params": {"message": message},
    }

    headers = {"Accept": "application/json"}
    async with session.post(url, json=payload, headers=headers) as response:
        result = await response.json()

    text = extract_text(result)
    returned_context_id = result.get("result", {}).get("contextId") or context_id
    return text, returned_context_id


async def simulate_conversation():
    """Simulate a conversation between two agents."""

    #Assistant IDs
    agent_a_id = os.getenv("AGENT_A_ID")
    agent_b_id = os.getenv("AGENT_B_ID")

    if not agent_a_id or not agent_b_id:
        print("Set AGENT_A_ID and AGENT_B_ID environment variables")
        return

    message = "Hello! Let's have a conversation."
    context_id = None

    async with aiohttp.ClientSession() as session:
        for i in range(3):
            print(f"--- Round {i + 1} ---")

            message, context_id = await send_message(
                session, 2024, agent_a_id, message, context_id=context_id
            )
            print(f"🔵 Agent A: {message}")

            message, context_id = await send_message(
                session, 2025, agent_b_id, message, context_id=context_id
            )
            print(f"🔴 Agent B: {message}\n")


if __name__ == "__main__":
    asyncio.run(simulate_conversation())

완전한 작동 예시는 다음을 참고하세요:

• 두 LangGraph 에이전트 통신 - A2A 프로토콜을 사용하는 두 LangGraph 에이전트 예시
• Google ADK 에이전트 + LangChain 에이전트 - A2A 프로토콜로 LangChain 에이전트와 상호작용하는 Google ADK 에이전트 예시

분산 트레이싱 (Distributed tracing)

여러 에이전트가 A2A로 통신할 때, LangSmith는 그들의 모든 트레이스를 단일 스레드로 그룹화할 수 있으며, 이는 전체 멀티에이전트 대화에 대한 통합된 보기를 제공합니다.

contextId가 thread_id에 매핑되는 방식

Agent Server A2A 엔드포인트는 LangSmith 트레이싱을 위해 A2A contextId를 자동으로 thread_id로 변환합니다. 이는 추가 구성 없이도 대화의 모든 메시지가 참여하는 모든 에이전트에 걸쳐 LangSmith의 같은 스레드로 그룹화된다는 뜻입니다.

흐름은 다음과 같습니다:

1. 첫 번째 메시지에서 클라이언트는 contextId를 생략합니다. 서버가 생성해 응답에 반환합니다.
2. 클라이언트는 대화 연속성을 유지하기 위해 모든 후속 메시지에 contextId를 전달합니다.
3. Agent Server는 contextId를 LangSmith 메타데이터의 thread_id로 매핑하므로 모든 턴이 같은 스레드에 나타납니다.

여러 에이전트 간 트레이싱

서로 다른 프레임워크의 에이전트가 A2A로 통신할 때 contextId가 그들의 트레이스를 통합합니다. 첫 번째 에이전트가 반환한 contextId를 이후의 모든 요청에서 그 에이전트와 다른 에이전트에게 재사용하세요.

다음 코드 스니펫은 핵심 개념을 보여줍니다. 두 에이전트가 있는 완전한 실행 가능한 구현은 Google ADK + LangChain 예시를 참고하세요.

import asyncio
import aiohttp
import uuid


async def send_message(session, url, text, context_id=None):
    """Send an A2A message and return (response_text, context_id)."""

    # --- 1. Build the message ---
    # On follow-up turns, include contextId inside the message object so the server
    # associates them with the ongoing conversation. Do not resend taskId: each turn
    # opens a new task within that conversation.
    message = {
        "role": "user",
        "parts": [{"kind": "text", "text": text}],
        "messageId": str(uuid.uuid4()),
    }
    if context_id:
        message["contextId"] = context_id

    # --- 2. Send it ---
    # contextId travels inside the message. Agent Server turns it into the
    # LangGraph thread_id, so no separate tracing field is needed.
    payload = {
        "jsonrpc": "2.0",
        "id": str(uuid.uuid4()),
        "method": "message/send",
        "params": {"message": message},
    }

    async with session.post(url, json=payload, headers={"Accept": "application/json"}) as response:
        if response.status != 200:
            raise RuntimeError(f"HTTP {response.status}: {await response.text()}")
        result = await response.json()

    if "error" in result:
        raise RuntimeError(result["error"].get("message", "Unknown error"))

    result_obj = result.get("result", {})
    returned_context_id = result_obj.get("contextId") or context_id
    text_out = next(
        (
            part.get("text", "")
            for art in result_obj.get("artifacts", []) or []
            for part in art.get("parts", []) or []
            if part.get("kind") == "text"
        ),
        "(no text)",
    )
    return text_out, returned_context_id


async def run_conversation(agent_a_url, agent_b_url):
    # --- 3. Share the context across agents ---
    # The first response carries a contextId. Pass it to every agent from then
    # on, and all their traces land in one LangSmith thread.
    context_id = None
    message = "Hello! Let's collaborate."

    async with aiohttp.ClientSession() as session:
        for _ in range(3):
            message, context_id = await send_message(
                session, agent_a_url, message, context_id=context_id
            )

            message, context_id = await send_message(
                session, agent_b_url, message, context_id=context_id
            )


asyncio.run(run_conversation(
    "http://localhost:2024/a2a/<agent_a_assistant_id>",
    "http://localhost:2025/a2a/<agent_b_assistant_id>",
))

1. 메시지 구성: 후속 턴에서는 message 객체 안에 contextId를 포함해 서버가 진행 중인 대화와 연결할 수 있게 하세요. 첫 메시지에서는 생략하세요. 서버가 contextId를 생성해 응답에 반환하기 때문입니다. 끝난 턴의 taskId를 다시 보내지 마세요.

2. 보내기: contextId는 params.message 안에서 이동합니다. Agent Server는 이를 LangGraph thread_id로 사용하므로 설정할 별도의 트레이싱 필드가 없습니다.

3. 에이전트 간 컨텍스트 공유: 첫 에이전트가 contextId를 만들게 한 다음, 대화의 나머지 동안 같은 값을 모든 에이전트에게 전달하세요. 이것이 그들의 트레이스를 한 스레드로 그룹화하는 것입니다.

비-LangGraph 에이전트에서 thread_id 수신하기

이전 섹션은 클라이언트 측 — 메시지 전송 시 contextId 전파 — 을 다룹니다. 에이전트 중 하나가 LangGraph로 구축되지 않았다면 수신 측에서도 그 contextId를 읽어 스레드 식별자로 첨부해야, 그 트레이스가 같은 LangSmith 스레드에 들어갑니다. langsmith.integrations.otel.configure()로 자동 트레이싱을 설정하고 수신 A2A 요청에서 params.message.contextId를 읽으세요.

from fastapi import FastAPI, Request
from langsmith.integrations.otel import configure as configure_otel
from opentelemetry import trace
import json

# --- 1. Configure OTel ---
# Set up automatic tracing to LangSmith for your non-LangGraph agent.
configure_otel(project_name="my-a2a-project")
tracer = trace.get_tracer(__name__)

app = FastAPI()

@app.middleware("http")
async def set_thread_id_middleware(request: Request, call_next):
    thread_id = None
    if request.method == "POST":
        body_bytes = await request.body()
        if body_bytes:
            # --- 2. Extract contextId from the incoming A2A message ---
            try:
                body = json.loads(body_bytes)
                thread_id = body["params"]["message"].get("contextId")
            except (ValueError, KeyError, TypeError):
                pass
            # Re-inject the body so downstream handlers can still read it
            async def receive():
                return {"type": "http.request", "body": body_bytes}
            request._receive = receive

    # --- 3. Attach thread_id to the trace ---
    # langsmith.metadata.thread_id groups this trace with others in the same thread.
    with tracer.start_as_current_span("agent") as span:
        if thread_id:
            span.set_attribute("langsmith.metadata.thread_id", thread_id)
        return await call_next(request)

이 미들웨어 이후에 app에 에이전트 라우트를 등록하세요.

LangSmith에서 트레이스 보기

멀티에이전트 대화를 실행한 후 LangSmith UI을 열고 Threads로 이동하세요. 참여한 모든 에이전트의 모든 턴이 공유된 thread_id로 식별되는 단일 스레드 아래에 나타납니다.

통합 테스트하기 (Test your integration)

자체 배포를 대상으로

카드를 가져온 다음 메시지를 보내세요:

curl "https://your-deployment/a2a/{assistant_id}/.well-known/agent-card.json"

curl -X POST "https://your-deployment/a2a/{assistant_id}" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "SendMessage",
    "params": {
      "message": {
        "role": "ROLE_USER",
        "parts": [{"text": "hello"}],
        "messageId": "test-1"
      }
    }
  }'

응답은 result.task.id와 result.task.contextId를 포함합니다. 다음 메시지에서 contextId를 재사용해 대화를 계속하세요.

스트리밍의 경우 Accept: text/event-stream을 보내고 SendStreamingMessage를 사용하세요. 첫 이벤트는 Task이며, 상태와 산출물 업데이트가 뒤따릅니다.

공식 적합성 스위트를 대상으로

A2A는 a2aproject/a2a-tck에서 Technology Compatibility Kit을 게시합니다. 구현을 RFC 2119 수준으로 등급화하며 여러분 것을 포함한 모든 A2A 엔드포인트에서 작동합니다.

./run_tck.py --sut-host https://your-deployment/a2a/{assistant_id} --transport jsonrpc

Agent Server가 현재 실패하는 것

Agent Server는 모든 CI 빌드에서 TCK를 필수 검사로 실행하며, 체크인된 알려진 실패 목록을 기준으로 게이트합니다. 새 실패가 나타나면 CI가 실패하고, 나열된 요건이 통과하기 시작해도 실패하므로, 목록이 서버가 실제로 하는 것과 어긋날 수 없습니다.

능력에 맞춰 구축하기 전에 이것을 읽으세요:

A2A 비활성화 (Disable A2A)

A2A 엔드포인트를 비활성화하려면 langgraph.json 구성 파일에서 disable_a2a를 true로 설정하세요:

{
  "$schema": "https://langgra.ph/schema.json",
  "http": {
    "disable_a2a": true
  }
}

더 알아보기

• 이 문서를 MCP로 연결하면 Claude, VSCode 등에서 실시간 답변을 받을 수 있어요.
• GitHub에서 이 페이지 편집하기 또는 이슈 제출하기.