MCP 서버 직접 만들기 — Python과 TypeScript로 첫 서버 구축

from mcp.server.fastmcp import FastMCP

# 서버 인스턴스 생성
mcp = FastMCP("my-first-server")

@mcp.tool()
async def add(a: int, b: int) -> str:
    """두 수를 더합니다."""
    return str(a + b)

@mcp.tool()
async def multiply(a: int, b: int) -> str:
    """두 수를 곱합니다.
    
    Args:
        a: 첫 번째 숫자
        b: 두 번째 숫자
    """
    return str(a * b)

if __name__ == "__main__":
    # STDIO 트랜스포트로 서버 실행
    mcp.run(transport="stdio")

@mcp.resource("greeting://{name}")
async def greeting(name: str) -> str:
    """특정 사람에게 인사를 건넵니다."""
    return f"안녕하세요, {name}님! MCP 서버에 오신 것을 환영합니다."

@mcp.resource("file://config")
async def get_config() -> str:
    """서버 설정을 반환합니다."""
    config = {
        "version": "1.0.0",
        "features": ["tools", "resources", "prompts"]
    }
    import json
    return json.dumps(config, indent=2, ensure_ascii=False)

@mcp.prompt()
async def code_review_prompt(language: str, code: str) -> str:
    """코드 리뷰를 위한 프롬프트를 생성합니다.
    
    Args:
        language: 프로그래밍 언어 (예: Python, JavaScript)
        code: 리뷰할 코드
    """
    return f"""다음 {language} 코드를 리뷰해주세요:

```{language}
{code}

# ❌ 절대 하지 말 것 (JSON-RPC 깨짐!)
print("서버가 시작되었습니다")

# ✅ 올바른 방법
import sys
print("서버가 시작되었습니다", file=sys.stderr)

# 또는 로깅 라이브러리 사용
import logging
logging.info("서버가 시작되었습니다")  # 기본적으로 stderr로 출력

from mcp.server.fastmcp import FastMCP
from typing import List
import json
import sys

# 서버 인스턴스
mcp = FastMCP("todo-manager")

# 메모리 기반 할 일 저장소
todos: List[dict] = []

@mcp.tool()
async def add_todo(task: str, priority: str = "medium") -> str:
    """할 일을 추가합니다.
    
    Args:
        task: 할 일 내용
        priority: 우선순위 (low, medium, high)
    """
    todo = {
        "id": len(todos) + 1,
        "task": task,
        "priority": priority,
        "completed": False
    }
    todos.append(todo)
    print(f"[DEBUG] 할 일 추가됨: {todo}", file=sys.stderr)
    return f"✅ 할 일이 추가되었습니다 (ID: {todo['id']})"

@mcp.tool()
async def list_todos(filter_priority: str = "all") -> str:
    """할 일 목록을 조회합니다.
    
    Args:
        filter_priority: 우선순위 필터 (all, low, medium, high)
    """
    if filter_priority == "all":
        filtered = todos
    else:
        filtered = [t for t in todos if t["priority"] == filter_priority]
    
    if not filtered:
        return "📋 할 일이 없습니다."
    
    result = "📋 할 일 목록:\n\n"
    for todo in filtered:
        status = "✅" if todo["completed"] else "⬜"
        result += f"{status} [{todo['id']}] {todo['task']} (우선순위: {todo['priority']})\n"
    
    return result

@mcp.tool()
async def complete_todo(todo_id: int) -> str:
    """할 일을 완료 처리합니다.
    
    Args:
        todo_id: 완료할 할 일의 ID
    """
    for todo in todos:
        if todo["id"] == todo_id:
            todo["completed"] = True
            return f"✅ 할 일 #{todo_id}가 완료되었습니다!"
    
    return f"❌ ID {todo_id}인 할 일을 찾을 수 없습니다."

@mcp.tool()
async def delete_todo(todo_id: int) -> str:
    """할 일을 삭제합니다.
    
    Args:
        todo_id: 삭제할 할 일의 ID
    """
    global todos
    original_len = len(todos)
    todos = [t for t in todos if t["id"] != todo_id]
    
    if len(todos) < original_len:
        return f"🗑️ 할 일 #{todo_id}가 삭제되었습니다."
    else:
        return f"❌ ID {todo_id}인 할 일을 찾을 수 없습니다."

@mcp.resource("todos://all")
async def get_all_todos() -> str:
    """모든 할 일을 JSON 형식으로 반환합니다."""
    return json.dumps(todos, indent=2, ensure_ascii=False)

if __name__ == "__main__":
    mcp.run(transport="stdio")

import asyncio
import httpx

@mcp.tool()
async def fetch_multiple_apis(urls: list[str]) -> str:
    """여러 API를 동시에 호출합니다."""
    async with httpx.AsyncClient() as client:
        # 병렬로 요청 실행
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
    
    return f"총 {len(responses)}개 API 호출 완료"

from functools import lru_cache
from datetime import datetime, timedelta

# 간단한 시간 기반 캐시
cache = {}
CACHE_TTL = timedelta(minutes=5)

@mcp.tool()
async def get_weather(city: str) -> str:
    """날씨 정보를 캐싱과 함께 조회합니다."""
    now = datetime.now()
    
    # 캐시 확인
    if city in cache:
        cached_time, cached_data = cache[city]
        if now - cached_time < CACHE_TTL:
            return f"[캐시] {cached_data}"
    
    # 실제 API 호출 (여기서는 더미)
    data = f"{city}의 날씨는 맑음입니다."
    cache[city] = (now, data)
    
    return data

@mcp.tool()
async def read_file(filepath: str) -> str:
    """파일을 읽습니다."""
    # ❌ 위험: 경로 조작 공격 가능
    # with open(filepath) as f:
    #     return f.read()
    
    # ✅ 안전: 허용된 디렉토리만 접근
    import os
    from pathlib import Path
    
    ALLOWED_DIR = Path("/safe/directory")
    full_path = (ALLOWED_DIR / filepath).resolve()
    
    if not full_path.is_relative_to(ALLOWED_DIR):
        return "❌ 허용되지 않은 경로입니다."
    
    with open(full_path) as f:
        return f.read()

# ❌ 절대 하지 말 것
@mcp.tool()
async def get_api_key() -> str:
    return "sk-abc123..."  # API 키 노출!

# ✅ 환경 변수 사용
import os

@mcp.tool()
async def call_external_api(query: str) -> str:
    api_key = os.getenv("EXTERNAL_API_KEY")
    # API 호출 로직...

from collections import defaultdict
from datetime import datetime, timedelta

# 간단한 레이트 리미터
rate_limits = defaultdict(list)
MAX_CALLS_PER_MINUTE = 10

@mcp.tool()
async def expensive_operation(user_id: str) -> str:
    """비용이 큰 작업 (레이트 리밋 적용)"""
    now = datetime.now()
    
    # 1분 이내 호출 기록 확인
    recent_calls = [
        t for t in rate_limits[user_id]
        if now - t < timedelta(minutes=1)
    ]
    
    if len(recent_calls) >= MAX_CALLS_PER_MINUTE:
        return "❌ 요청이 너무 많습니다. 잠시 후 다시 시도하세요."
    
    # 호출 기록
    rate_limits[user_id].append(now)
    
    # 실제 작업...
    return "✅ 작업 완료"