실제 사용 사례로 시작하기
지난 분기, 저는 한 중소 규모 이커머스 스타트업에서 AI 고객 서비스 시스템을 구축하는 프로젝트를 진행했습니다. 일일 문의량이 5,000건을 돌파하면서 기존 단일 모델 기반 챗봇은 한계에 부딪혔고, 고객은 "주문 상태 조회", "환불 처리", "상품 추천" 등 매우 다양한 요청을 보냈습니다. 정책 문서 요약에는 Claude가 강했고, 다국어 처리에는 Gemini Flash가 압도적으로 빨랐으며, 복잡한 추론에는 GPT-4.1이 안정적이었습니다. 결국 우리는 Model Context Protocol(MCP) 서버를 구축해 여러 모델을 도구(tool)로 노출하고, 사내 지식 베이스 RAG와 결제 시스템 API를 안전하게 연결했습니다. 이 글에서는 그 실전 경험을 바탕으로 FastMCP로 프로덕션 수준의 MCP 서버를 만들고,
Python 3.10 이상과 pip가 설치된 환경을 가정합니다. 이커머스 고객 서비스를 위한 핵심 도구 3개를 정의해 보겠습니다: 주문 조회, 환불 처리, 상품 검색입니다.개발 환경 설정
# 프로젝트 디렉터리 생성 및 가상환경 활성화
mkdir mcp-ecommerce-server && cd mcp-ecommerce-server
python -m venv venv
source venv/bin/activate # Windows는 venv\Scripts\activate
필수 패키지 설치
pip install "fastmcp>=0.4.5" "mcp>=1.2.0" httpx uvicorn "pydantic>=2.5,<3.0" python-dotenvFastMCP 서버 기본 구조
from fastmcp import FastMCP, Context
from pydantic import BaseModel, Field
import httpx
import os
import sys
import logging
from typing import Optional
stderr 로깅 설정 (stdout은 MCP 프로토콜 전용)
logging.basicConfig(level=logging.INFO, stream=sys.stderr)
logger = logging.getLogger("mcp-ecommerce")
MCP 서버 인스턴스 생성
mcp = FastMCP(
name="EcommerceCustomerService",
instructions="이커머스 고객 서비스 도구 모음. 주문 조회, 환불 처리, 상품 검색 기능을 제공합니다."
)
입력 검증을 위한 Pydantic 모델
class OrderQuery(BaseModel):
order_id: str = Field(..., description="조회할 주문 번호")
customer_email: str = Field(..., description="고객 이메일 (인증용)")
class RefundRequest(BaseModel):
order_id: str = Field(..., description="환불할 주문 번호")
reason: str = Field(..., description="환불 사유")
amount: Optional[float] = Field(None, description="부분 환불 금액 (전액 환불 시 None)")
도구 1: 주문 조회
@mcp.tool()
async def get_order_status(query: OrderQuery, ctx: Context) -> dict:
"""주문 상태와 배송 정보를 조회합니다."""
await ctx.info(f"주문 {query.order_id} 조회 요청 수신")
# 내부 ERP API 호출 (예시)
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"https://erp.internal/api/orders/{query.order_id}",
headers={"X-Customer-Email": query.customer_email}
)
response.raise_for_status()
data = response.json()
await ctx.report_progress(progress=100, total=100)
return {
"order_id": data["id"],
"status": data["status"],
"tracking_number": data.get("tracking"),
"estimated_delivery": data.get("eta")
}
도구 2: 환불 처리 (Human-in-the-loop)
@mcp.tool()
async def process_refund(req: RefundRequest, ctx: Context) -> dict:
"""환불을 처리하고 트랜