Agent2Agent (A2A)는 대화형 AI 에이전트 간의 통신을 가능하게 하는 Google의 프로토콜입니다. LangSmith는 A2A 지원을 구현하여, 표준화된 프로토콜을 통해 에이전트가 다른 A2A 호환 에이전트와 통신할 수 있도록 합니다. A2A endpoint는 LangGraph Server의 /a2a/{assistant_id}에서 사용할 수 있습니다.

​
Agent Card Discovery

각 assistant는 자동으로 기능을 설명하고 다른 에이전트가 연결하는 데 필요한 정보를 제공하는 A2A Agent Card를 노출합니다. 다음을 사용하여 모든 assistant의 agent card를 검색할 수 있습니다: 
GET /.well-known/agent-card.json?assistant_id={assistant_id}
agent card에는 assistant의 이름, 설명, 사용 가능한 skill, 지원되는 입출력 모드, 통신을 위한 A2A endpoint URL이 포함됩니다.

​
Requirements

A2A를 사용하려면 다음 dependency가 설치되어 있는지 확인하세요:
  • langgraph-api >= 0.4.9
다음 명령으로 설치: 
pip install "langgraph-api>=0.4.9"

​
사용 개요

A2A를 활성화하려면:
  • langgraph-api>=0.4.9를 사용하도록 업그레이드합니다.
  • 메시지 기반 state 구조로 에이전트를 배포합니다.
  • endpoint를 사용하여 다른 A2A 호환 에이전트와 연결합니다.

​
A2A 호환 에이전트 만들기

이 예제는 OpenAI의 API를 사용하여 수신 메시지를 처리하고 대화 state를 유지하는 A2A 호환 에이전트를 만듭니다. 에이전트는 메시지 기반 state 구조를 정의하고 A2A 프로토콜의 메시지 형식을 처리합니다. A2A “text” parts와 호환되려면, 에이전트는 state에 messages key를 가져야 합니다. 다음은 예제입니다: 
"""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 통신

에이전트가 langgraph dev를 통해 로컬에서 실행되거나 프로덕션에 배포되면, A2A 프로토콜을 사용하여 에이전트 간의 통신을 촉진할 수 있습니다. 이 예제는 두 에이전트가 서로의 A2A endpoint에 JSON-RPC 메시지를 전송하여 통신하는 방법을 보여줍니다. 스크립트는 각 에이전트가 상대방의 응답을 처리하고 대화를 계속하는 다중 턴 대화를 시뮬레이션합니다. 
#!/usr/bin/env python3
"""Agent-to-Agent conversation simulation using LangGraph A2A protocol."""

import asyncio
import aiohttp
import os

async def send_message(session, port, assistant_id, text):
    """Send a message to an agent and return the response text."""
    url = f"http://127.0.0.1:{port}/a2a/{assistant_id}"
    payload = {
        "jsonrpc": "2.0",
        "id": "",
        "method": "message/send",
        "params": {
            "message": {
                "role": "user",
                "parts": [{"kind": "text", "text": text}]
            },
            "messageId": "",
            "thread": {"threadId": ""}
        }
    }

    headers = {"Accept": "application/json"}
    async with session.post(url, json=payload, headers=headers) as response:
        try:
            result = await response.json()
            return result["result"]["artifacts"][0]["parts"][0]["text"]
        except Exception as e:
            text = await response.text()
            print(f"Response error from port {port}: {response.status} - {text}")
            return f"Error from port {port}: {response.status}"

async def simulate_conversation():
    """Simulate a conversation between two agents."""
    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."

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

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

            # Agent B responds
            message = await send_message(session, 2025, agent_b_id, message)
            print(f"🔴 Agent B: {message}")
            print()

if __name__ == "__main__":
    asyncio.run(simulate_conversation())
Edit the source of this page on GitHub.
Connect these docs programmatically to Claude, VSCode, and more via MCP for real-time answers.
⌘I