15. 도구/리소스/프롬프트 등록과 입력 스키마·검증

14장에서 최소 서버를 띄웠습니다. 이제 각 primitive를 제대로 정의하는 법을 봅니다 — 스키마 정밀화, 검증, 구조화 출력, 그리고 Context.

입력 스키마 — 타입 힌트에서 Pydantic까지

FastMCP는 함수 시그니처의 타입 힌트로 inputSchema를 자동 생성합니다(14장). 기본 타입은 그대로 매핑됩니다.

@mcp.tool()
def add(a: int, b: int) -> int:
    """두 정수를 더한다"""
    return a + b
# inputSchema: { type: object, properties: {a:{type:integer}, b:{type:integer}}, required:[a,b] }

더 정밀한 제약(범위·설명·기본값·열거)이 필요하면 pydantic.Field나 Annotated를 씁니다. Pydantic은 파이썬에서 가장 널리 쓰이는 데이터 검증 라이브러리로, 타입 힌트를 근거로 "들어온 값이 규칙에 맞는지" 자동으로 확인하고 안 맞으면 명확한 에러를 냅니다. FastMCP가 입력 스키마를 만들고 검증하는 데 내부적으로 Pydantic을 씁니다. Annotated는 파이썬 표준 기능으로, 타입에 부가 메타데이터(여기서는 Field(...) 제약)를 덧붙이는 문법입니다.

from typing import Annotated, Literal
from pydantic import Field

@mcp.tool()
def search_issues(
    query: Annotated[str, Field(description="검색 키워드", min_length=1, max_length=200)],
    state: Literal["open", "closed", "all"] = "open",
    limit: Annotated[int, Field(ge=1, le=100)] = 20,
) -> str:
    """GitHub 이슈를 키워드로 검색해 일치하는 결과를 반환한다."""
    ...

• Field(description=...)는 스키마에 필드 설명을 넣어 모델이 인자를 더 정확히 만들게 합니다.
• ge/le/min_length 같은 제약은 스키마에 반영되고, FastMCP가 호출 시 검증합니다. 범위를 벗어난 인자는 도구 본문에 닿기 전에 걸러집니다.
• Literal[...]은 열거형(enum)으로 변환됩니다.

복잡한 입력은 Pydantic 모델로 묶을 수 있습니다.

from pydantic import BaseModel, Field

class BookingRequest(BaseModel):
    city: str = Field(description="도시명")
    nights: int = Field(ge=1, le=30, description="숙박 일수")
    breakfast: bool = False

@mcp.tool()
def book_hotel(req: BookingRequest) -> str:
    """호텔을 예약한다."""
    return f"{req.city} {req.nights}박 예약 완료 (조식: {req.breakfast})"

🔒 검증은 1차 방어선이지 전부가 아니다: 스키마 검증은 "타입·범위"를 막아 주지만, 의미적 안전(경로 탈출, SQL/명령 주입)은 막지 못합니다. city가 문자열인 건 검증되지만 그 문자열이 '; DROP TABLE인지는 검증되지 않습니다. 도구 본문에서 파일 경로·SQL·셸에 인자를 쓸 때는 추가 정화(sanitization)가 필요합니다(21장). OWASP의 MCP 서버 보안 가이드도 "엄격한 스키마 검증 + 셸/SQL/파일에 사용자 입력을 직접 넘기지 말 것"을 핵심 수칙으로 듭니다.

리소스 등록 — 정적·동적·템플릿

# 정적 리소스 — 고정 데이터
@mcp.resource("config://app")
def get_config() -> str:
    """앱 설정을 반환"""
    return "theme=dark; lang=ko"

# 동적 리소스 (URI 템플릿) — {city}가 함수 인자로 매핑
@mcp.resource("weather://{city}/current")
def current_weather(city: str) -> str:
    """도시의 현재 날씨"""
    return f"{city}: 23°C 맑음"

# 이진(blob) 리소스 — bytes 반환 + mime_type 지정
@mcp.resource("images://logo.png", mime_type="image/png")
def get_logo() -> bytes:
    """로고 이미지(base64로 인코딩되어 전달됨)"""
    from pathlib import Path
    return Path("logo.png").read_bytes()

• 반환 타입이 str이면 텍스트, bytes면 base64 blob으로 직렬화됩니다(11장).
• URI의 {city} 자리표시자는 함수의 동명 인자와 자동 매핑됩니다 — 이것이 URI 템플릿의 Python 표현입니다.

🔒 경로 탈출 차단: @mcp.resource("files://{path}")처럼 경로를 받으면, 본문에서 반드시 정규화·경계 검사를 하세요. Path(base, path).resolve()가 base 밖으로 나가면 거부해야 ../../etc/passwd 류를 막습니다.

from pathlib import Path

BASE = Path("/srv/data").resolve()

@mcp.resource("files://{name}")
def read_file(name: str) -> str:
    """data 디렉터리 안의 파일만 읽는다"""
    target = (BASE / name).resolve()
    if not str(target).startswith(str(BASE)):   # 경계 검사
        raise ValueError("허용되지 않은 경로")
    return target.read_text(encoding="utf-8")

구조화된 출력 — outputSchema/structuredContent 안전하게

10장에서 본 규칙을 코드로 옮깁니다. FastMCP는 반환 타입 힌트로 outputSchema를 자동 생성하고 structuredContent를 채워 줍니다.

from pydantic import BaseModel

class WeatherData(BaseModel):
    temperature: float
    conditions: str

@mcp.tool()
def get_weather(location: str) -> WeatherData:
    """위치의 현재 날씨 데이터를 구조화해 반환한다."""
    return WeatherData(temperature=23.0, conditions="구름 조금")
    # FastMCP가 outputSchema 생성 + structuredContent 채움 +
    # 하위 호환용 JSON 텍스트를 content에도 넣어 줌

dict를 직접 반환할 수도 있지만, 스키마 일치를 스스로 보장해야 합니다.

@mcp.tool()
def stats() -> dict:
    """간단한 통계를 반환"""
    return {"count": 3, "avg": 1.5}   # 스키마를 명시하지 않으면 느슨하게 처리됨

Context — 도구 안에서 프로토콜 기능 쓰기

도구 본문에서 로깅·진행 보고·리소스 읽기·sampling·elicitation에 접근하려면, 함수에 Context 인자를 추가합니다. FastMCP가 주입해 줍니다.

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("Demo")

@mcp.tool()
async def summarize(uri: str, ctx: Context) -> str:
    """리소스를 읽어 LLM으로 요약한다."""
    await ctx.info(f"요약 시작: {uri}")          # 로깅 (notifications/message)
    await ctx.report_progress(0, 2)              # 진행률 (13장)

    data = await ctx.read_resource(uri)          # 서버가 자기 리소스를 읽음
    await ctx.report_progress(1, 2)

    # sampling — 호스트 LLM에 요약 요청 (클라이언트가 sampling 광고 시)
    result = await ctx.session.create_message(
        messages=[{"role": "user",
                   "content": {"type": "text", "text": f"다음을 요약:\n{data}"}}],
        max_tokens=200,
    )
    await ctx.report_progress(2, 2)
    return result.content.text

• ctx.info/debug/warning/error(...) → 프로토콜 로깅. stdout print 대신 이걸 쓰세요(07장).
• ctx.report_progress(...) → 긴 작업 진행 알림(13장).
• ctx.read_resource(...) → 서버가 자신의 리소스에 접근.
• ctx.session.create_message(...) → sampling. ctx.elicit(...) → elicitation. 클라이언트가 해당 capability를 광고했을 때만 동작하므로, 지원 여부 확인 후 사용하거나 미지원을 우아하게 처리하세요(13장).

등록 패턴 정리

flowchart TD
    F["파이썬 함수"] --> D{"무슨 데코레이터?"}
    D -->|"@mcp.tool"| T["Tool\n타입힌트→inputSchema\n반환타입→outputSchema\n독스트링→description"]
    D -->|"@mcp.resource(uri)"| R["Resource\nuri {x}→인자\nstr→text / bytes→blob"]
    D -->|"@mcp.prompt"| P["Prompt\n인자→arguments\n반환→messages"]
    T -.->|"Context 인자 추가 시"| C["로깅·진행·read_resource·sampling·elicit"]
    classDef f fill:#bfdbfe,stroke:#1d4ed8,color:#000;
    classDef p fill:#5eead4,stroke:#0f766e,color:#000;
    classDef c fill:#fde68a,stroke:#b45309,color:#000;
    class F,D f;
    class T,R,P p;
    class C c;

이 장에서 배운 것

• 입력 스키마는 타입 힌트 + pydantic.Field/Annotated/BaseModel 로 정밀화하고, FastMCP가 호출 시 검증한다.
• 스키마 검증은 1차 방어선 — 경로 탈출·주입은 본문에서 추가로 정화해야 한다.
• 리소스는 str(text)/bytes(blob), URI {x}는 인자 매핑. 구조화 출력은 outputSchema와 structuredContent를 일치시킨다(불일치·None 금지).
• Context 인자로 로깅·진행·read_resource·sampling·elicitation에 접근한다. print 대신 ctx.info.

✍️ 확인 문제

1. 도구 인자 limit을 1~100으로 제한하고 싶다. 어떻게 선언하나? 범위를 벗어나면 어디서 걸러지나?
2. outputSchema가 자동 생성됐는데 본문에서 스키마와 다른 dict를 반환하면 무슨 일이 생기나?
3. 도구 본문에서 진행률을 보고하고 호스트 LLM에 요약을 시키려면 무엇을 함수에 추가하고, 무엇을 호출하나?

다음 장: 16. 에러 처리·로깅·테스트와 Inspector 디버깅