6장: Python FastMCP로 서버 구축하기

pip install fastmcp

from fastmcp import FastMCP
 
mcp = FastMCP("calculator-server")
 
 
@mcp.tool()
def add(a: float, b: float) -> float:
    """두 숫자를 더합니다."""
    return a + b
 
 
@mcp.tool()
def multiply(a: float, b: float) -> float:
    """두 숫자를 곱합니다."""
    return a * b
 
 
if __name__ == "__main__":
    mcp.run()

python server.py

from typing import Optional
from pydantic import BaseModel, Field
 
 
class SearchParams(BaseModel):
    query: str = Field(description="검색어")
    max_results: int = Field(default=10, ge=1, le=100, description="최대 결과 수")
    language: Optional[str] = Field(default="ko", description="결과 언어 코드")
 
 
@mcp.tool()
def search_documents(params: SearchParams) -> str:
    """문서를 검색합니다. 키워드 기반 전문 검색을 지원합니다."""
    results = perform_search(params.query, params.max_results, params.language)
    return format_results(results)

import httpx
 
 
@mcp.tool()
async def fetch_webpage(url: str) -> str:
    """웹 페이지의 내용을 가져옵니다."""
    async with httpx.AsyncClient() as client:
        response = await client.get(url, follow_redirects=True, timeout=30.0)
        response.raise_for_status()
        return response.text

@mcp.tool(name="query_db", description="SQL 쿼리를 실행하여 데이터베이스에서 데이터를 조회합니다")
async def execute_database_query(query: str, database: str = "main") -> str:
    """내부적으로는 긴 이름을 사용하지만, MCP에 노출되는 이름은 query_db입니다."""
    result = await db.execute(query, database)
    return str(result)

@mcp.tool()
async def read_file(path: str) -> str:
    """지정된 경로의 파일을 읽습니다."""
    import os
 
    if not os.path.exists(path):
        raise FileNotFoundError("파일을 찾을 수 없습니다: " + path)
 
    if not os.path.isfile(path):
        raise ValueError("지정된 경로는 파일이 아닙니다: " + path)
 
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

@mcp.resource("config://app")
def get_app_config() -> str:
    """애플리케이션 설정 정보를 제공합니다."""
    import json
 
    config = {
        "app_name": "Weather Service",
        "version": "1.0.0",
        "supported_cities": ["서울", "부산", "제주", "인천", "대구"],
        "api_version": "v2",
        "rate_limit": {"requests_per_minute": 60},
    }
    return json.dumps(config, ensure_ascii=False, indent=2)

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """특정 사용자의 프로필 정보를 제공합니다."""
    import json
 
    profiles = {
        "user001": {"name": "김개발", "role": "backend", "team": "platform"},
        "user002": {"name": "이디자인", "role": "designer", "team": "product"},
    }
    profile = profiles.get(user_id)
    if not profile:
        return json.dumps({"error": "사용자를 찾을 수 없습니다"}, ensure_ascii=False)
 
    return json.dumps(profile, ensure_ascii=False, indent=2)

from fastmcp.prompts import Message
 
 
@mcp.prompt()
def code_review(code: str, language: str = "python") -> list[Message]:
    """코드 리뷰를 수행합니다."""
    return [
        Message(
            role="user",
            content=(
                "다음 " + language + " 코드를 리뷰해 주세요.\n\n"
                "리뷰 관점:\n"
                "1. 버그 가능성\n"
                "2. 보안 취약점\n"
                "3. 성능 개선 여지\n"
                "4. 코드 가독성\n"
                "5. 모범 사례 준수 여부\n\n"
                "코드:\n```" + language + "\n" + code + "\n```"
            ),
        )
    ]
 
 
@mcp.prompt()
def debug_error(error_message: str, stack_trace: str = "") -> list[Message]:
    """에러를 분석하고 해결 방안을 제시합니다."""
    content = "다음 에러를 분석하고 해결 방안을 제시해 주세요.\n\n"
    content += "에러 메시지:\n" + error_message + "\n"
    if stack_trace:
        content += "\n스택 트레이스:\n" + stack_trace + "\n"
    content += (
        "\n분석 형식:\n"
        "1. 원인 분석\n"
        "2. 해결 방안 (우선순위 순)\n"
        "3. 재발 방지 대책"
    )
    return [Message(role="user", content=content)]

import os
import json
from datetime import datetime
from pathlib import Path
from typing import Optional
 
from fastmcp import FastMCP
from fastmcp.prompts import Message
from pydantic import BaseModel, Field
 
mcp = FastMCP("file-manager", description="파일 시스템 관리 MCP 서버")
 
# 작업 디렉토리 설정
WORKSPACE = os.environ.get("WORKSPACE_DIR", os.path.expanduser("~/workspace"))
 
 
# --- 도구 ---
 
@mcp.tool()
def list_files(
    directory: str = ".",
    pattern: str = "*",
    recursive: bool = False,
) -> str:
    """디렉토리의 파일 목록을 반환합니다."""
    base = Path(WORKSPACE) / directory
    if not base.exists():
        raise FileNotFoundError("디렉토리가 존재하지 않습니다: " + str(base))
 
    if recursive:
        files = list(base.rglob(pattern))
    else:
        files = list(base.glob(pattern))
 
    result = []
    for f in sorted(files)[:100]:
        stat = f.stat()
        result.append({
            "name": f.name,
            "path": str(f.relative_to(WORKSPACE)),
            "size": stat.st_size,
            "is_directory": f.is_dir(),
            "modified": datetime.fromtimestamp(stat.st_mtime).isoformat(),
        })
 
    return json.dumps(result, ensure_ascii=False, indent=2)
 
 
@mcp.tool()
def read_file(path: str, encoding: str = "utf-8") -> str:
    """파일의 내용을 읽습니다."""
    file_path = Path(WORKSPACE) / path
    if not file_path.exists():
        raise FileNotFoundError("파일을 찾을 수 없습니다: " + path)
    if not file_path.is_file():
        raise ValueError("경로가 파일이 아닙니다: " + path)
 
    return file_path.read_text(encoding=encoding)
 
 
@mcp.tool()
def write_file(path: str, content: str, encoding: str = "utf-8") -> str:
    """파일에 내용을 씁니다. 파일이 없으면 생성합니다."""
    file_path = Path(WORKSPACE) / path
    file_path.parent.mkdir(parents=True, exist_ok=True)
    file_path.write_text(content, encoding=encoding)
    return "파일이 저장되었습니다: " + path
 
 
@mcp.tool()
def search_in_files(
    query: str,
    directory: str = ".",
    file_pattern: str = "*.py",
) -> str:
    """파일 내용에서 텍스트를 검색합니다."""
    base = Path(WORKSPACE) / directory
    matches = []
 
    for file_path in base.rglob(file_pattern):
        if not file_path.is_file():
            continue
        try:
            content = file_path.read_text(encoding="utf-8")
            for i, line in enumerate(content.splitlines(), 1):
                if query.lower() in line.lower():
                    matches.append({
                        "file": str(file_path.relative_to(WORKSPACE)),
                        "line": i,
                        "content": line.strip(),
                    })
        except (UnicodeDecodeError, PermissionError):
            continue
 
    return json.dumps(matches[:50], ensure_ascii=False, indent=2)
 
 
# --- 리소스 ---
 
@mcp.resource("workspace://structure")
def get_workspace_structure() -> str:
    """작업 디렉토리의 구조를 트리 형태로 제공합니다."""
    def build_tree(path: Path, prefix: str = "", max_depth: int = 3, depth: int = 0) -> str:
        if depth >= max_depth:
            return ""
        lines = []
        entries = sorted(path.iterdir(), key=lambda e: (not e.is_dir(), e.name))
        for i, entry in enumerate(entries):
            if entry.name.startswith("."):
                continue
            is_last = i == len(entries) - 1
            connector = "--- " if is_last else "|-- "
            lines.append(prefix + connector + entry.name)
            if entry.is_dir():
                extension = "    " if is_last else "|   "
                subtree = build_tree(entry, prefix + extension, max_depth, depth + 1)
                if subtree:
                    lines.append(subtree)
        return "\n".join(lines)
 
    tree = build_tree(Path(WORKSPACE))
    return WORKSPACE + "\n" + tree
 
 
@mcp.resource("workspace://stats")
def get_workspace_stats() -> str:
    """작업 디렉토리의 통계 정보를 제공합니다."""
    base = Path(WORKSPACE)
    file_count = 0
    dir_count = 0
    total_size = 0
    extensions: dict[str, int] = {}
 
    for item in base.rglob("*"):
        if item.name.startswith("."):
            continue
        if item.is_file():
            file_count += 1
            total_size += item.stat().st_size
            ext = item.suffix or "(없음)"
            extensions[ext] = extensions.get(ext, 0) + 1
        elif item.is_dir():
            dir_count += 1
 
    stats = {
        "workspace": WORKSPACE,
        "files": file_count,
        "directories": dir_count,
        "total_size_mb": round(total_size / (1024 * 1024), 2),
        "top_extensions": dict(sorted(extensions.items(), key=lambda x: -x[1])[:10]),
    }
    return json.dumps(stats, ensure_ascii=False, indent=2)
 
 
# --- 프롬프트 ---
 
@mcp.prompt()
def analyze_project() -> list[Message]:
    """프로젝트 구조를 분석하고 개선 사항을 제안합니다."""
    return [
        Message(
            role="user",
            content=(
                "현재 작업 디렉토리의 프로젝트를 분석해 주세요.\n\n"
                "1. workspace://structure 리소스에서 디렉토리 구조를 확인하세요.\n"
                "2. workspace://stats 리소스에서 통계 정보를 확인하세요.\n"
                "3. 주요 설정 파일(package.json, pyproject.toml 등)을 read_file로 확인하세요.\n\n"
                "분석 항목:\n"
                "- 프로젝트 유형과 기술 스택\n"
                "- 디렉토리 구조의 적절성\n"
                "- 발견된 문제점이나 개선 사항\n"
                "- 권장 사항"
            ),
        )
    ]
 
 
if __name__ == "__main__":
    mcp.run()

if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

import os
 
if __name__ == "__main__":
    transport = os.environ.get("MCP_TRANSPORT", "stdio")
    if transport == "http":
        mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)
    else:
        mcp.run()

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
 
const server = new McpServer({ name: "calc", version: "1.0.0" });
 
server.tool(
  "add",
  "두 숫자를 더합니다",
  { a: z.number(), b: z.number() },
  async (params) => ({
    content: [{ type: "text", text: String(params.a + params.b) }],
  })
);

from fastmcp import FastMCP
 
mcp = FastMCP("calc")
 
@mcp.tool()
def add(a: float, b: float) -> float:
    """두 숫자를 더합니다."""
    return a + b