Server 구현 (Python)

pip install "mcp[cli]" httpx

# 디렉터리 구조
my_mcp_server/
├── server.py        # 서버 진입점
├── tools/
│   ├── search.py    # 검색 툴
│   └── database.py  # DB 툴
└── resources/
    └── config.py    # 설정 리소스

from mcp.server.fastmcp import FastMCP
import mcp.types as types
import httpx, json, sqlite3
from pathlib import Path

# ── 서버 인스턴스 ──────────────────────────────
mcp = FastMCP(
    name="my-server",
    version="1.0.0",
    instructions="웹 검색과 DB 조회를 제공하는 MCP 서버입니다."
)

# ── Tool 등록 ──────────────────────────────────
@mcp.tool()
async def web_search(
    query: str,
    max_results: int = 5
) -> str:
    """웹에서 정보를 검색합니다.

    Args:
        query: 검색 쿼리 문자열
        max_results: 반환할 최대 결과 수 (기본값 5, 최대 20)
    """
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://api.search.example.com/search",
            params={"q": query, "limit": max_results}
        )
        data = resp.json()
    results = [f"{r['title']}: {r['url']}" for r in data["results"]]
    return "\n".join(results)


@mcp.tool()
def query_db(sql: str, db_path: str = "app.db") -> list[dict]:
    """SQLite DB를 쿼리합니다. SELECT 문만 허용됩니다."""
    if not sql.strip().upper().startswith("SELECT"):
        raise ValueError("SELECT 문만 허용됩니다")
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    rows = conn.execute(sql).fetchall()
    conn.close()
    return [dict(row) for row in rows]


# ── Resource 등록 ──────────────────────────────
@mcp.resource("config://app/settings")
def get_settings() -> str:
    """애플리케이션 설정 JSON을 반환합니다."""
    settings = {"debug": False, "version": "1.0.0", "max_tokens": 4096}
    return json.dumps(settings, ensure_ascii=False, indent=2)


# URI 템플릿 — {path} 변수 자동 매핑
@mcp.resource("file://{path}")
def read_file(path: str) -> str:
    """로컬 파일 내용을 읽어 반환합니다."""
    target = Path(path).resolve()
    # 경로 순회 공격 방어
    if not str(target).startswith("/allowed/base/path"):
        raise PermissionError("허용되지 않은 경로입니다")
    return target.read_text(encoding="utf-8")


# ── Prompt 등록 ──────────────────────────────
@mcp.prompt()
def code_review(
    code: str,
    language: str = "python"
) -> list[types.Message]:
    """코드 리뷰 요청 프롬프트 템플릿."""
    return [
        types.UserMessage(
            content=f"다음 {language} 코드를 리뷰해주세요:\n\n```{language}\n{code}\n```\n\n"
                    "보안 취약점, 성능 문제, 가독성 개선 사항을 구체적으로 알려주세요."
        )
    ]


# ── 서버 실행 ──────────────────────────────────
if __name__ == "__main__":
    mcp.run()  # 기본값: stdio

# 1. stdio — 로컬 프로세스 (Claude Desktop, Cursor 등)
mcp.run(transport="stdio")

# 2. HTTP+SSE — 원격 서버 (네트워크 접근 필요 시)
mcp.run(transport="sse", host="0.0.0.0", port=8080)

# 3. CLI로 빠른 테스트 (mcp[cli] 설치 필요)
# $ mcp dev server.py
# → MCP Inspector 웹 UI가 열립니다

from mcp.server.fastmcp import FastMCP
from mcp.types import ToolAnnotations

mcp = FastMCP("annotated-server")

# 읽기 전용 — Host가 자동 승인 가능
@mcp.tool(annotations=ToolAnnotations(
    readOnlyHint=True,
    idempotentHint=True,
))
def get_user(user_id: str) -> dict:
    """사용자 정보를 읽어옵니다 (읽기 전용)."""
    ...

# 파괴적 작업 — Host가 반드시 사용자 확인 요청
@mcp.tool(annotations=ToolAnnotations(
    destructiveHint=True,
    readOnlyHint=False,
))
def delete_record(record_id: str) -> bool:
    """레코드를 영구 삭제합니다 (되돌릴 수 없음)."""
    ...

from mcp import McpError
from mcp.types import ErrorCode

@mcp.tool()
def risky_operation(item_id: str) -> str:
    """위험한 작업 예시."""
    if not item_id:
        # 잘못된 파라미터 → InvalidParams
        raise McpError(ErrorCode.INVALID_PARAMS, "item_id는 필수입니다")

    item = db.get(item_id)
    if item is None:
        # 리소스 없음 → InvalidRequest
        raise McpError(ErrorCode.INVALID_REQUEST, f"{item_id} 를 찾을 수 없습니다")

    return item.process()