[MCP&A2A] 03. A2A 프로토콜 이해

Agent-to-Agent Protocol (A2A) 개요

A2A는 다양한 프레임워크와 벤더가 구축한 AI 에이전트들이 서로 통신하고 협업할 수 있도록 하는 오픈 프로토콜입니다. 2025년 4월 Google이 발표했으며, 2025년 6월 Linux Foundation에 기부되었습니다.

A2A의 핵심 가치 제안

문제:

각 에이전트가 독립적으로 작동
→ 협업 불가능
→ 복잡한 태스크 처리 어려움
→ 수동 오케스트레이션 필요

해결:

에이전트 A ← A2A 프로토콜 → 에이전트 B
→ 표준화된 통신
→ 자동 협업
→ 멀티스텝 워크플로우

주요 특징

1. 에이전트 디스커버리: Agent Card로 기능 광고
2. 비동기 실행: 장기 실행 태스크 지원
3. 실시간 스트리밍: SSE로 진행 상황 업데이트
4. 상태 관리: 태스크 생명주기 추적
5. 프레임워크 중립: LangGraph, CrewAI, Semantic Kernel 등 모두 지원

A2A 아키텍처

기본 구조

┌─────────────────────────────────┐
│    클라이언트 에이전트          │
│  (태스크 요청자)                │
└────────────┬────────────────────┘
             │ A2A 프로토콜
             │ (HTTP/gRPC + SSE)
             ▼
┌─────────────────────────────────┐
│    서버 에이전트                │
│  (태스크 실행자)                │
│  ┌───────────────────────────┐  │
│  │ Agent Card                │  │
│  │ • 이름, 설명              │  │
│  │ • 스킬 목록               │  │
│  │ • 입출력 모드             │  │
│  └───────────────────────────┘  │
│  ┌───────────────────────────┐  │
│  │ 태스크 실행 엔진          │  │
│  │ • 태스크 생성             │  │
│  │ • 실행 관리               │  │
│  │ • 결과 반환               │  │
│  └───────────────────────────┘  │
└─────────────────────────────────┘

핵심 개념

1. Agent Card

에이전트의 기능을 설명하는 JSON 문서:

{
  "name": "Research Assistant",
  "description": "멀티소스 리서치 및 보고서 생성 전문 에이전트",
  "url": "https://api.example.com/agents/research",
  "version": "1.0.0",
  "skills": [
    {
      "name": "web_research",
      "description": "웹에서 정보 검색 및 분석",
      "inputModes": ["text"],
      "outputModes": ["text", "data"]
    },
    {
      "name": "report_generation",
      "description": "구조화된 보고서 생성",
      "inputModes": ["text", "data"],
      "outputModes": ["text", "file"]
    }
  ],
  "authentication": {
    "type": "bearer",
    "description": "JWT 토큰 필요"
  }
}

Agent Card 조회:

GET /.well-known/agent-card.json?assistant_id=research-assistant
Host: api.example.com

2. 태스크 (Task)

A2A의 작업 단위:

{
  "taskId": "task_abc123",
  "agentId": "research-assistant",
  "skill": "web_research",
  "input": {
    "query": "AI 에이전트 프로토콜 비교 분석",
    "depth": "comprehensive"
  },
  "status": "running",
  "createdAt": "2024-12-13T10:00:00Z",
  "updatedAt": "2024-12-13T10:05:30Z"
}

태스크 생명주기:

pending → running → completed
                  ↘ failed
                  ↘ cancelled

3. 메시지 (Message)

에이전트 간 통신 단위:

{
  "role": "user",  // 또는 "agent"
  "parts": [
    {
      "kind": "text",
      "text": "MCP와 A2A의 차이점을 설명해주세요"
    }
  ],
  "metadata": {
    "timestamp": "2024-12-13T10:00:00Z"
  }
}

Part 타입:

• text: 텍스트 콘텐츠
• data: JSON 구조화 데이터
• file: 파일 참조
• image: 이미지 데이터

4. Artifact

태스크 실행 결과물:

{
  "artifactId": "artifact_xyz789",
  "name": "연구 보고서",
  "parts": [
    {
      "kind": "text",
      "text": "# MCP vs A2A 비교 분석\n\n..."
    },
    {
      "kind": "file",
      "url": "https://storage.example.com/reports/mcp-a2a.pdf",
      "mimeType": "application/pdf"
    }
  ],
  "createdAt": "2024-12-13T10:10:00Z"
}

A2A 프로토콜 메서드

1. 태스크 생성 (tasks/create)

POST /a2a/{agentId}
Content-Type: application/json

{
  "jsonrpc": "2.0",
  "id": "req_1",
  "method": "tasks/create",
  "params": {
    "skill": "web_research",
    "input": {
      "query": "AI 에이전트 협업 사례"
    }
  }
}

응답:

{
  "jsonrpc": "2.0",
  "id": "req_1",
  "result": {
    "taskId": "task_abc123",
    "status": "pending",
    "estimatedDuration": 300
  }
}

2. 태스크 조회 (tasks/get)

POST /a2a/{agentId}
Content-Type: application/json

{
  "jsonrpc": "2.0",
  "id": "req_2",
  "method": "tasks/get",
  "params": {
    "taskId": "task_abc123"
  }
}

응답:

{
  "jsonrpc": "2.0",
  "id": "req_2",
  "result": {
    "taskId": "task_abc123",
    "status": "completed",
    "artifacts": [
      {
        "artifactId": "artifact_xyz789",
        "name": "검색 결과",
        "parts": [...]
      }
    ]
  }
}

3. 메시지 전송 (message/send)

POST /a2a/{agentId}
Content-Type: application/json

{
  "jsonrpc": "2.0",
  "id": "req_3",
  "method": "message/send",
  "params": {
    "message": {
      "role": "user",
      "parts": [
        {
          "kind": "text",
          "text": "추가 분석 필요: 성능 비교"
        }
      ]
    },
    "messageId": "msg_123",
    "thread": {
      "threadId": "thread_abc"
    }
  }
}

4. 스트림 메시지 (message/stream)

Server-Sent Events로 실시간 응답:

POST /a2a/{agentId}
Accept: text/event-stream
Content-Type: application/json

{
  "jsonrpc": "2.0",
  "id": "req_4",
  "method": "message/stream",
  "params": {
    "message": {
      "role": "user",
      "parts": [{"kind": "text", "text": "보고서 생성"}]
    }
  }
}

SSE 응답:

event: message
data: {"role":"agent","parts":[{"kind":"text","text":"분석 시작..."}]}

event: message
data: {"role":"agent","parts":[{"kind":"text","text":"데이터 수집 중..."}]}

event: artifact
data: {"artifactId":"artifact_1","name":"중간 결과",...}

event: done
data: {"status":"completed"}

Go 구현 예제

A2A 서버 구조

// a2a-server/internal/handlers/tasks.go
package handlers

import (
    "context"
    "encoding/json"
    "net/http"
    "time"
    
    "github.com/google/uuid"
)

type TaskHandler struct {
    taskStore TaskStore
    executor  TaskExecutor
}

// 태스크 생성
func (h *TaskHandler) CreateTask(w http.ResponseWriter, r *http.Request) {
    var req struct {
        JSONRPC string `json:"jsonrpc"`
        ID      string `json:"id"`
        Method  string `json:"method"`
        Params  struct {
            Skill string                 `json:"skill"`
            Input map[string]interface{} `json:"input"`
        } `json:"params"`
    }
    
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        respondError(w, -32700, "Parse error")
        return
    }
    
    // 태스크 생성
    task := &Task{
        ID:        uuid.New().String(),
        AgentID:   chi.URLParam(r, "agentId"),
        Skill:     req.Params.Skill,
        Input:     req.Params.Input,
        Status:    StatusPending,
        CreatedAt: time.Now(),
    }
    
    // 저장
    if err := h.taskStore.Save(r.Context(), task); err != nil {
        respondError(w, -32603, "Failed to create task")
        return
    }
    
    // 비동기 실행
    go h.executor.Execute(context.Background(), task)
    
    // 응답
    respondSuccess(w, req.ID, map[string]interface{}{
        "taskId": task.ID,
        "status": task.Status,
    })
}

// 태스크 조회
func (h *TaskHandler) GetTask(w http.ResponseWriter, r *http.Request) {
    var req struct {
        JSONRPC string `json:"jsonrpc"`
        ID      string `json:"id"`
        Params  struct {
            TaskID string `json:"taskId"`
        } `json:"params"`
    }
    
    json.NewDecoder(r.Body).Decode(&req)
    
    task, err := h.taskStore.Get(r.Context(), req.Params.TaskID)
    if err != nil {
        respondError(w, -32004, "Task not found")
        return
    }
    
    respondSuccess(w, req.ID, task)
}

SSE 스트리밍

// a2a-server/internal/handlers/events.go
func (h *TaskHandler) StreamEvents(w http.ResponseWriter, r *http.Request) {
    taskID := chi.URLParam(r, "taskId")
    
    // SSE 헤더 설정
    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")
    
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "Streaming not supported", 500)
        return
    }
    
    // 이벤트 채널 구독
    eventChan := h.eventBus.Subscribe(taskID)
    defer h.eventBus.Unsubscribe(taskID, eventChan)
    
    // 이벤트 스트리밍
    for {
        select {
        case event := <-eventChan:
            data, _ := json.Marshal(event)
            fmt.Fprintf(w, "event: %s\n", event.Type)
            fmt.Fprintf(w, "data: %s\n\n", data)
            flusher.Flush()
            
            if event.Type == "done" || event.Type == "error" {
                return
            }
            
        case <-r.Context().Done():
            return
        }
    }
}

Python 클라이언트 예제

# orchestration/clients/a2a_client.py
import requests
import json
from typing import Iterator, Dict, Any

class A2AClient:
    def __init__(self, base_url: str, agent_id: str):
        self.base_url = base_url
        self.agent_id = agent_id
    
    def create_task(self, skill: str, input_data: Dict[str, Any]) -> str:
        """태스크 생성"""
        response = requests.post(
            f"{self.base_url}/a2a/{self.agent_id}",
            json={
                "jsonrpc": "2.0",
                "id": "1",
                "method": "tasks/create",
                "params": {
                    "skill": skill,
                    "input": input_data
                }
            }
        )
        
        result = response.json()["result"]
        return result["taskId"]
    
    def get_task(self, task_id: str) -> Dict[str, Any]:
        """태스크 조회"""
        response = requests.post(
            f"{self.base_url}/a2a/{self.agent_id}",
            json={
                "jsonrpc": "2.0",
                "id": "2",
                "method": "tasks/get",
                "params": {"taskId": task_id}
            }
        )
        
        return response.json()["result"]
    
    def stream_events(self, task_id: str) -> Iterator[Dict[str, Any]]:
        """SSE 이벤트 스트리밍"""
        url = f"{self.base_url}/a2a/{self.agent_id}/tasks/{task_id}/events"
        
        with requests.get(url, stream=True) as response:
            for line in response.iter_lines():
                if line.startswith(b'data:'):
                    data = json.loads(line[5:])
                    yield data

MCP와 A2A의 통합

에이전트가 MCP 도구 사용

# orchestration/workflows/research_workflow.py
from langgraph.graph import StateGraph
from typing import TypedDict

class ResearchState(TypedDict):
    query: str
    documents: list
    report: str

class ResearchAgent:
    def __init__(self, mcp_url: str):
        self.mcp_client = MCPClient(mcp_url)
        self.workflow = self.build_workflow()
    
    def build_workflow(self):
        workflow = StateGraph(ResearchState)
        
        workflow.add_node("search", self.search_docs)
        workflow.add_node("analyze", self.analyze_docs)
        workflow.add_node("generate", self.generate_report)
        
        workflow.add_edge("search", "analyze")
        workflow.add_edge("analyze", "generate")
        
        return workflow.compile()
    
    def search_docs(self, state: ResearchState) -> ResearchState:
        # MCP 도구 사용
        results = self.mcp_client.hybrid_search(
            query=state["query"],
            limit=20
        )
        
        state["documents"] = results
        emit_progress("search_complete", f"Found {len(results)} documents")
        
        return state
    
    def analyze_docs(self, state: ResearchState) -> ResearchState:
        # 문서 분석
        # ...
        emit_progress("analysis_complete", "Analysis done")
        return state
    
    def generate_report(self, state: ResearchState) -> ResearchState:
        # 보고서 생성
        # ...
        emit_progress("report_complete", "Report generated")
        return state

핵심 요약

A2A의 강점

✅ 에이전트 협업: 표준화된 통신 프로토콜 ✅ 비동기 실행: 장기 실행 태스크 지원 ✅ 실시간 피드백: SSE 스트리밍 ✅ 상태 관리: 태스크 생명주기 추적 ✅ 프레임워크 중립: 모든 에이전트 프레임워크 지원

MCP와의 차이

측면	MCP	A2A
용도	AI ↔ 도구	AI ↔ AI
실행	동기식	비동기식
상태	무상태	상태 유지
시나리오	빠른 데이터 접근	복잡한 워크플로우

다음 장: MCP와 A2A 상세 비교 및 통합 전략

---

참고 자료:

• A2A 공식 사이트: https://a2a-protocol.org/
• GitHub: https://github.com/a2aproject/A2A
• Google 발표: https://developers.googleblog.com/en/a2a-a-new-era-of-agent-interoperability/

작성일: 2024년 12월 13일