AI Sparkup

복잡한 AI 세상을 읽는 힘

A2A vs MCP: AI 에이전트 개발자를 위한 완벽 가이드

AI 에이전트가 단순한 챗봇을 넘어 복잡한 업무를 자동화하는 시대가 도래했습니다. 하지만 지금까지 대부분의 AI 에이전트들은 고립된 상태로 작동해왔습니다. 각자 다른 도구를 사용하고, 서로 소통할 수 있는 표준화된 방법이 없었기 때문입니다.

2025년, 이런 상황을 변화시킬 두 가지 혁신적인 프로토콜이 등장했습니다. Google의 A2A(Agent-to-Agent)와 Anthropic의 MCP(Model Context Protocol)입니다. 이 두 프로토콜은 AI 에이전트의 협업과 확장성을 획기적으로 향상시킬 수 있는 열쇠를 제공합니다.

이 가이드에서는 두 프로토콜의 핵심 개념부터 실제 구현 방법까지, 개발자가 알아야 할 모든 것을 다룹니다.

AI Agent Collaboration

A2A와 MCP 핵심 개념 비교

A2A(Agent-to-Agent) 프로토콜이란?

A2A는 서로 다른 AI 에이전트들이 직접 소통하고 협업할 수 있게 해주는 Google의 오픈 프로토콜입니다. 마치 팀 프로젝트에서 각 전문가가 자신의 역할을 맡아 협력하는 것처럼, AI 에이전트들도 각자의 전문 분야에서 작업을 분담하고 결과를 공유할 수 있게 됩니다.

핵심 특징:

  • HTTP/JSON 기반의 간단한 통신 방식
  • 에이전트 발견(Agent Discovery) 기능
  • 비동기 작업 처리 지원
  • 보안과 인증 기능 내장

MCP(Model Context Protocol)란?

MCP는 AI 에이전트가 외부 도구와 데이터 소스에 접근할 수 있게 해주는 Anthropic의 표준 프로토콜입니다. AI용 ‘USB-C 포트’라고 비유할 수 있으며, 한 번 구현하면 다양한 AI 모델에서 동일한 도구를 사용할 수 있게 합니다.

핵심 특징:

  • 통합된 도구 인터페이스 제공
  • 실시간 데이터 접근 가능
  • JSON-RPC 기반 통신
  • 도구별 권한 관리

주요 차이점 비교

구분A2AMCP
목적에이전트 간 협업에이전트-도구 연결
통신 대상AI 에이전트 ↔ AI 에이전트AI 에이전트 ↔ 외부 도구/데이터
개발사GoogleAnthropic
통신 방식HTTP/JSONJSON-RPC
주요 기능작업 위임, 상태 공유도구 등록, 데이터 접근
적용 사례멀티 에이전트 워크플로우단일 에이전트 도구 확장

언제 무엇을 사용해야 할까?

A2A를 선택해야 하는 경우:

  • 여러 전문 에이전트가 협력해야 하는 복잡한 워크플로우
  • 각 에이전트가 서로 다른 클라우드나 서비스에서 실행되는 경우
  • 에이전트 간 작업 분담과 결과 통합이 필요한 경우

MCP를 선택해야 하는 경우:

  • 단일 에이전트가 다양한 외부 도구에 접근해야 하는 경우
  • 실시간 데이터 연동이 중요한 애플리케이션
  • 기존 시스템과의 통합이 우선인 경우

A2A 프로토콜 실무 가이드

Agent Communication Network

A2A 작동 원리

A2A는 다음과 같은 구성 요소로 작동합니다:

  1. Agent Card: 에이전트의 능력과 정보를 담은 프로필
  2. Task Lifecycle: 작업의 생성부터 완료까지의 과정 관리
  3. Message Structure: 에이전트 간 주고받는 메시지 형식
  4. Artifacts: 작업 결과물을 구조화된 형태로 전달

Agent Card 작성하기

Agent Card는 .well-known/agent.json 경로에 위치하며, 다음과 같은 형식을 따릅니다:

{
  "name": "TravelPlannerAgent",
  "description": "여행 계획을 수립하는 전문 에이전트",
  "url": "https://api.example.com/travel-agent",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "defaultInputModes": ["text"],
  "defaultOutputModes": ["text", "json"],
  "skills": [
    {
      "id": "flight_search",
      "name": "항공편 검색",
      "description": "최적의 항공편을 찾아 추천합니다"
    },
    {
      "id": "hotel_booking",
      "name": "호텔 예약",
      "description": "예산과 선호도에 맞는 숙소를 찾습니다"
    }
  ]
}

실제 구현 예시

Python을 사용한 기본적인 A2A 클라이언트 구현:

import asyncio
import aiohttp
import json

class A2AClient:
    def __init__(self, agent_url):
        self.agent_url = agent_url
        self.session = None
    
    async def discover_agent(self, agent_url):
        """에이전트 정보 조회"""
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{agent_url}/.well-known/agent.json") as response:
                return await response.json()
    
    async def send_task(self, task_data):
        """작업 요청 전송"""
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.agent_url}/tasks/send",
                json=task_data,
                headers={"Content-Type": "application/json"}
            ) as response:
                return await response.json()

# 사용 예시
async def main():
    client = A2AClient("https://api.example.com/travel-agent")
    
    # 에이전트 정보 조회
    agent_info = await client.discover_agent("https://api.example.com/travel-agent")
    print("에이전트 정보:", agent_info)
    
    # 작업 요청
    task = {
        "task_id": "travel_001",
        "request": "서울에서 도쿄로 가는 3박 4일 여행 계획을 세워주세요",
        "parameters": {
            "budget": 1000000,
            "travel_date": "2025-07-01"
        }
    }
    
    result = await client.send_task(task)
    print("작업 결과:", result)

if __name__ == "__main__":
    asyncio.run(main())

MCP 프로토콜 실무 가이드

MCP 아키텍처 이해하기

MCP는 3가지 주요 구성 요소로 이루어져 있습니다:

  1. Host: LLM 애플리케이션 (Claude Desktop, Cursor 등)
  2. Client: Host 내부의 MCP 연결 관리자
  3. Server: 도구와 리소스를 제공하는 외부 서비스

도구 서버 설정하기

MCP 서버를 설정하는 방법을 GitHub 연동을 예로 살펴보겠습니다:

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "your_token_here"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/allowed/files"]
    }
  }
}

Python으로 커스텀 MCP 서버 만들기

from mcp.server import NotificationOptions, Server
import mcp.types as types

# MCP 서버 생성
server = Server("example-server")

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """사용 가능한 도구 목록 반환"""
    return [
        types.Tool(
            name="calculate",
            description="수학 계산을 수행합니다",
            inputSchema={
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "계산할 수학 표현식"
                    }
                },
                "required": ["expression"]
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    """도구 호출 처리"""
    if name == "calculate":
        try:
            expression = arguments["expression"]
            result = eval(expression)  # 실제 환경에서는 보안상 안전한 계산기 사용
            return [types.TextContent(type="text", text=f"결과: {result}")]
        except Exception as e:
            return [types.TextContent(type="text", text=f"오류: {e}")]
    else:
        raise ValueError(f"알 수 없는 도구: {name}")

async def main():
    # 서버 실행
    async with server.run_stdio() as (read_stream, write_stream):
        await server.run(
            read_stream, write_stream, 
            notification_options=NotificationOptions()
        )

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

A2A + MCP 결합 활용법

Integrated AI System

상호 보완하는 방식

A2A와 MCP를 함께 사용하면 다음과 같은 시너지 효과를 얻을 수 있습니다:

  • 확장성: MCP로 각 에이전트를 도구로 확장하고, A2A로 에이전트 간 협업 구현
  • 전문화: 각 에이전트가 MCP를 통해 특정 도메인 도구에 접근하고, A2A로 전문성 공유
  • 실시간 적응: MCP로 최신 데이터 접근, A2A로 상황 변화에 따른 작업 재분배

여행 계획 시스템 구현 예시

다음은 A2A와 MCP를 결합한 여행 계획 시스템의 구현 예시입니다:

# 메인 코디네이터 에이전트
class TravelCoordinator:
    def __init__(self):
        self.agents = {
            'flight': 'https://api.example.com/flight-agent',
            'hotel': 'https://api.example.com/hotel-agent',  
            'activity': 'https://api.example.com/activity-agent'
        }
        
    async def plan_trip(self, request):
        # 1. 각 전문 에이전트에게 작업 위임 (A2A)
        tasks = []
        
        # 항공편 검색 에이전트
        flight_task = {
            "agent": self.agents['flight'],
            "request": f"Find flights from {request['origin']} to {request['destination']}",
            "budget": request['budget'] * 0.4  # 예산의 40%
        }
        
        # 호텔 검색 에이전트  
        hotel_task = {
            "agent": self.agents['hotel'],
            "request": f"Find hotels in {request['destination']}",
            "budget": request['budget'] * 0.4  # 예산의 40%
        }
        
        # 액티비티 추천 에이전트
        activity_task = {
            "agent": self.agents['activity'],
            "request": f"Recommend activities in {request['destination']}",
            "budget": request['budget'] * 0.2  # 예산의 20%
        }
        
        # 병렬로 작업 실행
        results = await asyncio.gather(
            self.send_to_agent(flight_task),
            self.send_to_agent(hotel_task), 
            self.send_to_agent(activity_task)
        )
        
        # 결과 통합
        return self.integrate_results(results)

# 항공편 전문 에이전트 (MCP 활용)
class FlightAgent:
    def __init__(self):
        # MCP를 통해 항공편 검색 API 연결
        self.mcp_tools = [
            "flight_search_api",
            "price_comparison_tool",
            "schedule_optimizer"
        ]
    
    async def search_flights(self, origin, destination, date, budget):
        # MCP 도구를 사용하여 항공편 검색
        search_results = await self.call_mcp_tool(
            "flight_search_api",
            {
                "origin": origin,
                "destination": destination, 
                "date": date,
                "max_price": budget
            }
        )
        
        # 가격 비교 도구로 최적 옵션 선택
        best_options = await self.call_mcp_tool(
            "price_comparison_tool",
            {"flights": search_results}
        )
        
        return best_options

브라우저 자동화 시스템 예시

Google ADK를 활용한 실제 구현:

from google.adk.agents.llm_agent import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters

async def create_browser_automation_system():
    # Puppeteer MCP 서버 연결
    tools, exit_stack = await MCPToolset.from_server(
        connection_params=StdioServerParameters(
            command='npx',
            args=["-y", "@modelcontextprotocol/server-puppeteer"]
        )
    )
    
    # 브라우저 자동화 에이전트 생성
    browser_agent = LlmAgent(
        model='gemini-2.0-flash',
        name='browser_automation_agent',
        instruction=(
            "웹 자동화 작업을 수행하는 전문 에이전트입니다. "
            "브라우저 도구를 사용하여 페이지 탐색, 클릭, 데이터 추출을 수행합니다."
        ),
        tools=tools
    )
    
    return browser_agent, exit_stack

# 사용 예시
async def automate_news_extraction():
    agent, exit_stack = await create_browser_automation_system()
    
    query = (
        "Google 뉴스 사이트에 접속하여 "
        "상위 5개 헤드라인을 추출해주세요."
    )
    
    # A2A를 통해 다른 에이전트와 협업 가능
    result = await agent.process_request(query)
    
    await exit_stack.aclose()
    return result

도입 시 고려사항

보안과 인증 관리

A2A 보안 고려사항:

  • OAuth 2.0/2.1 기반 인증 구현
  • HTTPS 통신 필수
  • 에이전트 간 신뢰 관계 설정
  • 작업 권한 세분화

MCP 보안 고려사항:

  • 도구별 접근 권한 제한
  • 민감한 데이터 암호화
  • 샌드박스 환경에서 도구 실행
  • API 키 안전한 관리

성능 최적화 팁

  1. 연결 풀링: HTTP 연결을 재사용하여 오버헤드 감소
  2. 비동기 처리: 여러 에이전트/도구와 동시 통신
  3. 캐싱 전략: 자주 사용되는 데이터나 결과 캐시
  4. 타임아웃 설정: 응답 없는 에이전트/도구 처리
# 성능 최적화 예시
import asyncio
from aiohttp import ClientSession, TCPConnector

class OptimizedA2AClient:
    def __init__(self):
        self.connector = TCPConnector(
            limit=100,  # 최대 연결 수
            limit_per_host=10,  # 호스트당 최대 연결 수
            ttl_dns_cache=300,  # DNS 캐시 TTL
            use_dns_cache=True
        )
        self.session = ClientSession(
            connector=self.connector,
            timeout=aiohttp.ClientTimeout(total=30)  # 30초 타임아웃
        )
    
    async def batch_request(self, tasks):
        """여러 작업을 병렬로 처리"""
        semaphore = asyncio.Semaphore(5)  # 동시 실행 제한
        
        async def limited_request(task):
            async with semaphore:
                return await self.send_task(task)
        
        return await asyncio.gather(*[
            limited_request(task) for task in tasks
        ])

일반적인 실수와 해결책

실수 1: 에이전트 간 무한 루프

# 잘못된 예시
async def bad_delegation():
    if condition:
        return await delegate_to_agent_a()
    else:
        return await delegate_to_agent_b()  # Agent B가 다시 현재 에이전트 호출 시 무한 루프

# 올바른 예시  
async def proper_delegation():
    max_depth = 3
    if delegation_depth < max_depth:
        return await delegate_with_depth_tracking()
    else:
        return fallback_response()

실수 2: MCP 도구 오류 처리 부족

# 올바른 오류 처리
async def safe_mcp_call(tool_name, params):
    try:
        result = await call_mcp_tool(tool_name, params)
        return result
    except ConnectionError:
        return {"error": "도구 서버 연결 실패", "fallback": True}
    except TimeoutError:
        return {"error": "응답 시간 초과", "retry": True}
    except Exception as e:
        return {"error": f"예상치 못한 오류: {str(e)}"}

마무리

A2A와 MCP는 AI 에이전트 개발의 새로운 패러다임을 제시합니다. A2A는 에이전트 간 협업을, MCP는 에이전트의 능력 확장을 가능하게 하며, 두 프로토콜을 함께 사용할 때 그 진가를 발휘합니다.

Microsoft의 사티아 나델라 CEO가 말했듯이, “A2A와 MCP 같은 오픈 프로토콜은 에이전트 웹을 가능하게 하는 핵심”입니다. 고객들은 설계부터 상호 운용 가능한 에이전트 시스템을 구축할 수 있게 되었습니다.

이제 선택의 문제가 아닙니다. 두 프로토콜을 어떻게 조합하여 더 지능적이고 협력적인 AI 시스템을 만들 것인가가 핵심입니다. 이 가이드가 여러분의 AI 에이전트 개발 여정에 실질적인 도움이 되기를 바랍니다.


참고자료:

Comments