부록 B. Python SDK 빠른 참조

uv add "mcp[cli]"          # 권장 ([cli]가 mcp dev/run/install 포함)
# 또는
pip install "mcp[cli]"
# 버전 상한 권장: mcp>=1.27,<2

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("Demo", host="127.0.0.1", port=8000,
              stateless_http=True, json_response=True)  # HTTP 옵션 (08장)

# Tool — 타입힌트→inputSchema, 독스트링→description, 반환타입→outputSchema
@mcp.tool()
def add(a: int, b: int) -> int:
    """두 정수를 더한다"""
    return a + b

# Resource — str→text, bytes→blob, {x}→인자
@mcp.resource("greeting://{name}")
def greet(name: str) -> str:
    """인사말"""
    return f"안녕, {name}"

# Prompt — 인자→arguments, 반환→messages
@mcp.prompt()
def review(code: str) -> str:
    """코드 리뷰 프롬프트"""
    return f"다음 코드를 리뷰해줘:\n{code}"

if __name__ == "__main__":
    mcp.run()                              # 기본 stdio
    # mcp.run(transport="streamable-http") # 원격

from typing import Annotated, Literal
from pydantic import Field, BaseModel

@mcp.tool()
def search(
    query: Annotated[str, Field(description="키워드", min_length=1)],
    state: Literal["open", "closed"] = "open",
    limit: Annotated[int, Field(ge=1, le=100)] = 20,
) -> str:
    """이슈 검색"""
    ...

class Req(BaseModel):
    city: str = Field(description="도시")
    nights: int = Field(ge=1, le=30)

@mcp.tool()
async def work(uri: str, ctx: Context) -> str:
    await ctx.info("시작")               # 로깅 (stdout print 금지!)
    await ctx.report_progress(0, 2)      # 진행률
    data = await ctx.read_resource(uri)  # 서버 자기 리소스 읽기
    res = await ctx.session.create_message(  # sampling (클라가 광고 시)
        messages=[{"role": "user",
                   "content": {"type": "text", "text": f"요약:{data}"}}],
        max_tokens=200)
    return res.content.text

import mcp.types as types

@mcp.tool()
def risky(x: int) -> str:
    """..."""
    if x < 0:
        raise ValueError("음수 불가")     # → 프로토콜 에러 (JSON-RPC error)
    try:
        return do(x)
    except BusinessError as e:
        return types.CallToolResult(       # → 실행 에러 (모델이 읽고 교정)
            content=[types.TextContent(type="text", text=str(e))],
            isError=True)

python server.py                 # 직접 실행 (stdio)
mcp run server.py                # CLI 실행 (mcp/app/server 객체 탐색)
mcp run server.py:my_server      # 커스텀 객체명
mcp dev server.py                # Inspector 포함 (localhost:6274)
mcp dev server.py --with pandas  # 의존성 주입
mcp install server.py --name "X" -e KEY=val   # 호스트 등록 + 환경변수

# stdio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

params = StdioServerParameters(command="python", args=["server.py"], env=None)
async with stdio_client(params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        tools = (await session.list_tools()).tools
        r = await session.call_tool("add", {"a": 3, "b": 4})
        print(r.content[0].text, r.isError)
        await session.read_resource("greeting://Alice")
        await session.get_prompt("review", {"code": "..."})

# HTTP — read, write, get_session_id (3개)
from mcp.client.streamable_http import streamablehttp_client
async with streamablehttp_client("http://host:8000/mcp", headers={"Authorization": f"Bearer {tok}"}) as (read, write, get_session_id):
    async with ClientSession(read, write) as session:
        await session.initialize()
        ...

# sampling 콜백 등록 (서버 역방향 요청 처리)
async with ClientSession(read, write, sampling_callback=handle_sampling) as session:
    ...

import pytest
from mcp.shared.memory import create_connected_server_and_client_session as connect
from server import mcp

@pytest.mark.anyio
async def test_add():
    async with connect(mcp._mcp_server) as client:
        r = await client.call_tool("add", {"a": 2, "b": 3})
        assert r.content[0].text == "5"