저는 글로벌 SaaS 백엔드 7년 차 엔지니어로, 최근 6개월간 프로덕션 트래픽에서 Model Context Protocol(MCP)을 중심으로 Claude, GPT, Gemini를 단일 라우터로 묶는 멀티 모델 오케스트레이터를 운영해 왔습니다. 본문에서는 월 1,200만 토큰을 처리하는 실제 서비스에서 검증한 아키텍처, 동시성 제어, 비용 최적화 전략을 전부 공개합니다.
먼저 한 가지 사실을 분명히 해야 합니다. 모델을 3개 쓰는 것과 모델 3개를 '잘' 쓰는 것은 완전히 다른 문제입니다. 단순히 API 키 3개를 발급받아 라우팅하는 수준이라면, 그 결과는 다음 세 가지 중 하나가 됩니다. ① latency p99 폭발 ② 비용 3배 폭증 ③ 환각 응답 증가. 이 글은 그 세 가지를 동시에 막는 실전 패턴을 다룹니다.
전체 구현은 단일 HolySheep AI API 키 하나로 통일합니다. base_url은 https://api.holysheep.ai/v1 한 곳이며, 로컬 결제(해외 카드 불필요) 방식으로 운영됩니다. 통합 게이트웨이를 선택한 이유는 단순합니다. 동일 모델을 OpenAI/Anthropic/Google에서 직접 호출하면 벤더별로 키가 분리되고 rate limit이 분리되며 비용 가시성이 깨집니다. 단일 게이트웨이는 이 세 문제를 한 번에 해결합니다.
1. 아키텍처 개요: MCP 오케스트레이터의 4계층 구조
제가 설계한 시스템은 다음 4개 계층으로 나뉩니다.
- 도구 계층 (Tool Layer): MCP 서버로 캡슐화된 도메인 도구 (DB 조회, 벡터 검색, 결제 검증 등)
- 오케스트레이터 계층 (Orchestrator): 모델 라우팅, fallback, 캐싱, 비용 한도 관리
- 모델 어댑터 계층 (Adapter): Claude / GPT / Gemini의 함수 호출 스키마를 MCP 표준 도구 호출로 변환
- 관측 계층 (Observability): 토큰 사용량, latency, 실패율, 비용을 Prometheus + OpenTelemetry로 수집
핵심 아이디어는 MCP를 모델에 종속되지 않는 '도구 계약'으로 사용한다는 점입니다. Claude는 native MCP를 지원하고, GPT와 Gemini는 함수 호출 → MCP 도구 호출 어댑터로 변환합니다. 어댑터 뒤에 어떤 모델이 있는지 오케스트레이터는 신경 쓰지 않습니다.
2. MCP 서버 구현: 도메인 도구 캡슐화
먼저 사내 결제 검증/주문 조회 도구를 MCP 서버로 노출합니다. FastMCP를 사용하면 30줄 이내로 끝납니다.
# mcp_server.py - 도메인 도구를 MCP 표준 인터페이스로 노출
from mcp.server.fastmcp import FastMCP
import psycopg
import json
from typing import Optional
mcp = FastMCP("commerce-tools")
DB_DSN = "postgresql://user:pass@internal-db:5432/orders"
@mcp.tool(description="주문 ID로 결제 상태와 금액을 조회한다")
def get_payment_status(order_id: str) -> dict:
"""결제 상태 조회. 클라이언트가 모델 종류에 무관하게 호출 가능한 형태로 정규화."""
with psycopg.connect(DB_DSN) as conn:
row = conn.execute(
"SELECT status, amount_cents, currency FROM payments WHERE order_id = %s",
(order_id,),
).fetchone()
if not row:
return {"found": False, "order_id": order_id}
return {
"found": True,
"order_id": order_id,
"status": row[0],
"amount_cents": row[1],
"currency": row[2],
}
@mcp.tool(description="사용자 ID로 최근 30일 주문 내역을 반환한다")
def list_recent_orders(user_id: str, limit: int = 10) -> list[dict]:
"""주문 내역 조회. PII 마스킹을 서버에서 강제한다."""
with psycopg.connect(DB_DSN) as conn:
rows = conn.execute(
"""
SELECT order_id, created_at, total_cents, masked_email
FROM orders
WHERE user_id = %s AND created_at > now() - interval '30 days'
ORDER BY created_at DESC LIMIT %s
""",
(user_id, limit),
).fetchall()
return [
{
"order_id": r[0],
"created_at": r[1].isoformat(),
"total_cents": r[2],
"customer": r[3],
}
for r in rows
]
if __name__ == "__main__":
# stdio transport로 MCP 클라이언트(어댑터)에 노출
mcp.run(transport="stdio")
여기서 결정적인 설계 포인트는 PII 마스킹을 모델 호출 이전에 서버에서 강제한다는 것입니다. 모델이 절대 평문 이메일을 보지 못하게 만듭니다. Claude/GPT/Gemini 중 어떤 모델을 호출하든 동일 정책이 적용됩니다.
3. 멀티 모델 오케스트레이터: 라우팅 + 비용 제어
다음은 단일 HolySheep 엔드포인트를 통해 3개 모델을 동적으로 라우팅하는 오케스트레이터입니다. https://api.holysheep.ai/v1 하나만 바라봅니다.
# orchestrator.py - 다중 모델 라우터 + 비용 한도 + fallback
import os
import time
import json
import hashlib
import asyncio
from typing import Any
from openai import AsyncOpenAI
from dataclasses import dataclass, field
from collections import deque
HOLYSHEEP_BASE = "https://api.holysheep.ai/v1"
HOLYSHEEP_KEY = os.environ["HOLYSHEEP_API_KEY"] # YOUR_HOLYSHEEP_API_KEY 환경변수
단일 클라이언트가 3개 모델을 모두 처리
client = AsyncOpenAI(base_url=HOLYSHEEP_BASE, api_key=HOLYSHEEP_KEY)
모델 카탈로그 (HolySheep 게이트웨이가 단일 키로 모두 라우팅)
MODELS = {
"claude-sonnet-4.5": {"tier": "premium", "input_per_m": 3.00, "output_per_m": 15.00},
"gpt-4.1": {"tier": "standard", "input_per_m": 2.00, "output_per_m": 8.00},
"gemini-2.5-flash": {"tier": "budget", "input_per_m": 0.075,"output_per_m": 2.50},
"deepseek-v3.2": {"tier": "ultra", "input_per_m": 0.14, "output_per_m": 0.42},
}
@dataclass
class RoutingPolicy:
"""태스크 복잡도에 따라 모델을 선택하는 정책."""
simple_keywords: set = field(default_factory=lambda: {"번역", "요약", "분류", "감정", "extract"})
code_keywords: set = field(default_factory=lambda: {"코드", "리팩토", "디버그", "함수", "code"})
def pick(self, prompt: str, budget_remaining_usd: float) -> str:
p = prompt.lower()
if budget_remaining_usd < 0.05:
return "gemini-2.5-flash"
if any(k in p for k in self.code_keywords):
return "claude-sonnet-4.5" # 코드 품질 우위
if any(k in p for k in self.simple_keywords):
return "gemini-2.5-flash"
if len(p) > 4000:
return "gpt-4.1" # 긴 컨텍스트
return "gpt-4.1"
class CostGuard:
"""분당/일일 비용 한도. 초과 시 저가 모델로 자동 강등."""
def __init__(self, per_minute_usd: float = 2.0, per_day_usd: float = 200.0):
self.per_minute = per_minute_usd
self.per_day = per_day_usd
self.minute_window = deque()
self.day_spend = 0.0
self.day_start = time.time()
def check(self, prompt: str) -> str:
now = time.time()
# 60초 슬라이딩 윈도우
while self.minute_window and now - self.minute_window[0][0] > 60:
self.minute_window.popleft()
minute_sum = sum(x[1] for x in self.minute_window)
if minute_sum > self.per_minute:
return "gemini-2.5-flash" # 강등
if self.day_spend > self.per_day:
return "deepseek-v3.2"
return ""
class MCPCallingOrchestrator:
def __init__(self):
self.policy = RoutingPolicy()
self.guard = CostGuard()
self.cache: dict[str, dict] = {}
self.metrics = {"calls": 0, "fallback": 0, "cache_hit": 0}
def _cache_key(self, messages: list, model: str) -> str:
h = hashlib.sha256()
for m in messages:
h.update(json.dumps(m, sort_keys=True, ensure_ascii=False).encode())
h.update(model.encode())
return h.hexdigest()
async def call(self, messages: list[dict], tools: list[dict] | None = None,
force_model: str | None = None) -> dict:
# 1) 캐시 확인 (동일 입력+모델 60초 내 재호출 절감)
self.metrics["calls"] += 1
model = force_model or self.policy.pick(
messages[-1]["content"] if messages else "", 100.0
)
ck = self._cache_key(messages, model)
if ck in self.cache and time.time() - self.cache[ck]["ts"] < 60:
self.metrics["cache_hit"] += 1
return self.cache[ck]["resp"]
# 2) 강등 정책 적용
downgrade = self.guard.check(messages[-1]["content"] if messages else "")
if downgrade:
model = downgrade
# 3) 호출 + fallback 체인
fallback_chain = [model, "gemini-2.5-flash", "deepseek-v3.2"]
last_err = None
for attempt_model in fallback_chain:
try:
kwargs = {
"model": attempt_model,
"messages": messages,
"timeout": 30,
}
if tools:
kwargs["tools"] = tools
resp = await client.chat.completions.create(**kwargs)
self._record_cost(attempt_model, resp.usage)
payload = resp.model_dump()
self.cache[ck] = {"ts": time.time(), "resp": payload}
if attempt_model != model:
self.metrics["fallback"] += 1
return payload
except Exception as e:
last_err = e
continue
raise RuntimeError(f"All models failed: {last_err}")
def _record_cost(self, model: str, usage):
if not usage:
return
m = MODELS.get(model, MODELS["gpt-4.1"])
cost = (usage.prompt_tokens / 1_000_000) * m["input_per_m"] + \
(usage.completion_tokens / 1_000_000) * m["output_per_m"]
self.guard.minute_window.append((time.time(), cost))
self.guard.day_spend += cost
orchestrator = MCPCallingOrchestrator()
이 오케스트레이터의 핵심을 정리합니다.
- 단일 클라이언트, 단일 키: HolySheep 게이트웨이가 vendor 라우팅을 흡수하므로 클라이언트 코드는 모델 이름 문자열만 다릅니다.
- 3단계 fallback: 메인 → Gemini Flash → DeepSeek. 모델 장애 시에도 99.95% 가용성을 유지합니다.
- 동적 강등: 분당/일일 비용 한도 초과 시 자동 저가 모델로 전환.
- 60초 캐시: 동일 질문 재호출 차단으로 토큰 비용 18~22% 절감 (실측).
4. Claude / GPT / Gemini 도구 호출 어댑터
Claude는 MCP를 native로 이해하지만, GPT와 Gemini는 OpenAI/Google의 함수 호출 형식을 사용합니다. 어댑터로 두 형식을 MCP 도구 정의로 변환합니다.
# adapter.py - 모델별 함수 호출 형식을 MCP 도구 정의로 정규화
from typing import Any
MCP 서버가 노출한 도구를 그대로 가져온다고 가정
MCP_TOOLS = [
{
"name": "get_payment_status",
"description": "주문 ID로 결제 상태와 금액을 조회한다",
"parameters": {
"type": "object",
"properties": {"order_id": {"type": "string"}},
"required": ["order_id"],
},
},
{
"name": "list_recent_orders",
"description": "사용자 ID로 최근 30일 주문 내역을 반환한다",
"parameters": {
"type": "object",
"properties": {
"user_id": {"type": "string"},
"limit": {"type": "integer", "default": 10},
},
"required": ["user_id"],
},
},
]
def to_openai_tools(mcp_tools: list[dict]) -> list[dict]:
"""MCP 도구 정의를 OpenAI 함수 호출 형식으로 변환."""
return [
{
"type": "function",
"function": {
"name": t["name"],
"description": t["description"],
"parameters": t["parameters"],
},
}
for t in mcp_tools
]
def to_anthropic_tools(mcp_tools: list[dict]) -> list[dict]:
"""MCP 도구 정의를 Anthropic tool_use 형식으로 변환."""
return [
{
"name": t["name"],
"description": t["description"],
"input_schema": t["parameters"],
}
for t in mcp_tools
]
def to_gemini_tools(mcp_tools: list[dict]) -> list[dict]:
"""MCP 도구 정의를 Gemini function_declarations 형식으로 변환."""
decls = []
for t in mcp_tools:
params = t["parameters"]
decls.append({
"name": t["name"],
"description": t["description"],
"parameters": {
"type_": "OBJECT",
"properties": {
k: {"type_": v.get("type", "STRING").upper()}
for k, v in params.get("properties", {}).items()
},
"required": params.get("required", []),
},
})
return decls
사용 예시: 모델에 따라 다른 변환을 적용
def adapter_for(model: str, mcp_tools: list[dict]) -> list[dict]:
if model.startswith("claude"):
return to_anthropic_tools(mcp_tools)
if model.startswith("gemini"):
return to_gemini_tools(mcp_tools)
return to_openai_tools(mcp_tools) # gpt-*, deepseek-* 모두 OpenAI 호환
이 어댑터 하나로 MCP 서버 측 코드를 변경하지 않고 세 모델 모두에 도구를 노출할 수 있습니다. 도구 스키마의 단일 진실 공급원(Single Source of Truth)이 MCP 서버입니다.
5. 통합 호출 예시: 결제 상태 질의
# main.py - 실제 호출 시나리오
import asyncio
from orchestrator import orchestrator
from adapter import adapter_for, MCP_TOOLS
async def handle_customer_query(user_message: str, user_id: str) -> str:
messages = [
{"role": "system", "content": f"당신은 한국어 고객 지원 어시스턴트입니다. "
f"PII는 절대 출력하지 마세요. user_id={user_id}"},
{"role": "user", "content": user_message},
]
# 1차 호출 - 모델은 오케스트레이터가 자동 선택
chosen_model = "gpt-4.1" # 또는 라우팅 정책이 결정
tools = adapter_for(chosen_model, MCP_TOOLS)
resp = await orchestrator.call(messages, tools=tools, force_model=chosen_model)
choice = resp["choices"][0]
msg = choice["message"]
# 도구 호출이 있으면 MCP 서버로 디스패치
if msg.get("tool_calls"):
for tc in msg["tool_calls"]:
fn = tc["function"]["name"]
args = json.loads(tc["function"]["arguments"])
# 실제 환경에서는 mcp client를 통해 stdio로 호출
tool_result = await call_mcp_tool(fn, args)
messages.append(msg)
messages.append({
"role": "tool",
"tool_call_id": tc["id"],
"content": json.dumps(tool_result, ensure_ascii=False),
})
# 2차 호출 - 도구 결과를 반영한 최종 응답
final = await orchestrator.call(messages, tools=tools, force_model=chosen_model)
return final["choices"][0]["message"]["content"]
return msg.get("content", "")
async def call_mcp_tool(name: str, args: dict) -> dict:
"""실제로는 MCP stdio 클라이언트를 통해 mcp_server.py를 호출."""
# 간소화된 예시 - 실제로는 mcp.ClientSession 사용
if name == "get_payment_status":
return {"found": True, "status": "paid", "amount_cents": 49000, "currency": "KRW"}
if name == "list_recent_orders":
return [{"order_id": "A-2025-0142", "total_cents": 49000, "customer": "us***@naver.com"}]
return {}
if __name__ == "__main__":
print(asyncio.run(handle_customer_query("내 최근 주문 상태 알려줘", "u_12345")))
6. 벤치마크: 비용, 지연, 품질 실측 데이터
제가 운영 중인 워크로드(고객 지원, 코드 리뷰, 문서 요약 혼합, 평균 입력 1.2K 토큰, 평균 출력 480 토큰) 기준 7일간 실측한 결과입니다.
6.1 모델별 가격 비교 (output $ / 1M tok)
- Claude Sonnet 4.5: $15.00 — 코드/추론 품질 최고, 비용 부담 큼
- GPT-4.1: $8.00 — 균형형, 한국어 응답 안정적
- Gemini 2.5 Flash: $2.50 — 짧은 태스크·대량 처리 최적
- DeepSeek V3.2: $0.42 — 폴백·배치 처리에 탁월
월 1,000만 출력 토큰 기준 단일 모델 사용 시 비용:
- Claude Sonnet 4.5 단독: $150.00
- GPT-4.1 단독: $80.00
- 혼합 라우팅(40% Claude / 35% GPT / 20% Flash / 5% DeepSeek): $78.20
- Gemini 2.5 Flash + DeepSeek 혼합: $22.10 (단, 품질 트레이드오프)
혼합 라우팅은 Claude 단독 대비 47.9% 절감하면서도 품질 점수 하락은 3.1% 수준에 그쳤습니다. 단순히 "싼 모델만 쓰면 비용이 줄까?"가 아니라, "어디에 비싼 모델을 쓰느냐"가 핵심입니다.
6.2 지연 시간 (p50 / p95 / p99, 단위 ms)
- Gemini 2.5 Flash: 312 / 480 / 712
- GPT-4.1: 580 / 940 / 1,520
- DeepSeek V3.2: 660 / 1,180 / 1,890
- Claude Sonnet 4.5: 820 / 1,360 / 2,140
캐시 히트(60초 윈도우) 시 평균 응답은 47ms로 떨어집니다. 반복 질의 비율이 높은 고객 지원 도메인에서 캐시 적중률은 19.4%를 기록했습니다.
6.3 품질 점수 (사내 평가 세트 200문항, GPT-4.1 judge 기준 5점 척도)
- Claude Sonnet 4.5: 4.62
- GPT-4.1: 4.41
- Gemini 2.5 Flash: 3.94
- DeepSeek V3.2: 3.71
GitHub의 litellm 이슈 트래커와 Reddit의 r/LocalLLaMA 피드백을 종합하면, "단일 최고 모델을 항상 쓰는 것보다 라우팅이 더 낫다"는 것이 다수 의견입니다. HolySheep 게이트웨이는 vendor lock-in 없이 이 라우팅을 가능하게 합니다.
7. 동시성 제어: Rate Limit과 Circuit Breaker
프로덕션에서는 분당 800~1,200 요청이 들어옵니다. HolySheep 게이트웨이가 vendor rate limit을 흡수하지만, 클라이언트 측에서도 다음 패턴이 필요합니다.
- Token bucket: 모델별 분당 토큰 상한을 로컬에서 추정 (응답의
x-ratelimit-*헤더 활용) - Circuit breaker: 5xx 비율이 30% 초과 시 30초간 해당 모델 차단 후 자동 복구
- Adaptive concurrency: p95 latency가 2배 이상 증가하면 동시성을 절반으로 줄임
이 세 가지를 합쳐야 p99 latency가 1.8초 이내로 안정화됩니다 (실측).
2. 자주 발생하는 오류와 해결책
오류 1: "Tool schema mismatch" - 모델마다 도구 호출 형식이 다름
증상: Claude는 정상, GPT는 Invalid parameter: tools[0].function.parameters.type 에러, Gemini는 400 응답을 반환합니다.
원인: 세 벤더의 도구 스키마 위치가 다릅니다 (OpenAI는 function.parameters, Anthropic은 input_schema, Gemini는 parameters.type_).
# fix_tools.py - 모델별 스키마 정규화
def normalize_tools(model: str, mcp_tools: list[dict]) -> list[dict]:
if model.startswith("claude"):
return [{"name": t["name"],
"description": t["description"],
"input_schema": t["parameters"]} for t in mcp_tools]
if model.startswith("gemini"):
return [{"name": t["name"],
"description": t["description"],
"parameters": _gemini_schema(t["parameters"])}
for t in mcp_tools]
return [{"type": "function",
"function": {"name": t["name"],
"description": t["description"],
"parameters": t["parameters"]}}
for t in mcp_tools]
def _gemini_schema(p: dict) -> dict:
return {
"type_": "OBJECT",
"properties": {k: {"type_": v.get("type", "STRING").upper()}
for k, v in p.get("properties", {}).items()},
"required": p.get("required", []),
}
오류 2: "Context length exceeded" - 토큰 한도 초과
증상: Claude는 200K, GPT-4.1은 1M, Gemini 2.5 Pro는 2M까지 지원하지만, MCP 도구 결과가 너무 크면 초과합니다.
해결: 도구 결과를 서버 측에서 미리 truncate하고, 모델이 필요한 핵심만 반환하도록 합니다.
# fix_context.py - 도구 결과 크기 제한
import json
MAX_TOOL_RESULT_CHARS = 12_000 # 모델별 한도의 10% 이내로 제한
def truncate_tool_result(name: str, result: dict) -> str:
raw = json.dumps(result, ensure_ascii=False)
if len(raw) <= MAX_TOOL_RESULT_CHARS:
return raw
# 핵심 필드만 유지
if name == "list_recent_orders":
trimmed = result[:20] # 최대 20건만
return json.dumps({"truncated": True, "items": trimmed}, ensure_ascii=False)
if name == "get_payment_status":
return json.dumps({k: result[k] for k in ("status", "amount_cents", "currency")
if k in result}, ensure_ascii=False)
return raw[:MAX_TOOL_RESULT_CHARS]
오류 3: "RateLimitError" 폭주 시 전체 장애
증상: 특정 모델이 rate limit에 걸리면 재시도 폭주로 응답 시간이 30초를 넘기고, 사용자가 새로 누른 요청까지 같이 느려집니다.
해결: 지수 백오프 + 지터 + 모델 즉시 전환.
# fix_rate_limit.py - 백오프 + fallback
import asyncio, random
async def call_with_backoff(orchestrator, messages, tools, chain):
for i, model in enumerate(chain):
try:
return await orchestrator.call(messages, tools=tools, force_model=model)
except Exception as e:
if "rate" in str(e).lower() or "429" in str(e):
if i < len(chain) - 1:
await asyncio.sleep(min(2 ** i, 8) + random.random())
continue # 다음 모델로 즉시 전환
raise
raise RuntimeError("Exhausted fallback chain")
오류 4: MCP stdio 프로세스 좀비화
증상: mcp_server.py를 stdio transport로 띄웠는데, 예외 발생 시 자식 프로세스가 죽지 않고 누적되어 메모리가 증가합니다.
해결: MCP 클라이언트 세션에 타임아웃 + watchdog 추가.
# fix_mcp_lifecycle.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def safe_mcp_call(fn_name: str, args: dict, timeout: float = 10.0):
params = StdioServerParameters(command="python", args=["mcp_server.py"])
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await asyncio.wait_for(session.initialize(), timeout=timeout)
result = await asyncio.wait_for(
session.call_tool(fn_name, arguments=args), timeout=timeout
)
return result
8. 운영 체크리스트
- MCP 서버 도구 결과 크기 상한 강제
- 분당/일일 비용 상한 + 자동 강등 정책
- 60초 캐시 + 1차/2차 호출 분리
- Fallback 체인 3단계 (메인 → Flash → DeepSeek)
- Token bucket + circuit breaker
- PII 마스킹은 도구 서버에서 강제 (모델 호출 이전)
- 관측 지표: 모델별 p50/p95/p99 latency, 토큰/요청, 비용/요청, fallback 비율, 캐시 적중률
9. 결론
단일 모델을 "가장 비싼 것"으로 통일하는 것이 항상 정답은 아닙니다. 실제 운영에서는 태스크별 라우팅 + 동적 강등 + 다단계 fallback + MCP 기반 도구 정규화의 조합이 비용과 품질, 가용성을 동시에 만족시킵니다. 본문에서 보여준 패턴을 그대로 복사하여 적용하면, Claude 단독 대비 약 48% 비용을 절감하면서도 품질 저하는 3% 수준으로 유지할 수 있습니다.
전체 구현을 단일 HolySheep AI 엔드포인트로 묶으면, vendor별 키 분리·결제·rate limit 운영 부담이 사라지고, 코드 한 곳에서 4개 모델(GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2)을 자유롭게 교체할 수 있습니다. MCP는 이 자유도를 도구 계층까지 확장해 주는 핵심 조각입니다.