MCP와 A2A로 멀티에이전트 시스템 구축하기 — Model Context Protocol·Agent-to-Agent Protocol 실전 통합 가이드

에이전트 (MCP Client)
    │
    ├─ MCP 서버 A (파일 시스템)
    ├─ MCP 서버 B (데이터베이스)
    └─ MCP 서버 C (외부 API)

# pip install fastmcp httpx
from fastmcp import FastMCP
import httpx
 
mcp = FastMCP("weather-server")
 
@mcp.tool()
async def get_weather(city: str) -> dict:
    """도시명을 받아 현재 날씨 정보를 반환합니다."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.weather.example.com/current",
            params={"q": city, "units": "metric"}
        )
        data = response.json()
        return {
            "city": city,
            "temperature": data["main"]["temp"],
            "description": data["weather"][0]["description"]
        }
 
@mcp.tool()
async def get_forecast(city: str, days: int = 5) -> list[dict]:
    """도시의 날씨 예보를 반환합니다."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.weather.example.com/forecast",
            params={"q": city, "cnt": days, "units": "metric"}
        )
        return response.json()["list"]
 
if __name__ == "__main__":
    mcp.run()

// npm install @modelcontextprotocol/sdk
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
 
const transport = new StdioClientTransport({
  command: "python",
  args: ["weather_server.py"]
});
 
const client = new Client({ name: "weather-agent", version: "1.0.0" });
await client.connect(transport);
 
// 사용 가능한 도구 목록 동적 조회 — 하드코딩 불필요
const tools = await client.listTools();
console.log("사용 가능한 도구:", tools.tools.map(t => t.name));
 
// 도구 호출
const result = await client.callTool({
  name: "get_weather",
  arguments: { city: "Seoul" }
});
console.log("날씨 정보:", result.content);

{
  "name": "ResearchAgent",
  "version": "1.0.0",
  "description": "웹 검색과 사실 확인을 담당하는 에이전트",
  "url": "https://research-agent.example.com",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true
  },
  "skills": [
    {
      "id": "fact-check",
      "name": "Fact Check",
      "description": "주어진 정보의 사실 여부를 검증합니다",
      "inputModes": ["text"],
      "outputModes": ["text", "data"]
    },
    {
      "id": "web-search",
      "name": "Web Search",
      "description": "웹에서 관련 정보를 검색합니다",
      "inputModes": ["text"],
      "outputModes": ["text", "data"]
    }
  ],
  "authentication": {
    "schemes": ["bearer"]
  }
}

# pip install httpx
import httpx
import json
import uuid
from typing import AsyncIterator
 
async def delegate_task_to_agent(
    agent_url: str,
    task: str,
    context_id: str | None = None
) -> AsyncIterator[dict]:
    """A2A 프로토콜로 다른 에이전트에게 작업을 위임합니다."""
    # hash() 대신 uuid4()가 안전합니다 — hash()는 실행마다 값이 달라질 수 있습니다
    task_id = str(uuid.uuid4())
 
    payload = {
        "jsonrpc": "2.0",
        "method": "tasks/send",
        "params": {
            "id": task_id,
            "contextId": context_id,
            "message": {
                "role": "user",
                "parts": [{"kind": "text", "text": task}]
            }
        },
        "id": 1
    }
 
    # 프로덕션에서는 try-except와 타임아웃 설정이 필수입니다
    async with httpx.AsyncClient(timeout=60.0) as client:
        async with client.stream(
            "POST",
            f"{agent_url}/a2a",
            json=payload,
            headers={
                "Accept": "text/event-stream",
                "Content-Type": "application/json"
            }
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    event = json.loads(line[6:])
                    yield event

# crm_mcp_server.py
# pip install fastmcp
from fastmcp import FastMCP
 
mcp = FastMCP("crm-server")
 
# 실제 프로젝트에서는 아래를 CRM DB 클라이언트로 교체하세요
# (SQLite, PostgreSQL, 또는 CRM SaaS API 등)
class _MockDB:
    async def find_customer(self, customer_id):
        return None  # 실제 구현체로 교체 필요
 
    async def vector_search(self, query, limit):
        return []
 
    async def create_ticket(self, **kwargs):
        return type("Ticket", (), {"id": "TICKET-001"})()
 
db = _MockDB()
 
@mcp.tool()
async def get_customer_info(customer_id: str) -> dict:
    """고객 ID로 고객 정보를 조회합니다."""
    customer = await db.find_customer(customer_id)
    if not customer:
        return {"error": f"고객 {customer_id}를 찾을 수 없습니다"}
    return {
        "id": customer.id,
        "name": customer.name,
        "email": customer.email,
        "subscription_tier": customer.tier,
        "open_tickets": customer.open_ticket_count
    }
 
@mcp.tool()
async def search_knowledge_base(query: str, top_k: int = 3) -> list[dict]:
    """지식베이스에서 관련 문서를 검색합니다."""
    results = await db.vector_search(query, limit=top_k)
    return [
        {
            "title": doc.title,
            "content": doc.content[:500],
            "relevance_score": doc.score
        }
        for doc in results
    ]
 
@mcp.tool()
async def create_ticket(
    customer_id: str,
    subject: str,
    description: str,
    priority: str = "medium"
) -> dict:
    """새 지원 티켓을 생성합니다."""
    ticket = await db.create_ticket(
        customer_id=customer_id,
        subject=subject,
        description=description,
        priority=priority
    )
    return {"ticket_id": ticket.id, "status": "created"}
 
if __name__ == "__main__":
    mcp.run()

# support_agent.py
# pip install anthropic mcp
import anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
 
async def run_support_agent(customer_id: str, user_message: str):
    server_params = StdioServerParameters(
        command="python",
        args=["crm_mcp_server.py"]
    )
 
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
 
            # MCP 서버에서 도구 목록 동적 조회 — 하드코딩 불필요
            tools_result = await session.list_tools()
            tools = [
                {
                    "name": tool.name,
                    "description": tool.description,
                    "input_schema": tool.inputSchema
                }
                for tool in tools_result.tools
            ]
 
            client = anthropic.Anthropic()
            messages = [{"role": "user", "content": user_message}]
 
            # 에이전트 루프: end_turn이 나올 때까지 도구 호출을 반복
            while True:
                response = client.messages.create(
                    model="claude-sonnet-4-6",
                    max_tokens=1024,
                    tools=tools,
                    messages=messages
                )
 
                if response.stop_reason == "end_turn":
                    text_content = next(
                        block for block in response.content
                        if block.type == "text"
                    )
                    return text_content.text
 
                tool_uses = [
                    block for block in response.content
                    if block.type == "tool_use"
                ]
 
                messages.append({
                    "role": "assistant",
                    "content": response.content
                })
 
                tool_results = []
                for tool_use in tool_uses:
                    result = await session.call_tool(
                        tool_use.name,
                        arguments=tool_use.input
                    )
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": tool_use.id,
                        "content": str(result.content)
                    })
 
                messages.append({
                    "role": "user",
                    "content": tool_results
                })

# researcher_agent.py — A2A 서버 + MCP 클라이언트
# pip install fastapi uvicorn anthropic mcp
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import anthropic
import json
 
app = FastAPI()
 
AGENT_CARD = {
    "name": "ResearchAgent",
    "version": "1.0.0",
    "description": "뉴스 기사 팩트체크와 심층 조사를 담당",
    "url": "http://localhost:8001",
    "capabilities": {"streaming": True, "pushNotifications": False},
    "skills": [
        {
            "id": "fact-check",
            "name": "Fact Check",
            "description": "주장의 사실 여부를 데이터베이스로 검증",
            "inputModes": ["text"],
            "outputModes": ["text"]
        }
    ]
}
 
@app.get("/.well-known/agent.json")
async def get_agent_card():
    """Agent Card 엔드포인트 — 다른 에이전트가 역량을 탐색하는 진입점"""
    return AGENT_CARD

@app.post("/a2a")
async def handle_a2a_request(request: Request):
    body = await request.json()
    task_text = body["params"]["message"]["parts"][0]["text"]
 
    async def generate_response():
        # — MCP 클라이언트 역할: 뉴스 DB MCP 서버에 연결 —
        server_params = StdioServerParameters(
            command="python",
            args=["news_db_mcp_server.py"]
        )
 
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                tools_result = await session.list_tools()
                tools = [
                    {
                        "name": t.name,
                        "description": t.description,
                        "input_schema": t.inputSchema
                    }
                    for t in tools_result.tools
                ]
 
                client = anthropic.Anthropic()
                messages = [{"role": "user", "content": task_text}]
 
                # 작업 시작 상태를 Reporter에게 스트리밍
                yield f"data: {json.dumps({'status': 'working', 'message': '팩트체크 시작'})}\n\n"
 
                while True:
                    response = client.messages.create(
                        model="claude-sonnet-4-6",
                        max_tokens=1024,
                        tools=tools,
                        messages=messages
                    )
 
                    if response.stop_reason == "end_turn":
                        result_text = next(
                            b.text for b in response.content
                            if b.type == "text"
                        )
                        # 최종 결과를 Reporter에게 스트리밍
                        yield f"data: {json.dumps({'status': 'completed', 'result': result_text})}\n\n"
                        break
 
                    # — MCP 도구 호출 처리 —
                    tool_uses = [b for b in response.content if b.type == "tool_use"]
                    messages.append({"role": "assistant", "content": response.content})
 
                    tool_results = []
                    for tool_use in tool_uses:
                        yield f"data: {json.dumps({'status': 'working', 'message': f'{tool_use.name} 조회 중'})}\n\n"
                        result = await session.call_tool(tool_use.name, arguments=tool_use.input)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": tool_use.id,
                            "content": str(result.content)
                        })
 
                    messages.append({"role": "user", "content": tool_results})
 
    return StreamingResponse(generate_response(), media_type="text/event-stream")

# reporter_agent.py
# pip install httpx
import httpx
import json
import uuid
 
async def write_article_with_fact_check(topic: str) -> str:
    """기사 작성 중 팩트체크가 필요한 부분을 Researcher에게 위임합니다."""
 
    # 1. Researcher의 Agent Card를 읽어 역량 확인
    async with httpx.AsyncClient() as client:
        card_response = await client.get(
            "http://localhost:8001/.well-known/agent.json"
        )
        agent_card = card_response.json()
        print(f"협업 에이전트: {agent_card['name']} — {agent_card['description']}")
 
    # 2. A2A로 팩트체크 작업 위임
    fact_check_request = {
        "jsonrpc": "2.0",
        "method": "tasks/send",
        "params": {
            "id": str(uuid.uuid4()),  # uuid4가 충돌 없이 안전합니다
            "message": {
                "role": "user",
                "parts": [{
                    "kind": "text",
                    "text": f"다음 주제에 대한 핵심 사실을 검증해주세요: {topic}"
                }]
            }
        },
        "id": 1
    }
 
    verified_facts = ""  # SSE에서 받는 result는 문자열입니다
    async with httpx.AsyncClient(timeout=60.0) as client:
        async with client.stream(
            "POST",
            "http://localhost:8001/a2a",
            json=fact_check_request,
            headers={"Accept": "text/event-stream"}
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    event = json.loads(line[6:])
                    if event["status"] == "completed":
                        verified_facts = event["result"]
                    elif event["status"] == "working":
                        print(f"  진행 중: {event['message']}")
 
    # 3. 검증된 사실을 바탕으로 기사 작성
    return f"[검증된 정보 기반 기사]\n\n{verified_facts}"

사용자 요청
    │
    ▼
Reporter Agent (MCP Client + A2A Client)
    │
    ├── MCP ──► CRM MCP Server
    │               └── CRM Database
    │
    ├── MCP ──► KnowledgeBase MCP Server
    │               └── Vector DB
    │
    └── A2A ──► Researcher Agent (A2A Server + MCP Client)
                    │
                    └── MCP ──► NewsDB MCP Server
                                    └── Elasticsearch