MCP Python SDK
モデルコンテキストプロトコル (MCP) の Python 実装
目次
概要
モデルコンテキストプロトコル（MCP）は、アプリケーションがLLMに標準化された方法でコンテキストを提供することを可能にし、コンテキスト提供に関する懸念をLLMとの実際のやり取りから分離します。このPython SDKはMCP仕様を完全に実装しており、以下のことが容易になります。
- 任意のMCPサーバーに接続できるMCPクライアントを構築する
- リソース、プロンプト、ツールを公開するMCPサーバーを作成する
- stdioやSSEなどの標準トランスポートを使用する
- すべてのMCPプロトコルメッセージとライフサイクルイベントを処理する
インストール
PythonプロジェクトにMCPを追加する
Python プロジェクトを管理するには、 uvを使用することをお勧めします。
UV 管理プロジェクトをまだ作成していない場合は、作成します。
uv init mcp-server-demo
cd mcp-server-demo
次に、プロジェクトの依存関係に 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アプリケーションにデータと機能を安全かつ標準化された方法で公開するサーバーを構築できます。LLMとのやり取りに特化したWeb APIのようなものと考えてください。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 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 プロトコルは、サーバーが実装できる 3 つのコア プリミティブを定義します。
| 原生的 | コントロール | 説明 | 使用例 |
|---|
| プロンプト | ユーザー制御 | ユーザーの選択によって呼び出されるインタラクティブなテンプレート | スラッシュコマンド、メニューオプション |
| リソース | アプリケーション制御 | クライアントアプリケーションによって管理されるコンテキストデータ | ファイルの内容、API 応答 |
| ツール | モデル制御 | LLMに公開されアクションを実行する関数 | API呼び出し、データ更新 |
サーバー機能
MCP サーバーは初期化中に機能を宣言します。
| 能力 | 機能フラグ | 説明 |
|---|
prompts | listChanged | プロンプトテンプレート管理 |
resources | subscribe listChanged | リソースの公開と更新 |
tools | listChanged | ツールの発見と実行 |
logging | - | サーバーのログ記録構成 |
completion | - | 引数補完の提案 |
ドキュメント
貢献
私たちは、経験レベルを問わず、貢献者の方々を熱心にサポートし、プロジェクトへのご参加を心よりお待ちしております。まずは貢献ガイドをご覧ください。
ライセンス
このプロジェクトは MIT ライセンスに基づいてライセンスされています - 詳細については LICENSE ファイルを参照してください。