AI Agent의 MCP: 에이전트가 외부 도구를 연결하는 표준 프로토콜

M 에이전트 × N 서비스 = M×N 커스텀 통합

에이전트 3개 × 서비스 10개 = 30개 커스텀 통합

M 에이전트 + N 서비스 = M + N 구현

에이전트 3개 + 서비스 10개 = 13개 구현

┌─────────────────────────────────────────┐
│            Host (에이전트 앱)              │
│                                         │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│  │ Client 1 │ │ Client 2 │ │ Client 3 │ │
│  └────┬─────┘ └────┬─────┘ └────┬─────┘ │
└───────┼────────────┼────────────┼───────┘
        │            │            │
   ┌────▼─────┐ ┌────▼─────┐ ┌───▼──────┐
   │ Server 1 │ │ Server 2 │ │ Server 3 │
   │ (GitHub) │ │ (Slack)  │ │ (DB)     │
   └────┬─────┘ └────┬─────┘ └────┬─────┘
        │            │             │
   ┌────▼─────┐ ┌────▼─────┐ ┌────▼─────┐
   │ GitHub   │ │ Slack    │ │ PostgreSQL│
   │ API      │ │ API      │ │          │
   └──────────┘ └──────────┘ └──────────┘

Client                          Server
  │                               │
  │──── initialize ──────────────▶│  1. 초기화: 프로토콜 버전과
  │◀─── capabilities ────────────│     지원 기능을 교환
  │──── initialized ────────────▶│     초기화 완료 통지
  │                               │
  │──── tools/list ──────────────▶│  2. 작업: 도구/리소스 탐색
  │◀─── tool definitions ────────│
  │──── tools/call ──────────────▶│     도구 실행
  │◀─── result ──────────────────│
  │          ...                  │
  │                               │
  │──── (close transport) ───────▶│  3. 종료: stdin 닫기 또는
  │                               │     HTTP 연결 종료

# 전통적 방식: 모든 도구 스키마를 미리 로딩
tools = [github_tools + slack_tools + sentry_tools]  # ~72K tokens

# Tool Search 방식: 메타 도구만 로딩
tools = [
    tool_search_tool,     # ~500 tokens
    frequently_used_tools  # ~3K tokens (defer_loading: false인 도구만)
]
# 필요할 때 동적으로 로딩: ~3-5개 도구 = ~5K tokens

# 전통적 방식: 20번의 모델 추론, 중간 결과 모두 컨텍스트에 축적
# Turn 1: 모델 → get_team_members() → 결과 5명 → 컨텍스트에 추가
# Turn 2: 모델 → get_expenses(member_1) → 결과 400건 → 컨텍스트에 추가
# Turn 3: 모델 → get_expenses(member_2) → 결과 350건 → 컨텍스트에 추가
# ... (20턴, ~50KB의 중간 결과가 컨텍스트에 쌓임)

# Programmatic 방식: 모델이 코드를 한 번 작성, 샌드박스에서 실행
async def check_budget_compliance():
    members = await get_team_members()
    tasks = [get_expenses(m.id) for m in members]
    all_expenses = await asyncio.gather(*tasks)  # 병렬 실행

    over_budget = []
    for member, expenses in zip(members, all_expenses):
        total = sum(e.amount for e in expenses)
        if total > member.budget:
            over_budget.append(f"{member.name}: {total}/{member.budget}")

    print("\n".join(over_budget))  # stdout만 컨텍스트로 반환

import json
import asyncio
from dataclasses import dataclass, field

@dataclass
class MCPTool:
    name: str
    description: str
    input_schema: dict
    server_name: str
    annotations: dict = field(default_factory=dict)

@dataclass
class MCPClient:
    server_name: str
    command: list[str]
    process: asyncio.subprocess.Process | None = None
    capabilities: dict = field(default_factory=dict)
    tools: list[MCPTool] = field(default_factory=list)
    _request_id: int = 0

    async def connect(self):
        self.process = await asyncio.create_subprocess_exec(
            *self.command,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        response = await self._send("initialize", {
            "protocolVersion": "2025-06-18",
            "capabilities": {"roots": {"listChanged": True}},
            "clientInfo": {"name": "mini-agent", "version": "0.1"},
        })
        self.capabilities = response.get("capabilities", {})
        await self._notify("notifications/initialized", {})

    async def discover_tools(self) -> list[MCPTool]:
        if "tools" not in self.capabilities:
            return []
        response = await self._send("tools/list", {})
        self.tools = [
            MCPTool(
                name=t["name"],
                description=t.get("description", ""),
                input_schema=t.get("inputSchema", {}),
                server_name=self.server_name,
                annotations=t.get("annotations", {}),
            )
            for t in response.get("tools", [])
        ]
        return self.tools

    async def call_tool(self, name: str, arguments: dict) -> dict:
        return await self._send("tools/call", {
            "name": name,
            "arguments": arguments,
        })

    async def _send(self, method: str, params: dict) -> dict:
        self._request_id += 1
        message = json.dumps({
            "jsonrpc": "2.0",
            "id": self._request_id,
            "method": method,
            "params": params,
        }) + "\n"
        self.process.stdin.write(message.encode())
        await self.process.stdin.drain()

        line = await self.process.stdout.readline()
        response = json.loads(line)
        if "error" in response:
            raise RuntimeError(f"MCP error: {response['error']}")
        return response.get("result", {})

    async def _notify(self, method: str, params: dict):
        message = json.dumps({
            "jsonrpc": "2.0",
            "method": method,
            "params": params,
        }) + "\n"
        self.process.stdin.write(message.encode())
        await self.process.stdin.drain()

    async def disconnect(self):
        if self.process:
            self.process.terminate()
            await self.process.wait()

@dataclass
class ToolRegistry:
    clients: list[MCPClient] = field(default_factory=list)
    deferred_tools: dict[str, MCPTool] = field(default_factory=dict)
    active_tools: dict[str, MCPTool] = field(default_factory=dict)

    async def register_server(self, client: MCPClient, defer: bool = True):
        await client.connect()
        tools = await client.discover_tools()
        self.clients.append(client)

        for tool in tools:
            if defer:
                self.deferred_tools[tool.name] = tool
            else:
                self.active_tools[tool.name] = tool

    def search(self, query: str, max_results: int = 5) -> list[MCPTool]:
        query_lower = query.lower()
        scored = []
        for name, tool in self.deferred_tools.items():
            score = 0
            if query_lower in name.lower():
                score += 2
            if query_lower in tool.description.lower():
                score += 1
            if score > 0:
                scored.append((score, tool))

        scored.sort(key=lambda x: -x[0])
        found = [tool for _, tool in scored[:max_results]]

        for tool in found:
            self.active_tools[tool.name] = tool
            self.deferred_tools.pop(tool.name, None)

        return found

    def get_active_schemas(self) -> list[dict]:
        return [
            {
                "name": t.name,
                "description": t.description,
                "input_schema": t.input_schema,
            }
            for t in self.active_tools.values()
        ]

async def agent_loop_with_mcp(
    task: str,
    registry: ToolRegistry,
    llm_call,
    max_turns: int = 20,
):
    search_tool = {
        "name": "tool_search",
        "description": "Search for available tools by keyword",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"},
            },
            "required": ["query"],
        },
    }

    messages = [{"role": "user", "content": task}]
    active_schemas = registry.get_active_schemas()
    tools = [search_tool] + active_schemas

    for turn in range(max_turns):
        response = await llm_call(messages=messages, tools=tools)
        messages.append({"role": "assistant", "content": response.content})

        tool_calls = [b for b in response.content if b.type == "tool_use"]
        if not tool_calls:
            return response.content

        for call in tool_calls:
            if call.name == "tool_search":
                found = registry.search(call.input["query"])
                result = "\n".join(
                    f"- {t.name}: {t.description}" for t in found
                )
                tools = [search_tool] + registry.get_active_schemas()
            else:
                tool = registry.active_tools.get(call.name)
                if not tool:
                    result = f"Error: tool '{call.name}' not found"
                else:
                    client = next(
                        c for c in registry.clients
                        if c.server_name == tool.server_name
                    )
                    response_data = await client.call_tool(
                        call.name, call.input
                    )
                    result = json.dumps(response_data)

            messages.append({
                "role": "tool",
                "tool_use_id": call.id,
                "content": result,
            })