majiayu000 Python MCP Server Development
このスキルは、Python SDKとFastMCPフレームワークを使用したModel Context Protocol (MCP) サーバーの構築を支援します。
いつこのスキルを使用するか
以下の場合に本スキルを活用してください:
- Python でMCPサーバーを新規作成する
- ツール、リソース、プロンプトをMCPサーバーに実装する
- FastMCPデコレータを使用した開発を行う
- Pydanticモデルによる型安全な実装を行う
- STDIOまたはHTTPトランスポートを設定する
- MCP Inspector を使用したテストとデバッグを行う
- 既存のPython MCPサーバーを最適化・リファクタリングする
開発環境のセットアップ
1. プロジェクト初期化
# uvを使用した新規プロジェクト作成
uv init mcp-server-demo
cd mcp-server-demo
# MCP SDKのインストール
uv add "mcp[cli]"
# 開発依存関係（推奨）
uv add --dev pytest pytest-asyncio
2. pyproject.toml の設定例
プロジェクト設定ファイルを参照してください。
ツール実装パターン
基本的なツール
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My Server")
@mcp.tool()
def calculate(a: int, b: int, operation: str) -> int:
"""数値計算を実行します。
Args:
a: 第一オペランド
b: 第二オペランド
operation: 演算子 (add, subtract, multiply, divide)
Returns:
計算結果
"""
if operation == "add":
return a + b
elif operation == "subtract":
return a - b
elif operation == "multiply":
return a * b
elif operation == "divide":
if b == 0:
raise ValueError("ゼロ除算エラー")
return a / b
else:
raise ValueError(f"未知の演算: {operation}")
Pydanticモデルを使用した構造化出力
from pydantic import BaseModel, Field
from typing import List
class WeatherData(BaseModel):
"""天気データ構造"""
temperature: float = Field(description="摂氏温度")
condition: str = Field(description="天候状態")
humidity: float = Field(ge=0, le=100, description="湿度(%)")
wind_speed: float = Field(ge=0, description="風速(m/s)")
@mcp.tool()
def get_weather(city: str, country_code: str = "JP") -> WeatherData:
"""指定された都市の天気情報を取得します。
構造化された天気データを返すため、LLMが解析しやすくなります。
"""
# 実際のAPI呼び出しをここに実装
return WeatherData(
temperature=22.5,
condition="晴れ",
humidity=65.0,
wind_speed=3.2
)
Contextを使用した高度なツール
from mcp.server.fastmcp import Context
from mcp.server.session import ServerSession
@mcp.tool()
async def process_large_file(
file_path: str,
ctx: Context[ServerSession, None]
) -> str:
"""大きなファイルを段階的に処理し、進捗を報告します。"""
# ログ出力（stderrに送信）
await ctx.info(f"処理開始: {file_path}")
# 進捗報告
total_lines = 1000 # 例
for i in range(0, total_lines, 100):
await ctx.report_progress(i, total_lines, f"{i}/{total_lines}行処理完了")
# 処理ロジック
await ctx.info("処理完了")
return f"ファイル処理完了: {file_path}"
リソース実装パターン
静的リソース
@mcp.resource("config://app")
def get_app_config() -> str:
"""アプリケーション設定を返す静的リソース"""
return """
{
"version": "1.0.0",
"debug": false,
"max_connections": 100
}
"""
動的リソース（URIテンプレート）
@mcp.resource("users://{user_id}")
def get_user_profile(user_id: str) -> str:
"""ユーザーIDに基づいてプロファイルを動的に取得"""
# データベースクエリなど
return f"User {user_id} profile data"
@mcp.resource("files://{category}/{filename}")
def get_file_content(category: str, filename: str) -> str:
"""カテゴリとファイル名からファイル内容を取得"""
file_path = f"./data/{category}/{filename}"
with open(file_path, 'r') as f:
return f.read()
プロンプト実装パターン
from mcp.server.fastmcp.prompts import base
@mcp.prompt(title="Code Review Prompt")
def create_code_review_prompt(
code: str,
language: str = "python"
) -> list[base.Message]:
"""コードレビュー用のプロンプトを生成"""
return [
base.UserMessage(f"以下の{language}コードをレビューしてください:"),
base.UserMessage(f"```{language}\n{code}\n```"),
base.AssistantMessage("コードレビューを開始します。")
]
エラーハンドリングのベストプラクティス
from typing import Union
@mcp.tool()
async def safe_api_call(endpoint: str) -> Union[dict, str]:
"""エラーハンドリング付きAPI呼び出し"""
try:
# API呼び出しロジック
response = await make_api_request(endpoint)
return response.json()
except ConnectionError as e:
return f"接続エラー: {str(e)}"
except TimeoutError as e:
return f"タイムアウト: {str(e)}"
except ValueError as e:
return f"バリデーションエラー: {str(e)}"
except Exception as e:
return f"予期しないエラー: {type(e).__name__}: {str(e)}"
トランスポート設定
STDIOトランスポート（デフォルト）
if __name__ == "__main__":
# ローカル実行、Claude Desktopなどで使用
mcp.run() # または mcp.run(transport="stdio")
HTTPトランスポート
if __name__ == "__main__":
# リモートアクセス、Web統合で使用
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8000
)
Starlette/FastAPIへのマウント
from starlette.applications import Starlette
from starlette.routing import Mount
app = Starlette(routes=[
Mount("/mcp", app=mcp.streamable_http_app())
])
テストとデバッグ
MCP Inspector を使用したテスト
# サーバーを起動してInspectorで検査
uv run mcp dev server.py
# ブラウザで http://localhost:5173 を開く
Claude Desktop へのインストール
# Claude Desktopの設定に追加
uv run mcp install server.py
単体テストの例
テストサンプルを参照してください。
セキュリティチェックリスト
- 入力バリデーション: すべてのパラメータをPydanticで検証
- アクセス制御: ファイルシステム操作を許可ディレクトリに制限
- 環境変数: APIキーなどのシークレットをコードにハードコードしない
- エラーメッセージ: 内部実装の詳細を露出しない
- レート制限: 外部API呼び出しにタイムアウトを設定
- ログ: STDIO使用時はstderrのみにログ出力
- 型安全性: すべての関数に型ヒントを付与
一般的な問題と解決策
問題1: STDIO サーバーでログが出力されない
原因: print() を使用すると、JSON-RPCメッセージが破損します。
解決策:
# ❌ 悪い例
print("デバッグメッセージ")
# ✅ 良い例
import sys
sys.stderr.write("デバッグメッセージ\n")
# または Contextを使用
await ctx.info("デバッグメッセージ")
問題2: スキーマバリデーションエラー
原因: 型ヒントとPydanticモデルが一致していません。
解決策:
from pydantic import Field
class Config(BaseModel):
timeout: int = Field(ge=0, le=3600, description="タイムアウト秒")
@mcp.tool()
def set_config(config: Config) -> str:
# 型が完全に一致
return f"設定完了: timeout={config.timeout}"
問題3: リソースが見つからない
原因: URIテンプレートのパラメータ抽出エラー。
解決策:
# URIパターンを明確に定義
@mcp.resource("data://{category}/{id}")
def get_data(category: str, id: str) -> str:
# パラメータ名が一致していることを確認
return f"Category: {category}, ID: {id}"
参考リソース
次のステップ
- プロジェクトテンプレートからサーバーを作成
- MCP Inspectorでツールをテスト
- セキュリティチェックリストを確認
- Claude Desktopで統合テスト