MCP(Model Context Protocol) 완벽 가이드 - AI 에이전트 통합의 새로운 표준
AI 에이전트를 실무에 적용하다 보면 가장 큰 벽에 부딪히게 되는데요, 바로 "어떻게 안전하고 효율적으로 외부 데이터나 도구와 연결할 것인가"입니다. 각 LLM 제공자마다 다른 방식의 API를 사용하고, 보안 정책도 제각각이다 보니 통합 작업이 매번 악몽이죠. Anthropic이 발표한 MCP(Model Context Protocol)는 바로 이 문제를 해결하기 위한 오픈 표준 프로토콜이에요. 이 글에서는 MCP의 핵심 개념부터 실전 서버 구축까지, 실무에서 바로 활용할 수 있는 수준으로 깊이 있게 다뤄볼게요.
MCP란 무엇인가? - AI 에이전트 통합의 게임 체인저
MCP(Model Context Protocol)는 AI 애플리케이션과 외부 데이터 소스 및 도구를 연결하는 표준화된 프로토콜입니다. 쉽게 말해 USB처럼 어떤 AI 모델이든 같은 방식으로 데이터베이스, API, 파일 시스템 등과 연결할 수 있게 해주는 거예요.
MCP의 핵심 장점
1. 표준화된 인터페이스
- 한 번 MCP 서버를 구축하면 Claude, GPT 등 다양한 LLM에서 재사용 가능
- 각 LLM 제공자의 커스텀 통합 방식을 배울 필요 없음
2. 보안과 권한 관리
- 리소스별 세밀한 접근 제어
- 클라이언트-서버 아키텍처로 격리된 실행 환경 제공
3. 양방향 통신
- AI가 데이터를 읽는 것뿐만 아니라 도구를 실행하고 결과를 받을 수 있음
- 실시간 프롬프트 업데이트와 컨텍스트 관리
MCP가 없었다면 Slack 연동, Notion 연동, 데이터베이스 연동을 각각 다른 방식으로 구현해야 했을 거예요. 하지만 MCP를 사용하면 하나의 통일된 방식으로 모든 통합을 처리할 수 있습니다.
MCP 아키텍처 이해하기 - 클라이언트와 서버의 역할
MCP는 클라이언트-서버 모델로 동작해요. 명확한 역할 분리가 핵심입니다.
클라이언트 (MCP Host)
- AI 애플리케이션이 실행되는 환경 (예: Claude Desktop, IDE 플러그인)
- 사용자의 요청을 받아 MCP 서버와 통신
- 여러 MCP 서버를 동시에 연결 가능
서버 (MCP Server)
- 특정 데이터 소스나 도구에 대한 접근 제공
- Resources(읽기), Prompts(템플릿), Tools(실행) 세 가지 주요 기능 제공
- 독립적인 프로세스로 실행되어 보안 격리
통신 프로토콜
MCP는 JSON-RPC 2.0 기반으로 stdio(표준 입출력) 또는 HTTP/SSE(Server-Sent Events)를 전송 계층으로 사용합니다. stdio 방식이 로컬 개발에서 가장 간단하고 일반적이에요.
┌─────────────┐ ┌──────────────┐ ┌──────────────┐
│ Claude │ ◄────────► MCP Client │ ◄────────► MCP Server │
│ (AI Model) │ Prompts │ (Host) │ JSON-RPC │ (Your Code) │
└─────────────┘ └──────────────┘ └──────────────┘
│
▼
┌─────────────────┐
│ Database / API │
└─────────────────┘MCP 서버의 세 가지 핵심 기능
MCP 서버가 제공할 수 있는 기능은 크게 세 가지로 나뉩니다. 각각의 특징과 사용 시나리오를 이해하는 것이 중요해요.
1. Resources - 컨텍스트 제공 (읽기)
Resources는 AI가 참조할 수 있는 데이터를 제공합니다. 파일, 데이터베이스 레코드, API 응답 등이 해당돼요.
주요 특징:
- URI 스키마로 리소스 식별 (예:
file:///path/to/doc.txt,db://users/123) - 읽기 전용 데이터 제공
- 텍스트 또는 바이너리 데이터 모두 지원
사용 시나리오:
- 회사 문서 라이브러리를 AI에게 제공
- 고객 데이터베이스 조회
- 실시간 로그 파일 분석
2. Prompts - 재사용 가능한 템플릿
Prompts는 미리 정의된 프롬프트 템플릿을 제공합니다. 자주 사용하는 작업 패턴을 표준화할 수 있어요.
주요 특징:
- 파라미터화된 프롬프트 템플릿
- 컨텍스트 자동 주입
- 일관된 AI 응답 품질 유지
사용 시나리오:
- 코드 리뷰 템플릿
- 버그 리포트 분석 템플릿
- 데이터 요약 표준 포맷
3. Tools - 실행 가능한 작업
Tools는 AI가 직접 실행할 수 있는 함수나 작업을 제공합니다. Function calling의 MCP 버전이라고 보면 돼요.
주요 특징:
- 파라미터 스키마 정의 (JSON Schema)
- 실행 결과 반환
- 외부 시스템 변경 가능 (쓰기 작업)
사용 시나리오:
- 데이터베이스 레코드 생성/수정
- 외부 API 호출 (메일 발송, Slack 메시지 등)
- 시스템 명령 실행
첫 번째 MCP 서버 만들기 - Python 예제
이제 실전 코드로 간단한 MCP 서버를 구축해볼게요. Python의 mcp 라이브러리를 사용하면 정말 쉽습니다.
환경 설정
# MCP SDK 설치
pip install mcp
# 개발 의존성 (선택)
pip install python-dotenv기본 MCP 서버 구조
# server.py
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Resource, Tool, TextContent
# 서버 인스턴스 생성
app = Server("my-first-mcp-server")
# Resource 제공 예제
@app.list_resources()
async def list_resources() -> list[Resource]:
"""
사용 가능한 리소스 목록 반환
AI가 "어떤 데이터를 읽을 수 있나요?"라고 물을 때 호출됨
"""
return [
Resource(
uri="memo://daily",
name="Daily Memo",
mimeType="text/plain",
description="오늘의 메모"
)
]
@app.read_resource()
async def read_resource(uri: str) -> str:
"""
특정 리소스의 내용 반환
AI가 실제로 데이터를 읽을 때 호출됨
"""
if uri == "memo://daily":
# 실제로는 파일이나 DB에서 읽어올 것
return "오늘 할 일: MCP 서버 구축 완료하기!"
raise ValueError(f"Unknown resource: {uri}")
# Tool 제공 예제
@app.list_tools()
async def list_tools() -> list[Tool]:
"""
사용 가능한 도구 목록 반환
AI가 "어떤 작업을 실행할 수 있나요?"라고 물을 때 호출됨
"""
return [
Tool(
name="add_memo",
description="새로운 메모 추가",
inputSchema={
"type": "object",
"properties": {
"content": {
"type": "string",
"description": "메모 내용"
}
},
"required": ["content"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""
도구 실행 및 결과 반환
AI가 실제로 도구를 사용할 때 호출됨
"""
if name == "add_memo":
content = arguments["content"]
# 실제로는 DB에 저장하거나 파일에 기록
return [
TextContent(
type="text",
text=f"메모가 저장되었습니다: {content}"
)
]
raise ValueError(f"Unknown tool: {name}")
# 서버 실행
async def main():
async with stdio_server() as (read_stream, write_stream):
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())왜 이렇게 작성하는가:
@app.list_resources()와@app.read_resource()를 분리하는 이유는 성능 최적화 때문이에요. AI가 먼저 목록을 보고 필요한 것만 읽을 수 있게 하는 거죠.inputSchema는 JSON Schema 표준을 따르는데, 이렇게 하면 AI가 정확한 파라미터로 함수를 호출할 수 있어요.async/await를 사용하는 이유는 I/O 바운드 작업(DB 쿼리, API 호출)이 많기 때문입니다.
실전 예제: 데이터베이스 연동 MCP 서버
실무에서 가장 많이 쓰이는 패턴인 데이터베이스 연동 MCP 서버를 만들어볼게요. SQLite를 예제로 사용하지만 PostgreSQL, MySQL 등으로 쉽게 확장 가능합니다.
# db_mcp_server.py
import asyncio
import sqlite3
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Resource, Tool, TextContent
import json
app = Server("sqlite-mcp-server")
# 데이터베이스 연결 (실제로는 연결 풀 사용 권장)
DB_PATH = "example.db"
def init_db():
"""데이터베이스 초기화 - 실제로는 마이그레이션 도구 사용"""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
conn.close()
init_db()
@app.list_resources()
async def list_resources() -> list[Resource]:
"""
데이터베이스 테이블을 리소스로 노출
각 테이블을 별도 리소스로 관리하면 권한 제어가 쉬워짐
"""
return [
Resource(
uri="sqlite://users",
name="Users Table",
mimeType="application/json",
description="사용자 정보 테이블 (읽기 전용)"
)
]
@app.read_resource()
async def read_resource(uri: str) -> str:
"""
테이블 데이터를 JSON으로 반환
주의: 대용량 테이블은 페이지네이션 필수!
"""
if uri == "sqlite://users":
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row # dict 형태로 결과 반환
cursor = conn.cursor()
# 실전에서는 LIMIT/OFFSET 파라미터 받아서 페이지네이션
cursor.execute("SELECT * FROM users LIMIT 100")
rows = [dict(row) for row in cursor.fetchall()]
conn.close()
return json.dumps(rows, ensure_ascii=False, indent=2)
raise ValueError(f"Unknown resource: {uri}")
@app.list_tools()
async def list_tools() -> list[Tool]:
"""
CRUD 작업을 도구로 제공
읽기는 Resource, 쓰기는 Tool로 분리하는 게 MCP 베스트 프랙티스
"""
return [
Tool(
name="create_user",
description="새 사용자 생성",
inputSchema={
"type": "object",
"properties": {
"name": {"type": "string", "description": "사용자 이름"},
"email": {"type": "string", "description": "이메일 주소"}
},
"required": ["name", "email"]
}
),
Tool(
name="search_users",
description="사용자 검색 (이름 또는 이메일)",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "검색어"}
},
"required": ["query"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""도구 실행 핸들러"""
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
try:
if name == "create_user":
# SQL 인젝션 방지를 위해 파라미터화된 쿼리 사용 필수!
cursor.execute(
"INSERT INTO users (name, email) VALUES (?, ?)",
(arguments["name"], arguments["email"])
)
conn.commit()
user_id = cursor.lastrowid
return [TextContent(
type="text",
text=f"사용자 생성 완료 (ID: {user_id})"
)]
elif name == "search_users":
query = arguments["query"]
# LIKE 검색 - 실제로는 Full-Text Search 사용 권장
cursor.execute(
"""
SELECT * FROM users
WHERE name LIKE ? OR email LIKE ?
LIMIT 20
""",
(f"%{query}%", f"%{query}%")
)
results = [dict(row) for row in cursor.fetchall()]
return [TextContent(
type="text",
text=json.dumps(results, ensure_ascii=False, indent=2)
)]
raise ValueError(f"Unknown tool: {name}")
finally:
conn.close()
async def main():
async with stdio_server() as (read_stream, write_stream):
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())실전 팁:
- 연결 풀 사용: 프로덕션에서는
aiosqlite나asyncpg같은 비동기 드라이버와 연결 풀 필수 - 페이지네이션: 대용량 테이블은 LIMIT/OFFSET이나 커서 기반 페이징 적용
- 트랜잭션 관리: 여러 쿼리를 하나의 트랜잭션으로 묶어야 할 때는 컨텍스트 매니저 활용
Claude Desktop과 MCP 서버 연동하기
이제 만든 서버를 실제로 사용해볼 차례예요. Claude Desktop 설정 파일을 수정하면 됩니다.
macOS 설정
# Claude Desktop 설정 파일 경로
~/Library/Application\ Support/Claude/claude_desktop_config.jsonWindows 설정
%APPDATA%\Claude\claude_desktop_config.json설정 파일 작성
{
"mcpServers": {
"sqlite-db": {
"command": "python",
"args": [
"/absolute/path/to/db_mcp_server.py"
],
"env": {
"DB_PATH": "/path/to/example.db"
}
},
"my-first-server": {
"command": "python",
"args": [
"/absolute/path/to/server.py"
]
}
}
}중요 포인트:
command는 파이썬 인터프리터 경로 (가상환경 사용 시venv/bin/python같은 절대 경로)args는 반드시 절대 경로 사용 (상대 경로는 작동 안 함)env로 환경변수 주입 가능 (API 키, DB 경로 등)
설정 후 Claude Desktop을 재시작하면 채팅 인터페이스에서 MCP 서버의 도구를 사용할 수 있어요.
사용자: 이름이 "김철수"인 사용자를 찾아줘
Claude: [search_users 도구를 자동으로 호출하여 결과 반환]흔한 실수와 해결 방법
실수 1: 절대 경로 미사용
잘못된 예시:
{
"command": "python",
"args": ["./server.py"] // ❌ 상대 경로
}올바른 예시:
{
"command": "/Users/myname/venv/bin/python",
"args": ["/Users/myname/project/server.py"] // ✅ 절대 경로
}이유: Claude Desktop의 작업 디렉토리가 예측 불가능하기 때문에 항상 절대 경로를 써야 해요.
실수 2: 에러 처리 부재
잘못된 예시:
@app.call_tool()
async def call_tool(name: str, arguments: dict):
# 에러 처리 없음
result = some_dangerous_operation(arguments)
return [TextContent(type="text", text=result)]올바른 예시:
@app.call_tool()
async def call_tool(name: str, arguments: dict):
try:
# 입력 검증
if not arguments.get("required_field"):
raise ValueError("필수 필드 누락: required_field")
result = some_dangerous_operation(arguments)
return [TextContent(type="text", text=result)]
except ValueError as e:
# 사용자 에러 - AI가 이해할 수 있게 명확한 메시지
return [TextContent(
type="text",
text=f"입력 오류: {str(e)}"
)]
except Exception as e:
# 시스템 에러 - 로깅하고 일반적인 메시지 반환
logger.error(f"Tool execution failed: {e}")
return [TextContent(
type="text",
text="작업 실행 중 오류가 발생했습니다. 관리자에게 문의하세요."
)]이유: AI에게 반환되는 에러 메시지는 다음 액션을 결정하는 중요한 정보예요. 명확하고 실행 가능한 메시지를 제공해야 합니다.
실수 3: 보안 검증 생략
잘못된 예시:
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "execute_sql":
# ❌ 직접 SQL 실행 - SQL 인젝션 취약점!
query = arguments["query"]
cursor.execute(query)올바른 예시:
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "execute_sql":
# ✅ 화이트리스트 방식 + 파라미터화
allowed_operations = ["SELECT"]
query = arguments["query"].strip().upper()
if not any(query.startswith(op) for op in allowed_operations):
raise ValueError("허용되지 않은 SQL 작업입니다")
# 파라미터화된 쿼리 사용
cursor.execute(
"SELECT * FROM users WHERE id = ?",
(arguments["user_id"],)
)이유: AI가 생성한 입력이라도 절대 신뢰하면 안 돼요. 항상 서버 측에서 검증하고 제한해야 합니다.
성능 최적화와 모범 사례
1. 리소스 캐싱
자주 읽히는 리소스는 캐싱으로 성능을 크게 개선할 수 있어요.
from functools import lru_cache
from datetime import datetime, timedelta
# 간단한 TTL 캐시
class CacheWithTTL:
def __init__(self, ttl_seconds=60):
self.cache = {}
self.ttl = timedelta(seconds=ttl_seconds)
def get(self, key):
if key in self.cache:
value, timestamp = self.cache[key]
if datetime.now() - timestamp < self.ttl:
return value
del self.cache[key]
return None
def set(self, key, value):
self.cache[key] = (value, datetime.now())
resource_cache = CacheWithTTL(ttl_seconds=300) # 5분 캐시
@app.read_resource()
async def read_resource(uri: str) -> str:
# 캐시 확인
cached = resource_cache.get(uri)
if cached:
return cached
# 실제 데이터 로드
if uri == "sqlite://users":
# ... DB 쿼리
result = json.dumps(rows)
resource_cache.set(uri, result)
return result2. 스트리밍 응답 (대용량 데이터)
대용량 데이터는 한 번에 로드하지 말고 스트리밍하세요.
@app.read_resource()
async def read_resource(uri: str) -> str:
if uri.startswith("sqlite://large_table"):
# 청크 단위로 데이터 반환
def generate_chunks():
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
offset = 0
chunk_size = 1000
while True:
cursor.execute(
"SELECT * FROM large_table LIMIT ? OFFSET ?",
(chunk_size, offset)
)
rows = cursor.fetchall()
if not rows:
break
yield json.dumps([dict(row) for row in rows])
offset += chunk_size
conn.close()
# 첫 번째 청크만 반환하고 나머지는 페이지네이션 안내
return next(generate_chunks()) + "\n\n[총 데이터가 많아 일부만 표시. 추가 데이터는 search_large_table 도구 사용]"3. 로깅과 모니터링
프로덕션에서는 반드시 로깅을 추가하세요.
import logging
from datetime import datetime
# 로깅 설정
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('mcp_server.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
@app.call_tool()
async def call_tool(name: str, arguments: dict):
start_time = datetime.now()
try:
logger.info(f"Tool called: {name}, args: {arguments}")
# ... 도구 실행
result = execute_tool(name, arguments)
duration = (datetime.now() - start_time).total_seconds()
logger.info(f"Tool completed: {name}, duration: {duration}s")
return result
except Exception as e:
logger.error(f"Tool failed: {name}, error: {e}", exc_info=True)
raiseTypeScript로 MCP 서버 만들기
Python뿐만 아니라 TypeScript로도 MCP 서버를 만들 수 있어요. Node.js 환경에서 더 익숙하다면 이 방법을 추천합니다.
// server.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
ListResourcesRequestSchema,
ReadResourceRequestSchema,
ListToolsRequestSchema,
CallToolRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
// 서버 인스턴스 생성
const server = new Server(
{
name: "typescript-mcp-server",
version: "1.0.0",
},
{
capabilities: {
resources: {},
tools: {},
},
}
);
// Resources 핸들러
server.setRequestHandler(ListResourcesRequestSchema, async () => {
return {
resources: [
{
uri: "config://app",
name: "App Configuration",
mimeType: "application/json",
description: "애플리케이션 설정 파일",
},
],
};
});
server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
const { uri } = request.params;
if (uri === "config://app") {
const config = {
appName: "My App",
version: "1.0.0",
features: ["auth", "api", "storage"],
};
return {
contents: [
{
uri,
mimeType: "application/json",
text: JSON.stringify(config, null, 2),
},
],
};
}
throw new Error(`Unknown resource: ${uri}`);
});
// Tools 핸들러
server.setRequestHandler(ListToolsRequestSchema, async () => {
return {
tools: [
{
name: "send_notification",
description: "사용자에게 알림 전송",
inputSchema: {
type: "object",
properties: {
userId: {
type: "string",
description: "사용자 ID",
},
message: {
type: "string",
description: "알림 메시지",
},
},
required: ["userId", "message"],
},
},
],
};
});
server.setRequestHandler(CallToolRequestSchema, async (request) => {
const { name, arguments: args } = request.params;
if (name === "send_notification") {
const { userId, message } = args as {
userId: string;
message: string;
};
// 실제로는 푸시 알림 서비스 호출
console.log(`[알림] ${userId}에게: ${message}`);
return {
content: [
{
type: "text",
text: `알림이 ${userId}에게 전송되었습니다.`,
},
],
};
}
throw new Error(`Unknown tool: ${name}`);
});
// 서버 시작
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("TypeScript MCP Server running on stdio");
}
main().catch((error) => {
console.error("Server error:", error);
process.exit(1);
});TypeScript 버전 실행:
# 의존성 설치
npm install @modelcontextprotocol/sdk
# TypeScript 컴파일 및 실행
npx tsx server.ts
# 또는 빌드 후 실행
tsc server.ts
node server.jsPython vs TypeScript 선택 기준:
- Python: 데이터 분석, ML 통합, 기존 Python 스크립트 활용 시
- TypeScript: Node.js 생태계 활용, 웹 API 통합, 타입 안정성 중시 시
보안 고려사항과 권한 관리
MCP 서버는 강력한 만큼 보안에 각별히 신경 써야 해요.
1. 최소 권한 원칙
각 도구는 필요한 최소한의 권한만 가져야 합니다.
# 권한 매핑 (실제로는 DB나 설정 파일에서 로드)
TOOL_PERMISSIONS = {
"read_users": {"resource": "users", "action": "read"},
"create_user": {"resource": "users", "action": "write"},
"delete_user": {"resource": "users", "action": "delete"}, # 위험!
}
def check_permission(tool_name: str, required_action: str) -> bool:
"""도구 실행 전 권한 검사"""
perm = TOOL_PERMISSIONS.get(tool_name, {})
return perm.get("action") == required_action
@app.call_tool()
async def call_tool(name: str, arguments: dict):
# 삭제 작업은 추가 확인 필요
if name == "delete_user":
if not check_permission(name, "delete"):
raise PermissionError("삭제 권한이 없습니다")
# 추가 검증: 관리자만 삭제 가능
if not arguments.get("admin_token"):
raise PermissionError("관리자 인증 필요")2. 입력 검증과 새니타이제이션
모든 사용자 입력은 검증하고 정제해야 합니다.
import re
from typing import Any
def validate_email(email: str) -> bool:
"""이메일 형식 검증"""
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return bool(re.match(pattern, email))
def sanitize_sql_identifier(identifier: str) -> str:
"""SQL 식별자 새니타이제이션 (테이블명, 컬럼명)"""
# 알파벳, 숫자, 언더스코어만 허용
if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', identifier):
raise ValueError(f"Invalid SQL identifier: {identifier}")
return identifier
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "create_user":
email = arguments.get("email", "")
# 검증 실패 시 명확한 에러 메시지
if not validate_email(email):
return [TextContent(
type="text",
text="유효하지 않은 이메일 형식입니다. 예: user@example.com"
)]3. 레이트 리미팅
DoS 공격 방지를 위한 속도 제한 구현:
from collections import defaultdict
from datetime import datetime, timedelta
class RateLimiter:
def __init__(self, max_calls: int, time_window: int):
self.max_calls = max_calls
self.time_window = timedelta(seconds=time_window)
self.calls = defaultdict(list)
def is_allowed(self, key: str) -> bool:
now = datetime.now()
# 오래된 호출 기록 정리
self.calls[key] = [
call_time for call_time in self.calls[key]
if now - call_time < self.time_window
]
if len(self.calls[key]) >= self.max_calls:
return False
self.calls[key].append(now)
return True
# 1분에 최대 10회 호출
rate_limiter = RateLimiter(max_calls=10, time_window=60)
@app.call_tool()
async def call_tool(name: str, arguments: dict):
# 도구별 레이트 리미팅
if not rate_limiter.is_allowed(f"tool:{name}"):
return [TextContent(
type="text",
text="요청 제한에 도달했습니다. 잠시 후 다시 시도하세요."
)]결론 - MCP로 AI 통합의 미래를 준비하세요
MCP(Model Context Protocol)는 AI 에이전트와 외부 시스템을 연결하는 표준화된 방법을 제공하여, 복잡한 통합 작업을 크게 단순화합니다. 핵심은 표준화와 보안이에요.
핵심 요약
- MCP는 클라이언트-서버 아키텍처로 AI와 데이터 소스를 격리하여 안전하게 연결
- Resources, Prompts, Tools 세 가지 기본 개념을 이해하면 대부분의 유즈케이스 커버 가능
- Python과 TypeScript 모두 공식 SDK 제공으로 쉬운 구현
- 보안, 에러 처리, 성능 최적화는 프로덕션 배포의 필수 요소
실무 적용 팁
- 작게 시작하세요: 먼저 읽기 전용 Resource로 시작해서 도구 기능은 점진적으로 추가
- 로컬 개발 먼저: stdio 방식으로 로컬에서 충분히 테스트 후 프로덕션 배포
- 로깅과 모니터링: 초기부터 체계적인 로깅 구축해야 디버깅과 최적화가 쉬워짐
- 커뮤니티 활용: Anthropic의 공식 Discord와 GitHub에서 레퍼런스 서버 참고
다음 단계로 공부하면 좋을 주제
- MCP와 LangChain 통합 - MCP를 LangChain 에이전트와 결합하여 더 복잡한 워크플로우 구축
- MCP 서버 배포 전략 - Docker 컨테이너화, Kubernetes 오케스트레이션, 헬스체크 구현
- 멀티모달 MCP - 이미지, 오디오, 비디오 등 다양한 데이터 타입 처리
MCP는 아직 초기 단계지만 빠르게 성장하고 있는 생태계예요. 지금 시작하면 AI 에이전트 통합 분야에서 선두주자가 될 수 있습니다. 직접 서버를 만들어보고 실무에 적용해보세요!