Molayo

© 2026 Molayo

Qiita헤드라인2026. 06. 15. 12:08

Google A2A SDK로 멀티 에이전트 통신을 시각화하기

요약

Google의 A2A(Agent-to-Agent) 프로토콜을 사용하여 서로 다른 에이전트 간의 통신을 구현하고 시각화하는 방법을 다룹니다. Agent Card를 통한 서비스 디스커버리와 메시징 과정을 통해 멀티 에이전트 시스템의 상호작용을 모니터링하는 튜토리얼을 제공합니다.

핵심 포인트

  • Google A2A 프로토콜은 서로 다른 프레임워크 간 에이전트 통신을 위한 표준 사양임
  • Agent Card를 통해 에이전트의 능력과 지원 프로토콜을 선언하고 탐색함
  • A2A SDK의 AgentExecutor를 상속받아 동기 또는 비동기 방식으로 에이전트 로직 구현 가능
  • 멀티 에이전트 간의 상호작용을 실시간 시각화하여 디버깅 및 튜닝 효율을 높임

멀티 에이전트 시스템 (Multi-Agent System) 내에서는 여러 에이전트가 움직이고 있습니다.

그중 에이전트 간의 상호작용을 블랙박스로 둔 채 개발을 진행하면,

디버깅/튜닝 (Debugging/Tuning)도 막막해지기 마련입니다.

Google A2A (Agent-to-Agent) 프로토콜을 사용하여 에이전트를 3개 구현하고,

그 통신을 실시간으로 시각화하는 모니터링 툴을 만들어 내부를 확인해 보겠습니다!

먼저, A2A (Agent-to-Agent)는 Google이 2025년에 공개한 오픈 프로토콜로, 서로 다른 프레임워크 및 언어로 구현된 에이전트끼리 표준적인 방법으로 통신하기 위한 사양을 의미합니다.

핵심은 다음 두 가지라고 생각합니다.

Agent Card

각 에이전트가 /.well-known/agent-card.json에서 자신의 능력을 선언합니다. 이름, 스킬, 지원하는 프로토콜 바인딩 (Protocol Binding) (JSON-RPC / REST / gRPC), 스트리밍 대응 가능 여부 등이 적혀 있으며, 이 정보를 바탕으로 상호작용을 수행하게 됩니다.

{
"name": "Writer",
"supportedInterfaces": [{
...

메시징 (Messaging)

말을 걸기 전에 Agent Card를 취득하고 (서비스 디스커버리, Service Discovery), 그 정보에 기반하여 SendMessage를 던지며, 항상 GET /.well-known/agent-card.json

→ POST /

의 2단계로 이루어져 있습니다.

이번에는 스토리를 계획하는 사람, 이야기를 구성하는 사람, 이야기를 평가하는 사람을 만들어 각각의 상호작용을 확인합니다.

Story Planner (8001)
│ story_plan
▼
...

준비할 에이전트는 3개입니다.

Reviewer→Writer→Reviewer 루프를 임의의 턴만큼 반복하고,

마지막에 Writer가 완성 원고를 Story Planner에게 반환합니다.

역할기술
A2A SDKa2a-sdk 1.1.0
...
backend/
agents.py # AgentExecutor 구현 (Planner / Writer / Reviewer)
agent_servers.py # 3개의 A2A 준수 FastAPI 앱 빌드
...

A2A SDK의 AgentExecutor를 상속받아, execute() 메서드에 처리를 작성합니다!

SDK 문서에는 두 가지 패턴이 제시되어 있습니다.

Immediate response—Message를 하나 엔큐 (Enqueue) 하여 반환 (동기적)

Asynchronous—Task를 엔큐하여 비동기적으로 TaskStatusUpdateEvent / TaskArtifactUpdateEvent를 흘려보냄

이번에는 심플하게 Immediate response를 사용했습니다.

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.helpers.proto_helpers import new_text_message
...

Asynchronous 패턴을 사용할 경우, start_work()를 호출하기 전에 반드시 Task 객체 자체를 큐에 쌓아야 합니다. 이를 지키지 않으면 에러가 발생합니다.

각 에이전트를 독립된 FastAPI 앱으로 기동합니다.

DefaultRequestHandlerV2가 JSON-RPC 라우팅을 모두 담당합니다.

from a2a.server.request_handlers import DefaultRequestHandlerV2
from a2a.server.tasks import InMemoryTaskStore
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
...

4개의 서버를 1개 프로세스에서 구동하기 위해, asyncio.gather로 병렬 기동했습니다.

async def main():
    await asyncio.gather(
        serve(monitor, 8000), # 모니터링 API
...

A2A 통신을 관측하기 위해 httpx.AsyncBaseTransport를 래핑(wrapping)하는 커스텀 트랜스포트(custom transport)를 구현했습니다.

class CapturingTransport(httpx.AsyncBaseTransport):
    def __init__(self, wrapped: httpx.AsyncBaseTransport):
        self._wrapped = wrapped
...

스트림(stream)은 한 번 읽으면 끝이기 때문에, 그대로 전달하면 호출 측에서 바디(body)가 비어 있게 됩니다.

response.aread()로 바디를 소비한 후, 동일한 바이트 열(byte sequence)로 httpx.Response를 재구성하여 반환합니다.

이를 통해 각 통신의 실제 JSON-RPC 페이로드(payload)가 프론트엔드까지 전달될 수 있습니다.

UI는 다음과 같습니다.

왼쪽: 통신 선택

중앙: mermaid를 사용하여 전체 흐름을 시각화

오른쪽: 왼쪽에서 선택한 통신 내용을 상세히 표시

SSE 엔드포인트(endpoint)는 기존 메시지를 리플레이(replay)한 후 신규 메시지를 스트리밍하도록 되어 있습니다.

영역역할
Agent GraphSVG로 에이전트를 노드(node)로 표시. 통신 발생 시 에지(edge)를 하이라이트 + 애니메이션
...

페이즈 1: 서비스 디스커버리 (Service Discovery, 루프 시작 전 1회)

가장 먼저 Agent Card를 가져와(서비스 디스커버리), 에이전트의 능력을 확인합니다.

GET http://localhost:8001/.well-known/agent-card.json → 200
GET http://localhost:8002/.well-known/agent-card.json → 200
GET http://localhost:8003/.well-known/agent-card.json → 200

페이즈 2: 실행 (턴마다 POST)

서로 정보를 주고받습니다.

POST http://localhost:8002/ → 200 # Story Planner → Writer
POST http://localhost:8003/ → 200 # Writer → Reviewer
POST http://localhost:8002/ → 200 # Reviewer → Writer
...

하나를 발췌하면 다음과 같습니다.

{
  "method": "SendMessage",
  "params": {
    ...
  }
}

{
  "result": {
    "message": {
      ...
    }
  }
}

JSON-RPC 2.0 표준 포맷 위에 A2A의 메시지 구조(role / parts / messageId)가 얹혀 있는 형태입니다.

1. Agent Card의 capabilities.streaming: true와 실제 동작은 별개

Agent Card에는 "streaming": true라고 선언되어 있습니다.

def _make_card(name: str, description: str, port: int, skill_id: str) -> AgentCard:
    ...
    capabilities=AgentCapabilities(streaming=True),
    ...

하지만 클라이언트 측에서 ClientConfig(streaming=False)를 설정하고 있습니다.

실제로는 SendMessage(비스트리밍)로 통신하고 있으며, 이를 통해 클라이언트 측 설정이 우선시되고 능력 선언과 실제 사용 방식은 분리되어 있음을 알 수 있습니다.

2. 제어는 에이전트가 아닌 오케스트레이터(orchestrator)가 가짐

각 에이전트는 들어온 요청에 응답할 뿐, "다음에 누구에게 보낼지"는 알지 못합니다.

통신 순서와 라우팅(routing)은 orchestrator.py가 전부 제어하고 있으며, A2A는 통신의 표준화 프로토콜일 뿐 에이전트의 자율성을 보장하는 것은 아니라는 뜻입니다.

3. 통신은 직렬(serial)

POST 응답이 올 때까지 다음을 던지지 않는 설계이므로, 이번 8개의 메시지는 완전히 순차적(sequential)으로 실행되었습니다만,

오케스트레이터 (Orchestrator) 측의 변경을 통해 병렬화도 가능하며, 프로토콜 (Protocol) 레벨에서는 병렬 통신을 방해하는 요소가 없었습니다.

4. A2A는 어디까지나 통신 레이어 (Communication Layer)

현재의 Writer는 피드백 내용을 무시하고 무작위로 문장을 생성하고 있습니다.

이는 LLM을 사용하지 않고 있기 때문이지만, 실제 시스템에서는 context.get_user_input()을 통해 받은 텍스트를 LLM의 프롬프트 (Prompt)로 전달합니다. A2A는 통신 레이어의 책임을 담당할 뿐, 에이전트 (Agent)의 로직은 완전히 구현자에게 맡겨져 있는 셈입니다.

A2A는 HTTP 위에 올라가는 프로토콜 레이어 (Protocol Layer)로, 에이전트가 '무엇을 생각하는가'가 아니라 '어떻게 말하는가'를 표준화하는 것입니다.

이번에는 모니터링 툴 (Monitoring Tool)을 만들어 그 대화 방식을 시각화했습니다.

프로토콜 (Protocol, HTTP)을 확인함으로써 Agent Card의 취득부터 SendMessage의 JSON-RPC 페이로드 (Payload)까지, 'A2A란 무엇인가'를 코드를 읽지 않고도 체감할 수 있었습니다.

LLM을 결합하여 진정한 에이전트 루프 (Agent Loop)로 만들면, 에이전트가 무엇을 어떻게 판단했는가라는 별도의 관점이 생기므로 그 부분도 확인해보고 싶습니다.

AI 자동 생성 콘텐츠

본 콘텐츠는 Qiita AI의 원문을 AI가 자동으로 요약·번역·분석한 것입니다. 원 저작권은 원저작자에게 있으며, 정확한 내용은 반드시 원문을 확인해 주세요.

원문 바로가기
0

댓글

0