MCP 파이썬 SDK
모델 컨텍스트 프로토콜(MCP)의 Python 구현
목차
개요
모델 컨텍스트 프로토콜(MCP)을 사용하면 애플리케이션이 표준화된 방식으로 LLM에 대한 컨텍스트를 제공할 수 있으며, 컨텍스트 제공과 실제 LLM 상호작용 간의 연관성을 분리할 수 있습니다. 이 Python SDK는 전체 MCP 사양을 구현하여 다음과 같은 작업을 쉽게 수행할 수 있도록 합니다.
- 모든 MCP 서버에 연결할 수 있는 MCP 클라이언트를 구축합니다.
- 리소스, 프롬프트 및 도구를 노출하는 MCP 서버를 만듭니다.
- stdio 및 SSE와 같은 표준 전송을 사용하세요
- 모든 MCP 프로토콜 메시지 및 수명 주기 이벤트를 처리합니다.
설치
Python 프로젝트에 MCP 추가하기
Python 프로젝트를 관리하려면 uv를 사용하는 것이 좋습니다.
아직 uv 관리 프로젝트를 만들지 않았다면 하나 만드세요.
지엑스피1
그런 다음 프로젝트 종속성에 MCP를 추가합니다.
또는 종속성을 위해 pip를 사용하는 프로젝트의 경우:
독립형 MCP 개발 도구 실행
uv와 함께 mcp 명령을 실행하려면:
빠른 시작
계산기 도구와 일부 데이터를 공개하는 간단한 MCP 서버를 만들어 보겠습니다.
# server.py
from mcp.server.fastmcp import FastMCP
# Create an MCP server
mcp = FastMCP("Demo")
# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""Get a personalized greeting"""
return f"Hello, {name}!"
Claude Desktop 에 이 서버를 설치하고 다음을 실행하여 바로 상호 작용할 수 있습니다.
또는 MCP Inspector로 테스트할 수 있습니다.
MCP란 무엇인가요?
모델 컨텍스트 프로토콜(MCP)을 사용하면 안전하고 표준화된 방식으로 LLM 애플리케이션에 데이터와 기능을 제공하는 서버를 구축할 수 있습니다. 웹 API와 유사하지만 LLM 상호 작용을 위해 특별히 설계된 것으로 볼 수 있습니다. MCP 서버는 다음과 같은 작업을 수행할 수 있습니다.
- 리소스를 통해 데이터 노출(이를 GET 엔드포인트와 유사한 것으로 생각하세요. 이는 LLM 컨텍스트에 정보를 로드하는 데 사용됩니다)
- 도구를 통해 기능 제공(POST 엔드포인트와 유사, 코드를 실행하거나 다른 방식으로 부작용을 생성하는 데 사용됨)
- 프롬프트 (LLM 상호작용을 위한 재사용 가능한 템플릿)를 통해 상호작용 패턴 정의
- 그리고 더 많은 것들!
핵심 개념
섬기는 사람
FastMCP 서버는 MCP 프로토콜에 대한 핵심 인터페이스입니다. 연결 관리, 프로토콜 준수 및 메시지 라우팅을 처리합니다.
# Add lifespan support for startup/shutdown with strong typing
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from dataclasses import dataclass
from fake_database import Database # Replace with your actual DB type
from mcp.server.fastmcp import Context, FastMCP
# Create a named server
mcp = FastMCP("My App")
# Specify dependencies for deployment and development
mcp = FastMCP("My App", dependencies=["pandas", "numpy"])
@dataclass
class AppContext:
db: Database
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
"""Manage application lifecycle with type-safe context"""
# Initialize on startup
db = await Database.connect()
try:
yield AppContext(db=db)
finally:
# Cleanup on shutdown
await db.disconnect()
# Pass lifespan to server
mcp = FastMCP("My App", lifespan=app_lifespan)
# Access type-safe lifespan context in tools
@mcp.tool()
def query_db(ctx: Context) -> str:
"""Tool that uses initialized resources"""
db = ctx.request_context.lifespan_context["db"]
return db.query()
자원
리소스는 LLM에 데이터를 노출하는 방식입니다. REST API의 GET 엔드포인트와 유사합니다. 즉, 데이터를 제공하지만 상당한 계산을 수행하거나 부작용을 발생시켜서는 안 됩니다.
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My App")
@mcp.resource("config://app")
def get_config() -> str:
"""Static configuration data"""
return "App configuration here"
@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
"""Dynamic user data"""
return f"Profile data for user {user_id}"
도구
도구를 사용하면 LLM이 서버를 통해 작업을 수행할 수 있습니다. 리소스와 달리 도구는 계산을 수행해야 하며 다음과 같은 부작용이 있습니다.
import httpx
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My App")
@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> float:
"""Calculate BMI given weight in kg and height in meters"""
return weight_kg / (height_m**2)
@mcp.tool()
async def fetch_weather(city: str) -> str:
"""Fetch current weather for a city"""
async with httpx.AsyncClient() as client:
response = await client.get(f"https://api.weather.com/{city}")
return response.text
프롬프트
프롬프트는 LLM이 서버와 효과적으로 상호 작용하는 데 도움이 되는 재사용 가능한 템플릿입니다.
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base
mcp = FastMCP("My App")
@mcp.prompt()
def review_code(code: str) -> str:
return f"Please review this code:\n\n{code}"
@mcp.prompt()
def debug_error(error: str) -> list[base.Message]:
return [
base.UserMessage("I'm seeing this error:"),
base.UserMessage(error),
base.AssistantMessage("I'll help debug that. What have you tried so far?"),
]
이미지
FastMCP는 이미지 데이터를 자동으로 처리하는 Image 클래스를 제공합니다.
from mcp.server.fastmcp import FastMCP, Image
from PIL import Image as PILImage
mcp = FastMCP("My App")
@mcp.tool()
def create_thumbnail(image_path: str) -> Image:
"""Create a thumbnail from an image"""
img = PILImage.open(image_path)
img.thumbnail((100, 100))
return Image(data=img.tobytes(), format="png")
문맥
Context 객체는 도구와 리소스에 MCP 기능에 대한 액세스를 제공합니다.
from mcp.server.fastmcp import FastMCP, Context
mcp = FastMCP("My App")
@mcp.tool()
async def long_task(files: list[str], ctx: Context) -> str:
"""Process multiple files with progress tracking"""
for i, file in enumerate(files):
ctx.info(f"Processing {file}")
await ctx.report_progress(i, len(files))
data, mime_type = await ctx.read_resource(f"file://{file}")
return "Processing complete"
서버 실행
개발 모드
서버를 테스트하고 디버깅하는 가장 빠른 방법은 MCP Inspector를 사용하는 것입니다.
mcp dev server.py
# Add dependencies
mcp dev server.py --with pandas --with numpy
# Mount local code
mcp dev server.py --with-editable .
Claude 데스크톱 통합
서버가 준비되면 Claude Desktop에 설치하세요.
mcp install server.py
# Custom name
mcp install server.py --name "My Analytics Server"
# Environment variables
mcp install server.py -v API_KEY=abc123 -v DB_URL=postgres://...
mcp install server.py -f .env
직접 실행
사용자 정의 배포와 같은 고급 시나리오의 경우:
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My App")
if __name__ == "__main__":
mcp.run()
다음을 사용하여 실행하세요.
python server.py
# or
mcp run server.py
기존 ASGI 서버에 마운팅
sse_app 메서드를 사용하여 SSE 서버를 기존 ASGI 서버에 마운트할 수 있습니다. 이를 통해 SSE 서버를 다른 ASGI 애플리케이션과 통합할 수 있습니다.
from starlette.applications import Starlette
from starlette.routing import Mount, Host
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My App")
# Mount the SSE server to the existing ASGI server
app = Starlette(
routes=[
Mount('/', app=mcp.sse_app()),
]
)
# or dynamically mount as host
app.router.routes.append(Host('mcp.acme.corp', app=mcp.sse_app()))
Starlette에서 애플리케이션을 마운트하는 방법에 대한 자세한 내용은 Starlette 설명서를 참조하세요.
예시
에코 서버
리소스, 도구 및 프롬프트를 보여주는 간단한 서버:
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Echo")
@mcp.resource("echo://{message}")
def echo_resource(message: str) -> str:
"""Echo a message as a resource"""
return f"Resource echo: {message}"
@mcp.tool()
def echo_tool(message: str) -> str:
"""Echo a message as a tool"""
return f"Tool echo: {message}"
@mcp.prompt()
def echo_prompt(message: str) -> str:
"""Create an echo prompt"""
return f"Please process this message: {message}"
SQLite 탐색기
데이터베이스 통합을 보여주는 보다 복잡한 예:
import sqlite3
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("SQLite Explorer")
@mcp.resource("schema://main")
def get_schema() -> str:
"""Provide the database schema as a resource"""
conn = sqlite3.connect("database.db")
schema = conn.execute("SELECT sql FROM sqlite_master WHERE type='table'").fetchall()
return "\n".join(sql[0] for sql in schema if sql[0])
@mcp.tool()
def query_data(sql: str) -> str:
"""Execute SQL queries safely"""
conn = sqlite3.connect("database.db")
try:
result = conn.execute(sql).fetchall()
return "\n".join(str(row) for row in result)
except Exception as e:
return f"Error: {str(e)}"
고급 사용법
저수준 서버
더 세밀한 제어를 위해 저수준 서버 구현을 직접 사용할 수 있습니다. 이를 통해 프로토콜에 대한 전체 접근 권한을 확보하고 수명 주기 API를 통한 수명 주기 관리를 포함하여 서버의 모든 측면을 사용자 지정할 수 있습니다.
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from fake_database import Database # Replace with your actual DB type
from mcp.server import Server
@asynccontextmanager
async def server_lifespan(server: Server) -> AsyncIterator[dict]:
"""Manage server startup and shutdown lifecycle."""
# Initialize resources on startup
db = await Database.connect()
try:
yield {"db": db}
finally:
# Clean up on shutdown
await db.disconnect()
# Pass lifespan to server
server = Server("example-server", lifespan=server_lifespan)
# Access lifespan context in handlers
@server.call_tool()
async def query_db(name: str, arguments: dict) -> list:
ctx = server.request_context
db = ctx.lifespan_context["db"]
return await db.query(arguments["query"])
수명 API는 다음을 제공합니다.
- 서버가 시작될 때 리소스를 초기화하고 중지될 때 리소스를 정리하는 방법
- 핸들러의 요청 컨텍스트를 통해 초기화된 리소스에 액세스
- 수명과 요청 핸들러 간 유형 안전 컨텍스트 전달
import mcp.server.stdio
import mcp.types as types
from mcp.server.lowlevel import NotificationOptions, Server
from mcp.server.models import InitializationOptions
# Create a server instance
server = Server("example-server")
@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
return [
types.Prompt(
name="example-prompt",
description="An example prompt template",
arguments=[
types.PromptArgument(
name="arg1", description="Example argument", required=True
)
],
)
]
@server.get_prompt()
async def handle_get_prompt(
name: str, arguments: dict[str, str] | None
) -> types.GetPromptResult:
if name != "example-prompt":
raise ValueError(f"Unknown prompt: {name}")
return types.GetPromptResult(
description="Example prompt",
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(type="text", text="Example prompt text"),
)
],
)
async def run():
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="example",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
import asyncio
asyncio.run(run())
MCP 클라이언트 작성
SDK는 MCP 서버에 연결하기 위한 고급 클라이언트 인터페이스를 제공합니다.
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
# Create server parameters for stdio connection
server_params = StdioServerParameters(
command="python", # Executable
args=["example_server.py"], # Optional command line arguments
env=None, # Optional environment variables
)
# Optional: create a sampling callback
async def handle_sampling_message(
message: types.CreateMessageRequestParams,
) -> types.CreateMessageResult:
return types.CreateMessageResult(
role="assistant",
content=types.TextContent(
type="text",
text="Hello, world! from model",
),
model="gpt-3.5-turbo",
stopReason="endTurn",
)
async def run():
async with stdio_client(server_params) as (read, write):
async with ClientSession(
read, write, sampling_callback=handle_sampling_message
) as session:
# Initialize the connection
await session.initialize()
# List available prompts
prompts = await session.list_prompts()
# Get a prompt
prompt = await session.get_prompt(
"example-prompt", arguments={"arg1": "value"}
)
# List available resources
resources = await session.list_resources()
# List available tools
tools = await session.list_tools()
# Read a resource
content, mime_type = await session.read_resource("file://some/path")
# Call a tool
result = await session.call_tool("tool-name", arguments={"arg1": "value"})
if __name__ == "__main__":
import asyncio
asyncio.run(run())
MCP 기본 요소
MCP 프로토콜은 서버가 구현할 수 있는 세 가지 핵심 기본 요소를 정의합니다.
| 원어 | 제어 | 설명 | 예시 사용 |
|---|
| 프롬프트 | 사용자 제어 | 사용자 선택에 따라 호출되는 대화형 템플릿 | 슬래시 명령, 메뉴 옵션 |
| 자원 | 애플리케이션 제어 | 클라이언트 애플리케이션에서 관리하는 컨텍스트 데이터 | 파일 내용, API 응답 |
| 도구 | 모델 제어 | LLM에 노출되어 작업을 수행하는 기능 | API 호출, 데이터 업데이트 |
서버 기능
MCP 서버는 초기화 중에 기능을 선언합니다.
| 능력 | 기능 플래그 | 설명 |
|---|
prompts | listChanged | 신속한 템플릿 관리 |
resources | subscribe listChanged | 리소스 노출 및 업데이트 |
tools | listChanged | 도구 발견 및 실행 |
logging | - | 서버 로깅 구성 |
completion | - | 인수 완성 제안 |
선적 서류 비치
기여하다
저희는 모든 수준의 기여자를 지원하는 데 열정을 가지고 있으며, 여러분의 참여를 기다립니다. 참여 가이드를 참조하여 프로젝트에 참여해 주세요.
특허
이 프로젝트는 MIT 라이선스에 따라 라이선스가 부여되었습니다. 자세한 내용은 라이선스 파일을 참조하세요.