728x90
이 글은 Arjun Prabhulal님의 동의 하에 그분의 깊이 있는 분석과 자료를 사용하여 작성했습니다. 원문은 [Building Multi-Agent App with Google's A2A(Agent2Agent) Protocol, ADK, and MCP]에서 확인하실 수 있습니다. 독자 여러분께 더욱 유익한 정보를 전달할 수 있도록 귀한 내용을 공유해주신 Arjun Prabhulal님께 다시 한번 감사드립니다.
지난번 글에서는 구글의 ADK(Agent Development Kit)와 MCP(Model Context Protocol)를 살펴보았는데요.
Google Cloud Next '25에서 구글은 ADK와 더불어 A2A(Agent-2-Agent) 프로토콜을 발표했습니다. A2A는 AI 에이전트들이 조직의 경계를 넘어 서로 소통하고, 협업하며, 행동을 조율할 수 있도록 지원하는 강력한 개방형 표준입니다. A2A는 기반이 되는 에이전트 프레임워크에 상관없이 다양한 엔터프라이즈 시스템 상에서 안전하고 구조화된 에이전트 통신을 가능하게 합니다.
잠깐! 지난 내용 복습
본격적인 내용에 앞서, 이전 글에서 다뤘던 핵심 개념들을 간단히 짚고 넘어가겠습니다.
- MCP (Model Context Protocol): MCP는 기업의 API를 도구에 구애받지 않고 LLM 친화적인 방식으로 만들어주는 프로토콜입니다. 기존 API 통합 방식의 한계를 극복하고, 다양한 모델에서 외부 도구와 API를 원활하게 사용할 수 있도록 돕습니다.
- ADK (Agent Development Kit): 구글의 오픈소스 툴킷으로, 다양한 유형의 AI 에이전트를 개발하고, MCP 클라이언트로서 외부 도구와 연동하는 과정을 간소화합니다.
오늘은 이 두 가지 기술 위에 A2A 프로토콜을 더해, 더욱 강력하고 유연한 멀티 에이전트 시스템을 구축하는 여정을 떠나보겠습니다.
A2A 프로토콜이란 무엇일까요?
A2A(Agent-to-Agent) 프로토콜은 구글이 개발한 오픈 프로토콜로, 조직이나 기술의 경계를 넘어 AI 에이전트 간의 소통을 가능하게 합니다.
A2A를 사용하면 에이전트들은 메모리, 생각, 도구를 직접 공유하지 않고도 컨텍스트, 상태, 지침, 데이터를 교환하며 최종 사용자를 위한 작업을 수행할 수 있습니다. 마치 서로 다른 부서의 전문가들이 각자의 전문성을 바탕으로 협력하는 것과 비슷하죠.
A2A의 주요 특징
- 단순성: 기존 표준을 재사용하여 쉽게 적용할 수 있습니다.
- 엔터프라이즈 준비: 인증, 보안, 개인 정보 보호, 추적, 모니터링 등 기업 환경에 필요한 요소를 갖췄습니다.
- 비동기 우선: 매우 긴 작업이나 사람의 개입이 필요한 경우에도 유연하게 대응합니다.
- 다양한 형식 지원: 텍스트, 오디오/비디오, 양식, iframe 등 다양한 데이터 형식을 지원합니다.
- 불투명한 실행: 에이전트는 자신의 내부 계획이나 도구를 공유할 필요 없이 상호 작용합니다.
A2A vs MCP - 핵심 차이점
- MCP: 기업 API를 LLM 친화적인 도구로 만듭니다. (API for tools)
- A2A: ADK, LangChain, CrewAI 등 다양한 프레임워크로 구축된 분산 에이전트 간의 원활한 대화를 가능하게 합니다. (API for agents)
A2A 프로토콜의 핵심 개념
A2A 프로토콜은 여러 핵심 개념을 바탕으로 에이전트 간의 원활한 상호 운용성을 구현합니다. 멀티 에이전트 여행 플래너 애플리케이션 예시를 통해 쉽게 이해해 봅시다.
1. 작업 기반 통신 (Task-based Communication)
모든 상호 작용은 명확한 시작과 끝이 있는 '작업' 단위로 처리됩니다. 이를 통해 통신이 구조화되고 추적 가능해집니다.
Task-based Communication (Request)
{
"jsonrpc": "2.0",
"method": "tasks/send",
"params": {
"taskId": "20250615123456",
"message": {
"role": "user",
"parts": [
{
"text": "Find flights from New York to Miami on 2025-06-15"
}
]
}
},
"id": "12345"
}
Task-based Communication (Response)
{
"jsonrpc": "2.0",
"id": "12345",
"result": {
"taskId": "20250615123456",
"state": "completed",
"messages": [
{
"role": "user",
"parts": [
{
"text": "Find flights from New York to Miami on 2025-06-15"
}
]
},
{
"role": "agent",
"parts": [
{
"text": "I found the following flights from New York to Miami on June 15, 2025:\n\n1. Delta Airlines DL1234: Departs JFK 08:00, Arrives MIA 11:00, $320\n2. American Airlines AA5678: Departs LGA 10:30, Arrives MIA 13:30, $290\n3. JetBlue B9101: Departs JFK 14:45, Arrives MIA 17:45, $275"
}
]
}
],
"artifacts": []
}
}
2. 에이전트 검색 (Agent Discovery)
에이전트는 표준 위치(/.well-known/agent.json)에 있는 agent.json 파일을 읽어 다른 에이전트의 기능을 자동으로 발견할 수 있습니다. 수동 설정이 필요 없죠. 아래는 에이전트 카드 예시입니다.
- 목적: 공개적으로 접근 가능한 검색 엔드포인트
- 위치: A2A 프로토콜에 따른 표준화된 웹 접근 가능 경로
- 용도: 다른 에이전트/클라이언트에 의한 외부 검색
- 역할: 자동 에이전트 검색 활성화
{
"name": "Travel Itinerary Planner",
"displayName": "Travel Itinerary Planner",
"description": "An agent that coordinates flight and hotel information to create comprehensive travel itineraries",
"version": "1.0.0",
"contact": "code.aicloudlab@gmail.com",
"endpointUrl": "http://localhost:8005",
"authentication": {
"type": "none"
},
"capabilities": [
"streaming"
],
"skills": [
{
"name": "createItinerary",
"description": "Create a comprehensive travel itinerary including flights and accommodations",
"inputs": [
{
"name": "origin",
"type": "string",
"description": "Origin city or airport code"
},
{
"name": "destination",
"type": "string",
"description": "Destination city or area"
},
{
"name": "departureDate",
"type": "string",
"description": "Departure date in YYYY-MM-DD format"
},
{
"name": "returnDate",
"type": "string",
"description": "Return date in YYYY-MM-DD format (optional)"
},
{
"name": "travelers",
"type": "integer",
"description": "Number of travelers"
},
{
"name": "preferences",
"type": "object",
"description": "Additional preferences like budget, hotel amenities, etc."
}
],
"outputs": [
{
"name": "itinerary",
"type": "object",
"description": "Complete travel itinerary with flights, hotels, and schedule"
}
]
}
]
}
3. 프레임워크 독립적 상호 운용성 (Framework-agnostic Interoperability)
A2A는 ADK, CrewAI, LangChain 등 다양한 에이전트 프레임워크에서 작동하므로, 서로 다른 도구로 만들어진 에이전트도 함께 작업할 수 있습니다.
4. 다중 모드 메시징 (Multi-modal Messaging):
Parts 시스템을 통해 텍스트, 구조화된 데이터, 파일 등 다양한 콘텐츠 유형을 교환할 수 있습니다.
5. 표준화된 메시지 구조 (Standardized Message Structure)
깔끔한 JSON-RPC 스타일을 사용하여 메시지를 주고받으므로, 구현이 일관되고 파싱이 쉽습니다.
6. 기술 및 기능 (Skills and Capabilities)
에이전트는 자신이 할 수 있는 일('기술')과 필요한 입력, 제공하는 출력을 게시하여 다른 에이전트가 자신과 상호 작용하는 방법을 알 수 있도록 합니다.
// Skill declaration in agent card
"skills": [
{
"name": "createItinerary",
"description": "Creates a travel itinerary",
"inputs": [
{"name": "origin", "type": "string", "description": "Origin city or airport"},
{"name": "destination", "type": "string", "description": "Destination city or airport"},
{"name": "departureDate", "type": "string", "description": "Date of departure (YYYY-MM-DD)"},
{"name": "returnDate", "type": "string", "description": "Date of return (YYYY-MM-DD)"}
]
}
]
// Skill invocation in a message
{
"role": "user",
"parts": [
{
"text": "Create an itinerary for my trip from New York to Miami"
},
{
"type": "data",
"data": {
"skill": "createItinerary",
"parameters": {
"origin": "New York",
"destination": "Miami",
"departureDate": "2025-06-15",
"returnDate": "2025-06-20"
}
}
}
]
}
7. 작업 수명 주기 (Task Lifecycle)
각 작업은 제출됨 → 작업 중 → 완료됨 (또는 실패/취소됨)과 같이 명확하게 정의된 단계를 거칩니다. 언제든지 작업 상태를 추적할 수 있습니다.
8. 실시간 업데이트 (Streaming)
장시간 실행되는 작업은 SSE(Server-Sent Events)를 사용하여 업데이트를 스트리밍할 수 있어, 실시간으로 진행 상황을 받을 수 있습니다.
9. 푸시 알림 (Push Notifications)
에이전트는 웹훅을 사용하여 작업 업데이트를 다른 에이전트에게 능동적으로 알릴 수 있으며, JWT, OAuth 등 안전한 통신을 지원합니다.
10. 구조화된 양식 (Structured Forms)
DataPart를 사용하여 구조화된 양식을 요청하거나 제출할 수 있어 JSON이나 설정 같은 입력을 쉽게 처리할 수 있습니다.
아키텍처
이제 A2A와 MCP를 활용한 여행 플래너 아키텍처를 살펴보겠습니다.
아래 데모는 여러 에이전트 간의 A2A 프로토콜 이해를 돕기 위한 설명용 예시입니다.
이 시스템은 각기 다른 역할을 하는 여러 에이전트가 A2A 프로토콜을 통해 통신하는 모듈식 구조입니다.
핵심 구성 요소
- 사용자 인터페이스 (UI) 계층: Streamlit UI를 통해 사용자 요청을 받습니다.
- 에이전트 계층: 호스트 에이전트(여정 플래너)와 전문 에이전트(항공편, 호텔) 간의 조정을 담당합니다.
- 프로토콜 계층: A2A 프로토콜을 통해 에이전트 간 통신이 이루어집니다.
- 외부 데이터 계층: MCP를 사용하여 외부 API에 접근합니다.
에이전트 역할
- 여정 플래너 에이전트 (호스트 에이전트): 중앙 오케스트레이터 역할을 하며, 사용자와 전문 에이전트 간의 상호 작용을 조율합니다.
- 항공편 검색 에이전트 (에이전트 1): 사용자 입력에 따라 항공편 옵션을 가져오는 전담 에이전트입니다.
- 호텔 검색 에이전트 (에이전트 2): 사용자 선호도에 맞는 호텔 숙박 시설을 가져오는 전담 에이전트입니다.
MCP 구현
- 항공편 검색 MCP 서버: 항공편 검색 에이전트가 연결하여 항공편 예약 API 및 데이터베이스와 통신합니다.
- 호텔 검색 MCP 서버: 호텔 검색 에이전트가 연결하여 호텔 예약 시스템 및 애그리게이터와 통신합니다.
통신 흐름
- 사용자가 UI를 통해 여행 문의를 제출합니다.
- 여정 플래너가 문의를 분석하여 핵심 정보를 추출합니다.
- 여정 플래너가 항공편 검색 에이전트에게 항공편 정보를 요청합니다.
- 항공편 검색 에이전트는 MCP 서버를 호출하여 항공편 정보를 반환합니다.
- 여정 플래너가 목적지 정보를 추출합니다.
- 여정 플래너가 호텔 검색 에이전트에게 호텔 정보를 요청합니다.
- 호텔 검색 에이전트가 MCP 서버를 호출하여 숙박 옵션을 반환합니다.
- 여정 플래너가 모든 데이터를 종합하여 포괄적인 여행 일정을 생성합니다.
구현 단계별 가이드
이제 ADK, MCP, Gemini AI를 사용하여 이 멀티 에이전트 시스템을 구축하는 과정을 단계별로 살펴보겠습니다.
사전 준비
- Python 3.11+ 설치
- Google Gemini Generative AI API 키
- 유효한 SerpAPI 키
- 유효한 OpenAI GPT 키
프로젝트 폴더 구조
├── common
│ ├── __init__.py
│ ├── client
│ │ ├── __init__.py
│ │ ├── card_resolver.py
│ │ └── client.py
│ ├── server
│ │ ├── __init__.py
│ │ ├── server.py
│ │ ├── task_manager.py
│ │ └── utils.py
│ ├── types.py
│ └── utils
│ ├── in_memory_cache.py
│ └── push_notification_auth.py
├── flight_search_app
│ ├── a2a_agent_card.json
│ ├── agent.py
│ ├── main.py
│ ├── static
│ │ └── .well-known
│ │ └── agent.json
│ └── streamlit_ui.py
├── hotel_search_app
│ ├── README.md
│ ├── a2a_agent_card.json
│ ├── langchain_agent.py
│ ├── langchain_server.py
│ ├── langchain_streamlit.py
│ ├── static
│ │ └── .well-known
│ │ └── agent.json
│ └── streamlit_ui.py
└── itinerary_planner
├── __init__.py
├── a2a
│ ├── __init__.py
│ └── a2a_client.py
├── a2a_agent_card.json
├── event_log.py
├── itinerary_agent.py
├── itinerary_server.py
├── run_all.py
├── static
│ └── .well-known
│ └── agent.json
└── streamlit_ui.py
1단계: 가상 환경 설정 및 종속성 설치
# 가상 환경 생성
python -m venv .venv
# 가상 환경 활성화
source .venv/bin/activate
# 종속성 설치
pip install fastapi uvicorn streamlit httpx python-dotenv pydantic
pip install google-generativeai google-adk langchain langchain-openai
2단계: MCP 서버 패키지 설치
mcp hotel server —https://pypi.org/project/mcp-hotel-search/
mcp flight server — https://pypi.org/project/mcp-hotel-search/
# MCP 호텔 검색 서버 설치
pip install mcp-hotel-search
# MCP 항공편 검색 서버 설치
pip install mcp-flight-search
3단계: 환경 변수 설정
사전 준비에서 얻은 API 키들을 환경 변수로 설정합니다.
GOOGLE_API_KEY=your_google_api_key
OPENAI_API_KEY=your_openai_api_key
SERP_API_KEY=your_serp_api_key
4단계: 항공편 검색 에이전트 설정 (ADK + MCP + Gemini 2.0 Flash)
아래 설정은 ADK를 사용한 이전 글과 동일하지만, A2A 프로토콜이 추가되었고 https://github.com/google/A2A/tree/main/samples/python/common 의 재사용 가능한 모듈을 사용합니다.
├── common/ # Shared A2A protocol components
│ ├── __init__.py
│ ├── client/ # Client implementations
│ │ ├── __init__.py
│ │ └── client.py # Base A2A client
│ ├── server/ # Server implementations
│ │ ├── __init__.py
│ │ ├── server.py # A2A Server implementation
│ │ └── task_manager.py # Task management utilities
│ └── types.py # Shared type definitions for A2A
여기서는 ADK를 MCP 클라이언트로 사용하고, A2A 프로토콜을 추가합니다. 공유 A2A 모듈을 활용합니다.
4.1 MCP 서버에서 도구를 가져오도록 ADK 에이전트를 구현합니다.
from google.adk.agents.llm_agent import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
..
..
# Fetch tools from MCP Server
server_params = StdioServerParameters(
command="mcp-flight-search",
args=["--connection_type", "stdio"],
env={"SERP_API_KEY": serp_api_key},)
tools, exit_stack = await MCPToolset.from_server(
connection_params=server_params)
..
..
4.2 공통 A2A 서버 구성 요소를 사용하여 ADK 서버 진입점을 정의한다.
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.agents import Agent
from .agent import get_agent_async
# Import common A2A server components and types
from common.server.server import A2AServer
from common.server.task_manager import InMemoryTaskManager
from common.types import (
AgentCard,
SendTaskRequest,
SendTaskResponse,
Task,
TaskStatus,
Message,
TextPart,
TaskState,
)
# --- Custom Task Manager for Flight Search ---
class FlightAgentTaskManager(InMemoryTaskManager):
"""Task manager specific to the ADK Flight Search agent."""
def __init__(self, agent: Agent, runner: Runner, session_service: InMemorySessionService):
super().__init__()
self.agent = agent
self.runner = runner
self.session_service = session_service
logger.info("FlightAgentTaskManager initialized.")
...
...
4.3 에이전트 카드를 포함하여 A2A 서버 인스턴스를 생성합니다.
# --- Main Execution Block ---
async def run_server():
"""Initializes services and starts the A2AServer."""
logger.info("Starting Flight Search A2A Server initialization...")
session_service = None
exit_stack = None
try:
session_service = InMemorySessionService()
agent, exit_stack = await get_agent_async()
runner = Runner(
app_name='flight_search_a2a_app',
agent=agent,
session_service=session_service,
)
# Create the specific task manager
task_manager = FlightAgentTaskManager(
agent=agent,
runner=runner,
session_service=session_service
)
# Define Agent Card
port = int(os.getenv("PORT", "8000"))
host = os.getenv("HOST", "localhost")
listen_host = "0.0.0.0"
agent_card = AgentCard(
name="Flight Search Agent (A2A)",
description="Provides flight information based on user queries.",
url=f"http://{host}:{port}/",
version="1.0.0",
defaultInputModes=["text"],
defaultOutputModes=["text"],
capabilities={"streaming": False},
skills=[
{
"id": "search_flights",
"name": "Search Flights",
"description": "Searches for flights based on origin, destination, and date.",
"tags": ["flights", "travel"],
"examples": ["Find flights from JFK to LAX tomorrow"]
}
]
)
# Create the A2AServer instance
a2a_server = A2AServer(
agent_card=agent_card,
task_manager=task_manager,
host=listen_host,
port=port
)
# Configure Uvicorn programmatically
config = uvicorn.Config(
app=a2a_server.app, # Pass the Starlette app from A2AServer
host=listen_host,
port=port,
log_level="info"
)
server = uvicorn.Server(config)
...
...
4.4 flight search app 시작
5단계: 호텔 검색 에이전트 설정 (LangChain + MCP + OpenAI)
LangChain을 MCP 클라이언트로, OpenAI(GPT-4o)를 LLM으로 사용합니다.
├── hotel_search_app/ # Hotel Search Agent (Agent 2)
│ ├── __init__.py
│ ├── a2a_agent_card.json # Agent capabilities declaration
│ ├── langchain_agent.py # LangChain agent implementation
│ ├── langchain_server.py # Server entry point
│ └── static/ # Static files
│ └── .well-known/ # Agent discovery directory
│ └── agent.json # Standardized agent discovery file
5.1 LangChain 에이전트를 MCP 클라이언트로 구현합니다.
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_mcp_adapters.client import MultiServerMCPClient
# MCP client configuration
MCP_CONFIG = {
"hotel_search": {
"command": "mcp-hotel-search",
"args": ["--connection_type", "stdio"],
"transport": "stdio",
"env": {"SERP_API_KEY": os.getenv("SERP_API_KEY")},
}
}
class HotelSearchAgent:
"""Hotel search agent using LangChain MCP adapters."""
def __init__(self):
self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
def _create_prompt(self):
"""Create a prompt template with our custom system message."""
system_message = """You are a helpful hotel search assistant.
"""
return ChatPromptTemplate.from_messages([
("system", system_message),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
..
..
async def process_query(self, query):
...
# Create MCP client for this query
async with MultiServerMCPClient(MCP_CONFIG) as client:
# Get tools from this client instance
tools = client.get_tools()
# Create a prompt
prompt = self._create_prompt()
# Create an agent with these tools
agent = create_openai_functions_agent(
llm=self.llm,
tools=tools,
prompt=prompt
)
# Create an executor with these tools
executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
handle_parsing_errors=True,
)
5.2 공통 A2A 서버 구성 요소를 사용하여 A2A 서버 인스턴스를 생성합니다.
# Use the underlying agent directly
from hotel_search_app.langchain_agent import get_agent, HotelSearchAgent
# Import common A2A server components and types
from common.server.server import A2AServer
from common.server.task_manager import InMemoryTaskManager
from common.types import (
AgentCard,
SendTaskRequest,
SendTaskResponse,
Task,
TaskStatus,
Message,
TextPart,
TaskState
)
..
..
class HotelAgentTaskManager(InMemoryTaskManager):
"""Task manager specific to the Hotel Search agent."""
def __init__(self, agent: HotelSearchAgent):
super().__init__()
self.agent = agent # The HotelSearchAgent instance
logger.info("HotelAgentTaskManager initialized.")
async def on_send_task(self, request: SendTaskRequest) -> SendTaskResponse:
"""Handles the tasks/send request by calling the agent's process_query."""
task_params = request.params
task_id = task_params.id
user_message_text = None
logger.info(f"HotelAgentTaskManager handling task {task_id}")
# --- Main Execution Block ---
async def run_server():
"""Initializes services and starts the A2AServer for hotels."""
logger.info("Starting Hotel Search A2A Server initialization...")
agent_instance: Optional[HotelSearchAgent] = None
try:
agent_instance = await get_agent()
if not agent_instance:
raise RuntimeError("Failed to initialize HotelSearchAgent")
# Create the specific task manager
task_manager = HotelAgentTaskManager(agent=agent_instance)
# Define Agent Card
port = int(os.getenv("PORT", "8003")) # Default port 8003
host = os.getenv("HOST", "localhost")
listen_host = "0.0.0.0"
agent_card = AgentCard(
name="Hotel Search Agent (A2A)",
description="Provides hotel information based on location, dates, and guests.",
url=f"http://{host}:{port}/",
version="1.0.0",
defaultInputModes=["text"],
defaultOutputModes=["text"],
capabilities={"streaming": False},
skills=[
{
"id": "search_hotels",
"name": "Search Hotels",
"description": "Searches for hotels based on location, check-in/out dates, and number of guests.",
"tags": ["hotels", "travel", "accommodation"],
"examples": ["Find hotels in London from July 1st to July 5th for 2 adults"]
}
]
)
# Create the A2AServer instance WITHOUT endpoint parameter
a2a_server = A2AServer(
agent_card=agent_card,
task_manager=task_manager,
host=listen_host,
port=port
)
config = uvicorn.Config(
app=a2a_server.app, # Pass the Starlette app from A2AServer
host=listen_host,
port=port,
log_level="info"
)
5.3 MCP 서버를 호출하는 진입점으로서 호텔 검색 앱(Langchain)을 시작해 봅시다.
6단계: A2A 프로토콜을 사용하여 에이전트 간의 오케스트레이터(Orchestrator)로서 호스트 에이전트 구현하기
여정 플래너는 항공편 및 호텔 서비스 간의 A2A 프로토콜 통신을 조율하는 중앙 구성 요소입니다.
├── itinerary_planner/ # Travel Planner Host Agent (Agent 3)
│ ├── __init__.py
│ ├── a2a/ # A2A client implementations
│ │ ├── __init__.py
│ │ └── a2a_client.py # Clients for flight and hotel agents
│ ├── a2a_agent_card.json # Agent capabilities declaration
│ ├── event_log.py # Event logging utilities
│ ├── itinerary_agent.py # Main planner implementation
│ ├── itinerary_server.py # FastAPI server
│ ├── run_all.py # Script to run all components
│ ├── static/ # Static files
│ │ └── .well-known/ # Agent discovery directory
│ │ └── agent.json # Standardized agent discovery file
│ └── streamlit_ui.py # Main user interface
6.1 항공편 및 호텔 API URL을 사용한 A2A 프로토콜 구현
- 다른 서비스와 통신하기 위한 클라이언트 코드를 포함합니다.
- A2A(Agent-to-Agent) 프로토콜을 구현합니다.
- 항공편 및 호텔 검색 서비스를 호출하기 위한 모듈을 가집니다.
# Base URLs for the A2A compliant agent APIs
FLIGHT_SEARCH_API_URL = os.getenv("FLIGHT_SEARCH_API_URL", "http://localhost:8000")
HOTEL_SEARCH_API_URL = os.getenv("HOTEL_SEARCH_API_URL", "http://localhost:8003")
class A2AClientBase:
"""Base client for communicating with A2A-compliant agents via the root endpoint."""
async def send_a2a_task(self, user_message: str, task_id: Optional[str] = None, agent_type: str = "generic") -> Dict[str, Any]:
...
....
# Construct the JSON-RPC payload with the A2A method and corrected params structure
payload = {
"jsonrpc": "2.0",
"method": "tasks/send",
"params": {
"id": task_id,
"taskId": task_id,
"message": {
"role": "user",
"parts": [
{"type": "text", "text": user_message}
]
}
},
"id": task_id
}
6.2 Itinerary Planner Agent Card
에이전트의 기능, 엔드포인트, 인증 요구 사항 및 기술을 설명하는 JSON 메타데이터 파일입니다. A2A 프로토콜에서 서비스 검색에 사용됩니다.
{
"name": "Travel Itinerary Planner",
"displayName": "Travel Itinerary Planner",
"description": "An agent that coordinates flight and hotel information to create comprehensive travel itineraries",
"version": "1.0.0",
"contact": "code.aicloudlab@gmail.com",
"endpointUrl": "http://localhost:8005",
"authentication": {
"type": "none"
},
"capabilities": [
"streaming"
],
"skills": [
{
"name": "createItinerary",
"description": "Create a comprehensive travel itinerary including flights and accommodations",
"inputs": [
{
"name": "origin",
"type": "string",
"description": "Origin city or airport code"
},
{
"name": "destination",
"type": "string",
"description": "Destination city or area"
},
{
"name": "departureDate",
"type": "string",
"description": "Departure date in YYYY-MM-DD format"
},
{
"name": "returnDate",
"type": "string",
"description": "Return date in YYYY-MM-DD format (optional)"
},
{
"name": "travelers",
"type": "integer",
"description": "Number of travelers"
},
{
"name": "preferences",
"type": "object",
"description": "Additional preferences like budget, hotel amenities, etc."
}
],
"outputs": [
{
"name": "itinerary",
"type": "object",
"description": "Complete travel itinerary with flights, hotels, and schedule"
}
]
}
]
}
6.3 Google-GenAI SDK를 사용하는 Itinerary agent
데모를 간단하게 유지하기 위해 GenAI SDK를 사용합니다 (ADK, CrewAI 또는 다른 프레임워크도 사용 가능합니다).
Itinerary agent는 시스템의 중앙 호스트 에이전트 역할을 하며, 자연어 요청을 파싱하기 위해 언어 모델을 사용하여 항공편 및 호텔 검색 서비스와의 통신을 조율합니다.
import google.generativeai as genai # Use direct SDK
..
..
from itinerary_planner.a2a.a2a_client import FlightSearchClient, HotelSearchClient
# Configure the Google Generative AI SDK
genai.configure(api_key=api_key)
class ItineraryPlanner:
"""A planner that coordinates between flight and hotel search agents to create itineraries using the google.generativeai SDK."""
def __init__(self):
"""Initialize the itinerary planner."""
logger.info("Initializing Itinerary Planner with google.generativeai SDK")
self.flight_client = FlightSearchClient()
self.hotel_client = HotelSearchClient()
# Create the Gemini model instance using the SDK
self.model = genai.GenerativeModel(
model_name="gemini-2.0-flash",
)
..
..
6.4 Itinerary Server — itinerary planner를 위한 엔드포인트를 노출하는 FastAPI 서버로, 들어오는 HTTP 요청을 처리하고 요청을 Itinerary Agent로 라우팅합니다.
from fastapi import FastAPI, HTTPException, Request
from itinerary_planner.itinerary_agent import ItineraryPlanner
@app.post("/v1/tasks/send")
async def send_task(request: TaskRequest):
"""Handle A2A tasks/send requests."""
global planner
if not planner:
raise HTTPException(status_code=503, detail="Planner not initialized")
try:
task_id = request.taskId
# Extract the message from the user
user_message = None
for part in request.message.get("parts", []):
if "text" in part:
user_message = part["text"]
break
if not user_message:
raise HTTPException(status_code=400, detail="No text message found in request")
# Generate an itinerary based on the query
itinerary = await planner.create_itinerary(user_message)
# Create the A2A response
response = {
"task": {
"taskId": task_id,
"state": "completed",
"messages": [
{
"role": "user",
"parts": [{"text": user_message}]
},
{
"role": "agent",
"parts": [{"text": itinerary}]
}
],
"artifacts": []
}
}
return response
6.5 Streamlit_ui — Streamlit으로 구축된 사용자 인터페이스로, 여행 계획을 위한 양식을 제공하고 결과를 사용자 친화적인 형식으로 표시합니다.
...
...
# API endpoint
API_URL = "http://localhost:8005/v1/tasks/send"
def generate_itinerary(query: str):
"""Send a query to the itinerary planner API."""
try:
task_id = "task-" + datetime.now().strftime("%Y%m%d%H%M%S")
payload = {
"taskId": task_id,
"message": {
"role": "user",
"parts": [
{
"text": query
}
]
}
}
# Log the user query and the request to the event log
log_user_query(query)
log_itinerary_request(payload)
response = requests.post(
API_URL,
json=payload,
headers={"Content-Type": "application/json"}
)
response.raise_for_status()
result = response.json()
# Extract the agent's response message
agent_message = None
for message in result.get("task", {}).get("messages", []):
if message.get("role") == "agent":
for part in message.get("parts", []):
if "text" in part:
agent_message = part["text"]
break
if agent_message:
break
..
..
...
7단계: 최종 데모 실행
각 터미널에서 아래와 같이 에이전트 서버들을 시작합니다.
# 항공편 검색 에이전트 시작 (포트 8000)
python -m flight_search_app.main
# 호텔 검색 에이전트 시작 (포트 8003)
python -m hotel_search_app.langchain_server
# 여정 플래너 호스트 에이전트 시작 (포트 8005)
python -m itinerary_planner.itinerary_server
# 프론트엔드 UI 시작 (포트 8501)
streamlit run itinerary_planner/streamlit_ui.py
Flight Search Logs with Task ID initiated from Host Agents
Hotel Search Logs with Tasks initiated from Host Agents
Itinerary Planner — Host agent with all Request /Response
Agent Event Logs
이 데모는 구글 A2A 프로토콜의 핵심 원칙을 구현하여 에이전트가 구조화되고 상호 운용 가능한 방식으로 통신할 수 있도록 합니다. 위 데모에서 완전히 구현된 구성 요소는 다음과 같습니다.
- 에이전트 카드(Agent Cards): 모든 에이전트는 검색을 위해 .well-known/agent.json 파일을 노출합니다.
- A2A 서버(A2A Servers): 각 에이전트(flight_search_app, hotel_search_app, itinerary_planner)는 A2A 서버로 실행됩니다.
- A2A 클라이언트(A2A Clients): itinerary_planner는 항공편 및 호텔 에이전트를 위한 전용 A2A 클라이언트를 포함합니다.
- 작업 관리(Task Management): 각 요청/응답은 '제출됨(submitted)', '작업 중(working)', '완료됨(completed)'과 같은 상태를 가진 A2A 작업으로 모델링됩니다.
- 메시지 구조(Message Structure): 역할(사용자/에이전트)과 파트(주로 TextPart)를 포함하는 표준 JSON-RPC 형식을 사용합니다.
아래 구성 요소는 데모에서 구현되지 않았지만, 엔터프라이즈급 에이전트를 위해 확장될 수 있습니다.
- 스트리밍(Streaming, SSE): A2A는 장기 실행 작업을 위해 서버 전송 이벤트(SSE)를 지원하지만, 저희 데모는 3-5초 미만이 소요되는 간단한 요청/응답 방식을 사용합니다.
- 푸시 알림(Push Notifications): 웹훅(Web-hook) 업데이트 메커니즘은 아직 사용되지 않았습니다.
- 복잡한 파트(Complex Parts): 텍스트 파트(TextPart)만 사용되었습니다. 더 풍부한 페이로드를 위해 데이터 파트(DataPart), 파일 파트(FilePart) 등의 지원을 추가할 수 있습니다.
- 고급 검색(Advanced Discovery): 기본적인 .well-known/agent.json은 구현되었지만, 고급 인증, JWKS 또는 권한 부여 흐름은 없습니다.
결론
이번 글에서는 여행 계획 시나리오에서 재사용 가능한 A2A 구성 요소, ADK, LangChain 및 MCP를 사용하여 완전한 기능을 갖춘 멀티 에이전트 시스템을 구축하는 방법을 살펴보았습니다. 이러한 오픈소스 도구와 프레임워크를 결합함으로써 우리 에이전트는 다음을 수행할 수 있었습니다.
- A2A를 사용하여 동적으로 서로를 발견하고 호출합니다.
- MCP를 통해 모델 친화적인 방식으로 외부 API에 연결합니다.
- ADK 및 LangChain과 같은 최신 프레임워크를 활용합니다.
- 명확한 작업 수명 주기와 구조화된 결과를 통해 비동기적으로 통신합니다.
동일한 원칙은 소매, 고객 서비스 자동화, 운영 워크플로우 및 AI 지원 엔터프라이즈 툴링과 같은 더 많은 도메인으로 확장될 수 있습니다.
728x90
'Cook AI' 카테고리의 다른 글
| Google ADK를 MCP 클라이언트로 활용한 AI 에이전트 구축 심층 분석 (2) | 2025.05.22 |
|---|---|
| Model Context Protocol(MCP) 과 구글 Gemini 2.5 Pro — 심층 분석 (2) | 2025.05.21 |
| Ollama와 Gemma3를 활용한 함수 호출 (8) | 2025.05.20 |
| AI 개발자라면 놓칠 수 없는 기회! AI 스타터팩 -무료 및 할인 (2) | 2025.05.19 |
| Supabase API Key 생성하기 (2) | 2025.05.16 |