3장: 에이전트 간 통신 프로토콜 — A2A, MCP, 커스텀 패턴

{
  "name": "data-analyst-agent",
  "description": "데이터 분석과 시각화를 수행하는 에이전트",
  "url": "https://agents.example.com/data-analyst",
  "version": "1.0.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },
  "skills": [
    {
      "id": "analyze-csv",
      "name": "CSV 데이터 분석",
      "description": "CSV 파일을 읽고 통계 분석을 수행합니다",
      "tags": ["data", "analytics", "csv"]
    },
    {
      "id": "create-chart",
      "name": "차트 생성",
      "description": "분석 결과를 다양한 차트로 시각화합니다",
      "tags": ["visualization", "chart"]
    }
  ],
  "authentication": {
    "schemes": ["bearer"]
  }
}

import httpx
from dataclasses import dataclass
 
@dataclass
class A2ATask:
    id: str
    status: str
    artifacts: list[dict]
 
class A2AClient:
    def __init__(self, agent_url: str, auth_token: str):
        self.agent_url = agent_url
        self.client = httpx.AsyncClient(
            headers={"Authorization": f"Bearer {auth_token}"}
        )
 
    async def discover(self) -> dict:
        """Agent Card를 조회하여 에이전트 능력을 확인"""
        response = await self.client.get(
            f"{self.agent_url}/.well-known/agent.json"
        )
        return response.json()
 
    async def send_task(self, message: str) -> A2ATask:
        """작업을 전송하고 결과를 대기"""
        payload = {
            "jsonrpc": "2.0",
            "method": "tasks/send",
            "params": {
                "id": generate_task_id(),
                "message": {
                    "role": "user",
                    "parts": [{"type": "text", "text": message}]
                }
            }
        }
        response = await self.client.post(
            f"{self.agent_url}",
            json=payload
        )
        result = response.json()["result"]
        return A2ATask(
            id=result["id"],
            status=result["status"]["state"],
            artifacts=result.get("artifacts", [])
        )
 
    async def send_task_streaming(self, message: str):
        """SSE를 통해 실시간 결과 수신"""
        payload = {
            "jsonrpc": "2.0",
            "method": "tasks/sendSubscribe",
            "params": {
                "id": generate_task_id(),
                "message": {
                    "role": "user",
                    "parts": [{"type": "text", "text": message}]
                }
            }
        }
        async with self.client.stream(
            "POST", self.agent_url, json=payload
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data:"):
                    yield json.loads(line[5:])

from a2a.server import A2AServer, TaskHandler
from a2a.types import (
    AgentCard, Skill, TaskState,
    Message, TextPart, Artifact
)
 
class DataAnalystHandler(TaskHandler):
    async def handle_task(self, task_id: str, message: Message):
        # 상태를 working으로 업데이트
        await self.update_status(task_id, TaskState.WORKING)
 
        # 메시지에서 분석 요청 추출
        user_text = message.parts[0].text
 
        # 분석 수행 (내부 로직)
        result = await self.analyze_data(user_text)
 
        # Artifact로 결과 반환
        await self.add_artifact(
            task_id,
            Artifact(
                name="analysis-result",
                parts=[TextPart(text=result)]
            )
        )
 
        # 상태를 completed로 업데이트
        await self.update_status(task_id, TaskState.COMPLETED)
 
# 서버 설정
agent_card = AgentCard(
    name="data-analyst",
    description="데이터 분석 전문 에이전트",
    skills=[
        Skill(
            id="analyze",
            name="데이터 분석",
            description="정형 데이터를 분석합니다"
        )
    ]
)
 
server = A2AServer(
    agent_card=agent_card,
    handler=DataAnalystHandler()
)

# MCP 서버: 데이터베이스 도구 제공
from mcp.server import Server
from mcp.types import Tool, TextContent
 
server = Server("database-tools")
 
@server.tool()
async def query_users(department: str) -> list[dict]:
    """특정 부서의 사용자 목록을 조회합니다"""
    result = await db.execute(
        "SELECT * FROM users WHERE department = $1",
        department
    )
    return result.rows

# A2A: 분석 에이전트가 시각화 에이전트에게 작업 위임
async def orchestrate_report(query: str):
    # 1. MCP로 데이터 수집 (도구 접근)
    raw_data = await mcp_client.call_tool(
        "query_sales", {"period": "Q1-2026"}
    )
 
    # 2. 자체적으로 분석 수행
    analysis = await self.analyze(raw_data)
 
    # 3. A2A로 시각화 에이전트에게 위임 (에이전트 협업)
    viz_client = A2AClient("https://viz-agent.example.com")
    chart_task = await viz_client.send_task(
        f"다음 데이터를 바 차트로 시각화해주세요: {analysis}"
    )
 
    return {"analysis": analysis, "chart": chart_task.artifacts}

from pydantic import BaseModel
from enum import Enum
from typing import Any
 
class MessageType(str, Enum):
    TASK_REQUEST = "task_request"
    TASK_RESULT = "task_result"
    STATUS_UPDATE = "status_update"
    ERROR = "error"
    FEEDBACK = "feedback"
 
class AgentMessage(BaseModel):
    type: MessageType
    sender: str
    recipient: str
    correlation_id: str  # 작업 추적용
    payload: dict[str, Any]
    metadata: dict[str, Any] = {}
 
    class Config:
        json_schema_extra = {
            "example": {
                "type": "task_request",
                "sender": "supervisor",
                "recipient": "researcher",
                "correlation_id": "task-001",
                "payload": {
                    "instruction": "AI 에이전트 최신 트렌드를 조사하세요",
                    "constraints": {
                        "max_sources": 10,
                        "recency": "6months"
                    }
                },
                "metadata": {
                    "priority": "high",
                    "deadline": "2026-04-09T18:00:00Z"
                }
            }
        }

import asyncio
from collections import defaultdict
 
class AgentMessageBroker:
    """경량 인메모리 메시지 브로커"""
 
    def __init__(self):
        self._queues: dict[str, asyncio.Queue] = defaultdict(asyncio.Queue)
        self._subscribers: dict[str, list[callable]] = defaultdict(list)
 
    async def publish(self, topic: str, message: AgentMessage):
        """토픽에 메시지 발행"""
        await self._queues[topic].put(message)
        for callback in self._subscribers[topic]:
            asyncio.create_task(callback(message))
 
    async def subscribe(self, topic: str, callback: callable):
        """토픽 구독"""
        self._subscribers[topic].append(callback)
 
    async def consume(self, topic: str) -> AgentMessage:
        """토픽에서 메시지 소비 (블로킹)"""
        return await self._queues[topic].get()
 
# 사용 예시
broker = AgentMessageBroker()
 
# 감독자가 작업 요청 발행
await broker.publish("research-tasks", AgentMessage(
    type=MessageType.TASK_REQUEST,
    sender="supervisor",
    recipient="researcher",
    correlation_id="report-2026-q1",
    payload={"topic": "AI agent trends Q1 2026"}
))
 
# 리서치 에이전트가 결과 발행
await broker.publish("research-results", AgentMessage(
    type=MessageType.TASK_RESULT,
    sender="researcher",
    recipient="supervisor",
    correlation_id="report-2026-q1",
    payload={"findings": [...], "sources": [...]}
))

@dataclass
class TaskContext:
    """에이전트 간 전달되는 작업 컨텍스트"""
    task_id: str
    original_request: str
    completed_steps: list[dict]
    current_step: str
    accumulated_results: dict
    constraints: dict
 
    def summarize_for_next_agent(self, max_tokens: int = 2000) -> str:
        """다음 에이전트에게 전달할 요약 생성"""
        summary_parts = [
            f"원래 요청: {self.original_request}",
            f"현재 단계: {self.current_step}",
            "이전 단계 결과:"
        ]
        for step in self.completed_steps:
            summary_parts.append(
                f"  - {step['name']}: {step['summary']}"
            )
        return "\n".join(summary_parts)[:max_tokens]