AI エージェントや LLM アプリケーションが爆発的に普及する中で、「LLM に外部ツールやデータを安全・標準的に接続する仕組み」として Model Context Protocol（MCP） が急速に注目を集めています。
しかし、生の MCP SDK でサーバーを実装しようとすると、プロトコルの細部まで理解した上で大量のボイラープレートを書く必要があり、開発体験が非常に辛い。
そこで登場するのが FastMCP です。
FastMCP = "the fast, Pythonic way to build MCP servers, clients, and applications"
Python の @decorator だけでツール・リソース・プロンプトを宣言でき、スキーマ生成・バリデーション・プロトコル処理はすべて自動。FastAPI が Web 開発を変えたように、FastMCP は MCP 開発を変えました。
目次
- MCP とは何か
- FastMCP の概要と歴史
- インストール
- コアコンセプト：3つのコンポーネント
- Tools（ツール）
- Resources（リソース）
- Prompts（プロンプト）
- Context（コンテキスト）
- トランスポート：STDIO / HTTP / SSE
- クライアントの実装
- 認証・認可
- サーバーの合成：Mount と Import
- Claude Desktop との連携
- テスト
- 本番デプロイ
- 実践サンプル：天気予報 MCP サーバー
- よくあるハマりポイント
- まとめ
MCP とは何か
Model Context Protocol（MCP） は Anthropic が策定したオープン標準で、LLM アプリケーションと外部ツール・データソースを 標準化されたインターフェイス で接続します。
┌─────────────────────────────────────────┐
│ LLM Application │
│ (Claude Desktop, Cursor, etc.) │
└──────────────┬──────────────────────────┘
│ MCP Protocol
┌───────▼────────┐
│ MCP Client │
└───────┬────────┘
│
┌──────────┼──────────┐
▼ ▼ ▼
┌───────┐ ┌───────┐ ┌───────┐
│Server │ │Server │ │Server │
│ A │ │ B │ │ C │
│(DB) │ │(API) │ │(Files)│
└───────┘ └───────┘ └───────┘
MCP サーバーが提供できるものは3種類です：
| コンポーネント | 役割 | REST に例えると |
|---|---|---|
| Tool | LLM が呼び出す関数（副作用あり） | POST エンドポイント |
| Resource | LLM が読み取るデータ（副作用なし） | GET エンドポイント |
| Prompt | LLM とのインタラクションテンプレート | ユーザー起動のテンプレート |
FastMCP の概要と歴史
| バージョン | 内容 |
|---|---|
| v1.0（2024年） | jlowin 氏が開発。後に公式 MCP Python SDK に取り込まれる |
| v2.0 | スタンドアロン版として再出発。Client 機能・認証・Proxy・Middleware 追加 |
| v3.0（2026年1月） | PrefectHQ に引き継がれ、認可・OpenTelemetry・複数 Provider タイプを追加 |
| v3.1+（現在） | Transforms・CodeMode・Search Transform など高度な機能を続々追加中 |
現在は PrefectHQ/fastmcp として GitHub でメンテされており、PyPI での週間ダウンロード数は100万回以上。
インストール
# pip
pip install fastmcp
# uv（推奨）
uv add fastmcp
# 開発用ツールつき
uv add "mcp[cli]"
Python 3.10 以上が必要です。
コアコンセプト：3つのコンポーネント
FastMCP アプリの骨格はシンプルです。
from fastmcp import FastMCP
# サーバーインスタンスの作成
mcp = FastMCP(
name="MyServer",
instructions="このサーバーはデータ分析ツールを提供します。",
)
# Tool / Resource / Prompt をデコレータで宣言
if __name__ == "__main__":
mcp.run() # デフォルトは STDIO トランスポート
Tools（ツール）
ツールは「LLM が呼び出す関数」です。型ヒントと docstring を書くだけで、JSON Schema・バリデーション・ドキュメントが自動生成されます。
基本的なツール
from fastmcp import FastMCP
mcp = FastMCP("Calculator")
@mcp.tool
def add(a: int, b: int) -> int:
"""2つの整数を足し合わせます。"""
return a + b
@mcp.tool
def divide(a: float, b: float) -> float:
"""aをbで割ります。b=0 の場合は ValueError を送出します。"""
if b == 0:
raise ValueError("ゼロ除算はできません")
return a / b
非同期ツール
I/O 待ちが発生する処理（HTTP リクエスト・DB クエリなど）は async def で定義できます。
import httpx
from fastmcp import FastMCP
mcp = FastMCP("WebFetcher")
@mcp.tool
async def fetch_url(url: str) -> str:
"""指定URLのHTMLコンテンツを取得します。"""
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=10)
response.raise_for_status()
return response.text
複合型・Pydantic モデルを使ったツール
from pydantic import BaseModel
from typing import Optional
from fastmcp import FastMCP
mcp = FastMCP("UserManager")
class UserInput(BaseModel):
name: str
email: str
age: Optional[int] = None
class UserResult(BaseModel):
id: int
name: str
email: str
created: bool
@mcp.tool
def create_user(user: UserInput) -> UserResult:
"""新しいユーザーを作成します。"""
# 実際はDBに保存する処理
return UserResult(id=42, name=user.name, email=user.email, created=True)
タグとメタデータ
@mcp.tool(
name="search_products", # ツール名を明示指定
description="商品を検索します。複数の条件でフィルタリング可能。",
tags={"public", "search"}, # タグで管理
)
def search_products(query: str, limit: int = 10) -> list[dict]:
return [{"id": i, "name": f"商品{i}"} for i in range(limit)]
Resources（リソース）
リソースは「LLM が読み取る静的・準静的なデータ」です。
静的リソース
from fastmcp import FastMCP
mcp = FastMCP("DataStore")
@mcp.resource("config://app")
def get_app_config() -> dict:
"""アプリケーション設定を返します。"""
return {
"theme": "dark",
"language": "ja",
"version": "2.0",
}
@mcp.resource("data://readme")
def get_readme() -> str:
"""READMEドキュメントを返します。"""
with open("README.md", "r", encoding="utf-8") as f:
return f.read()
リソーステンプレート（動的 URI）
URI にパラメータを埋め込めます。
@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: int) -> dict:
"""ユーザーIDに対応するプロフィールを返します。"""
# 実際はDBからユーザーを取得
return {
"id": user_id,
"name": f"ユーザー{user_id}",
"status": "active",
}
@mcp.resource("products://{category}/{product_id}")
def get_product(category: str, product_id: str) -> dict:
"""カテゴリとIDで商品を取得します。"""
return {
"category": category,
"id": product_id,
"price": 1980,
}
Prompts（プロンプト）
プロンプトは「ユーザーが起動する再利用可能なテンプレート」です。LLM との一貫したインタラクションパターンを定義できます。
from fastmcp import FastMCP
mcp = FastMCP("PromptLibrary")
@mcp.prompt
def analyze_code(code: str, language: str = "python") -> str:
"""コードレビュー用プロンプトを生成します。"""
return f"""
以下の{language}コードをレビューしてください。
レビュー観点：
1. バグや潜在的なエラー
2. パフォーマンス上の問題
3. コードの読みやすさ
4. セキュリティ上のリスク
{language}
{code}
具体的な改善案をコードと一緒に提示してください。
@mcp.prompt
def summarize_document(text: str, max_words: int = 200) -> str:
"""文書要約プロンプトを生成します。"""
return f"以下の文書を{max_words}語以内で要約してください：\n\n{text}"
Context（コンテキスト）
Context オブジェクトを引数に追加することで、ツール・リソースの内部から MCP クライアントへのログ送信・進捗通知・リソース読み取りが行えます。
from fastmcp import FastMCP, Context
mcp = FastMCP("ProgressDemo")
@mcp.tool
async def process_batch(items: list[str], ctx: Context) -> str:
"""アイテムリストを処理し、進捗を報告します。"""
total = len(items)
results = []
await ctx.info(f"バッチ処理開始: {total}件")
for i, item in enumerate(items):
# 進捗を報告（0.0 〜 1.0）
await ctx.report_progress(
progress=i / total,
total=1.0,
)
await ctx.debug(f"処理中: {item}")
# 実際の処理
results.append(item.upper())
await ctx.info("バッチ処理完了")
return f"処理結果: {', '.join(results)}"
Context の主なメソッド：
| メソッド | 用途 |
|---|---|
ctx.debug(msg) |
デバッグログ送信 |
ctx.info(msg) |
情報ログ送信 |
ctx.warning(msg) |
警告ログ送信 |
ctx.error(msg) |
エラーログ送信 |
ctx.report_progress(progress, total) |
進捗通知 |
await ctx.read_resource(uri) |
サーバー内リソースの読み取り |
トランスポート：STDIO / HTTP / SSE
FastMCP は複数のトランスポートに対応しています。
from fastmcp import FastMCP
mcp = FastMCP("TransportDemo")
if __name__ == "__main__":
# STDIO（デフォルト）: Claude Desktop などローカル統合向け
mcp.run()
# HTTP（Streamable HTTP）: Web サービス向け
# mcp.run(transport="http", host="0.0.0.0", port=8000)
# SSE（Server-Sent Events）: レガシー、非推奨
# mcp.run(transport="sse", host="0.0.0.0", port=8000)
| トランスポート | 用途 | 備考 |
|---|---|---|
stdio |
ローカル CLI・Claude Desktop | デフォルト |
http |
Web サービス・リモートサーバー | 推奨（Streamable HTTP） |
sse |
レガシー Web 統合 | 非推奨 |
CLI から起動する方法
# デフォルト（stdio）
fastmcp run server.py
# HTTP モードで起動
fastmcp run server.py --transport http --port 8000
# MCP Inspector で検査
fastmcp inspect server.py
FastAPI に組み込む
from fastapi import FastAPI
from fastmcp import FastMCP
app = FastAPI()
mcp = FastMCP("Embedded")
@mcp.tool
def hello(name: str) -> str:
return f"こんにちは、{name}！"
# /mcp パス以下に MCP エンドポイントをマウント
app.mount("/mcp", mcp.get_asgi_app())
@app.get("/health")
def health():
return {"status": "ok"}
クライアントの実装
FastMCP は サーバー側だけでなく、クライアント側も提供しています。
import asyncio
from fastmcp import Client
async def main():
# HTTP サーバーに接続
async with Client("http://localhost:8000/mcp") as client:
# 利用可能なツール一覧を取得
tools = await client.list_tools()
print("ツール:", [t.name for t in tools])
# ツールを呼び出す
result = await client.call_tool("add", {"a": 10, "b": 32})
print("結果:", result)
# リソースを読み取る
content = await client.read_resource("config://app")
print("設定:", content)
asyncio.run(main())
STDIO サーバーに接続する
import asyncio
from fastmcp import Client
from fastmcp.client.transports import PythonStdioTransport
async def main():
transport = PythonStdioTransport(
script_path="./my_server.py"
)
async with Client(transport) as client:
result = await client.call_tool("greet", {"name": "世界"})
print(result)
asyncio.run(main())
認証・認可
Bearer Token 認証（シンプルな例）
from fastmcp import FastMCP
from fastmcp.server.auth import BearerTokenProvider
mcp = FastMCP(
name="SecureServer",
auth=BearerTokenProvider(
tokens={"my-secret-token-123"}
),
)
@mcp.tool
def secret_data() -> str:
"""認証済みクライアントのみアクセス可能。"""
return "機密情報"
if __name__ == "__main__":
mcp.run(transport="http")
タグベースのフィルタリング（認可の疑似実装）
from fastmcp import FastMCP
mcp = FastMCP("FilteredServer")
@mcp.tool(tags={"public"})
def public_tool() -> str:
"""誰でも使えるツール。"""
return "パブリックデータ"
@mcp.tool(tags={"admin"})
def admin_tool() -> str:
"""管理者専用ツール。"""
return "管理者データ"
# パブリックツールのみ公開
mcp.enable(tags={"public"}, only=True)
サーバーの合成：Mount と Import
複数のサーバーを組み合わせて大規模なアプリを構築できます。
from fastmcp import FastMCP
# サブサーバーを定義
weather_mcp = FastMCP("WeatherServer")
db_mcp = FastMCP("DatabaseServer")
@weather_mcp.tool
def get_temperature(city: str) -> float:
return 22.5
@db_mcp.tool
def query_users(limit: int = 10) -> list[dict]:
return [{"id": i} for i in range(limit)]
# メインサーバーに組み込む
main_mcp = FastMCP("MainServer")
# プレフィックスつきでマウント（ツール名が weather_get_temperature になる）
main_mcp.mount(weather_mcp, prefix="weather")
main_mcp.mount(db_mcp, prefix="db")
if __name__ == "__main__":
main_mcp.run()
Claude Desktop との連携
Claude Desktop の設定ファイル（~/Library/Application Support/Claude/claude_desktop_config.json）に追記します。
Python スクリプトを直接起動
{
"mcpServers": {
"my-server": {
"command": "python",
"args": ["/path/to/server.py"]
}
}
}
uv を使って起動（推奨）
{
"mcpServers": {
"my-server": {
"command": "uv",
"args": ["run", "--with", "fastmcp", "python", "/path/to/server.py"]
}
}
}
環境変数つき
{
"mcpServers": {
"weather-server": {
"command": "python",
"args": ["/path/to/weather_server.py"],
"env": {
"WEATHER_API_KEY": "your-api-key-here",
"DEBUG": "true"
}
}
}
}
テスト
FastMCP はインメモリでサーバーをテストできる仕組みを提供しています。
import pytest
from fastmcp import FastMCP, Client
# テスト対象のサーバー
mcp = FastMCP("TestServer")
@mcp.tool
def multiply(a: int, b: int) -> int:
"""2つの数を掛け算します。"""
return a * b
@mcp.resource("data://pi")
def get_pi() -> float:
return 3.14159
# テストコード
@pytest.mark.asyncio
async def test_multiply_tool():
async with Client(mcp) as client: # サーバーインスタンスを直接渡す
result = await client.call_tool("multiply", {"a": 6, "b": 7})
assert result[0].text == "42"
@pytest.mark.asyncio
async def test_list_tools():
async with Client(mcp) as client:
tools = await client.list_tools()
tool_names = [t.name for t in tools]
assert "multiply" in tool_names
@pytest.mark.asyncio
async def test_read_resource():
async with Client(mcp) as client:
content = await client.read_resource("data://pi")
assert "3.14" in str(content[0].text)
実行：
pip install pytest pytest-asyncio
pytest tests/ -v
本番デプロイ
Docker を使ったデプロイ
FROM python:3.12-slim
WORKDIR /app
# uv のインストール
RUN pip install uv
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev
COPY . .
EXPOSE 8000
CMD ["uv", "run", "python", "server.py"]
server.py 側：
from fastmcp import FastMCP
import os
mcp = FastMCP("ProductionServer")
@mcp.tool
def greet(name: str) -> str:
return f"こんにちは、{name}！"
if __name__ == "__main__":
mcp.run(
transport="http",
host="0.0.0.0",
port=int(os.getenv("PORT", 8000)),
)
ヘルスチェックエンドポイント
from fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import JSONResponse
mcp = FastMCP("MyServer")
@mcp.custom_route("/health", methods=["GET"])
async def health_check(request: Request) -> JSONResponse:
return JSONResponse({"status": "healthy", "version": "1.0.0"})
if __name__ == "__main__":
mcp.run(transport="http")
実践サンプル：天気予報 MCP サーバー
実際のユースケースに近い、完全なサンプルサーバーです。
"""
weather_server.py
Open-Meteo API（無料・APIキー不要）を使った天気予報 MCP サーバー
"""
import httpx
from typing import Optional
from fastmcp import FastMCP, Context
from pydantic import BaseModel
mcp = FastMCP(
name="WeatherServer",
instructions="""
天気情報を提供するサーバーです。
- get_current_weather: 現在の天気を取得
- get_forecast: 数日間の天気予報を取得
- get_city_coordinates: 都市名から座標を取得
""",
)
# ---- データモデル ----
class WeatherData(BaseModel):
city: str
temperature: float
feels_like: float
humidity: int
description: str
wind_speed: float
class ForecastDay(BaseModel):
date: str
max_temp: float
min_temp: float
precipitation: float
# ---- Tools ----
@mcp.tool
async def get_city_coordinates(city: str, ctx: Context) -> dict:
"""都市名から緯度・経度を取得します。"""
await ctx.info(f"{city} の座標を検索中...")
url = "https://geocoding-api.open-meteo.com/v1/search"
params = {"name": city, "count": 1, "language": "ja", "format": "json"}
async with httpx.AsyncClient() as client:
response = await client.get(url, params=params)
data = response.json()
if not data.get("results"):
raise ValueError(f"都市 '{city}' が見つかりませんでした")
result = data["results"][0]
return {
"city": result["name"],
"country": result.get("country", ""),
"latitude": result["latitude"],
"longitude": result["longitude"],
}
@mcp.tool
async def get_current_weather(city: str, ctx: Context) -> WeatherData:
"""指定した都市の現在の天気を取得します。"""
await ctx.info(f"{city} の天気を取得中...")
# 座標を取得
coords = await get_city_coordinates(city, ctx)
lat, lon = coords["latitude"], coords["longitude"]
url = "https://api.open-meteo.com/v1/forecast"
params = {
"latitude": lat,
"longitude": lon,
"current": [
"temperature_2m",
"apparent_temperature",
"relative_humidity_2m",
"weather_code",
"wind_speed_10m",
],
"timezone": "Asia/Tokyo",
}
async with httpx.AsyncClient() as client:
response = await client.get(url, params=params)
data = response.json()
current = data["current"]
wmo_code = current["weather_code"]
# WMO 天気コードを日本語に変換
desc = _wmo_to_description(wmo_code)
return WeatherData(
city=coords["city"],
temperature=current["temperature_2m"],
feels_like=current["apparent_temperature"],
humidity=current["relative_humidity_2m"],
description=desc,
wind_speed=current["wind_speed_10m"],
)
@mcp.tool
async def get_forecast(
city: str,
days: int = 7,
ctx: Context = None,
) -> list[ForecastDay]:
"""指定した都市の天気予報を取得します（最大16日間）。"""
days = min(days, 16)
await ctx.info(f"{city} の{days}日間予報を取得中...")
coords = await get_city_coordinates(city, ctx)
lat, lon = coords["latitude"], coords["longitude"]
url = "https://api.open-meteo.com/v1/forecast"
params = {
"latitude": lat,
"longitude": lon,
"daily": ["temperature_2m_max", "temperature_2m_min", "precipitation_sum"],
"forecast_days": days,
"timezone": "Asia/Tokyo",
}
async with httpx.AsyncClient() as client:
response = await client.get(url, params=params)
data = response.json()
daily = data["daily"]
return [
ForecastDay(
date=daily["time"][i],
max_temp=daily["temperature_2m_max"][i],
min_temp=daily["temperature_2m_min"][i],
precipitation=daily["precipitation_sum"][i] or 0.0,
)
for i in range(len(daily["time"]))
]
# ---- Resources ----
@mcp.resource("weather://cities/popular")
def get_popular_cities() -> list[str]:
"""よく使われる都市のリストを返します。"""
return [
"東京", "大阪", "名古屋", "札幌", "福岡",
"京都", "横浜", "神戸", "仙台", "広島",
]
# ---- Prompts ----
@mcp.prompt
def weather_report_prompt(city: str) -> str:
"""天気レポート生成用のプロンプトを返します。"""
return f"""
{city} の天気情報を取得して、以下の形式でレポートを作成してください：
## {city} 天気レポート
### 現在の天気
（現在の気温・体感温度・湿度・天気の説明・風速を記載）
### 今後7日間の予報
（日付ごとの最高・最低気温と降水量を表で記載）
### まとめ・アドバイス
（服装や持ち物など、実用的なアドバイスを2〜3文で記載）
"""
# ---- ヘルパー関数 ----
def _wmo_to_description(code: int) -> str:
"""WMO 天気コードを日本語の説明に変換します。"""
mapping = {
0: "快晴", 1: "ほぼ快晴", 2: "一部曇り", 3: "曇り",
45: "霧", 48: "着氷霧",
51: "霧雨（弱）", 53: "霧雨（中）", 55: "霧雨（強）",
61: "雨（弱）", 63: "雨（中）", 65: "雨（強）",
71: "雪（弱）", 73: "雪（中）", 75: "雪（強）",
80: "にわか雨（弱）", 81: "にわか雨（中）", 82: "にわか雨（強）",
95: "雷雨", 96: "ひょうを伴う雷雨", 99: "激しいひょうを伴う雷雨",
}
return mapping.get(code, f"不明な天気（コード: {code}）")
# ---- サーバー起動 ----
if __name__ == "__main__":
import os
transport = os.getenv("MCP_TRANSPORT", "stdio")
if transport == "http":
mcp.run(transport="http", host="0.0.0.0", port=8000)
else:
mcp.run()
このサーバーを Claude Desktop で使うには：
{
"mcpServers": {
"weather": {
"command": "python",
"args": ["/path/to/weather_server.py"]
}
}
}
よくあるハマりポイント
1. デコレータの書き方の違い（v1 vs v2/v3）
# ❌ 古い書き方（MCP Python SDK / FastMCP v1 スタイル）
@mcp.tool() # 括弧あり
def my_tool(): ...
# ✅ FastMCP v2/v3 の書き方（括弧なし）
@mcp.tool # 括弧なし
def my_tool(): ...
# ✅ どちらも動く（引数あり形式は括弧が必要）
@mcp.tool(name="custom_name")
def my_tool(): ...
2. 型ヒントは必ず書く
FastMCP は型ヒントから JSON Schema を生成するため、型ヒントなしだと期待通りに動きません。
# ❌ 型ヒントなし（動くが非推奨）
@mcp.tool
def add(a, b):
return a + b
# ✅ 型ヒントあり
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
3. Context の引数名は ctx に統一
# ✅ FastMCP が自動注入するのは Context 型アノテーションのパラメータ
@mcp.tool
async def my_tool(query: str, ctx: Context) -> str:
await ctx.info("処理中...")
return query
4. if __name__ == "__main__": は省略しない
Claude Desktop などから subprocess として起動される場合、このガードがないと import 時にサーバーが起動してしまいます。
# ✅ 必ずガードする
if __name__ == "__main__":
mcp.run()
まとめ
FastMCP が提供するもの：
| 機能 | 内容 |
|---|---|
| 宣言的な API | デコレータだけでツール・リソース・プロンプトを定義 |
| 自動スキーマ生成 | 型ヒント + docstring → JSON Schema を自動生成 |
| Pydantic 統合 | 入出力のバリデーションを自動化 |
| マルチトランスポート | STDIO・HTTP・SSE に対応 |
| クライアント機能 | サーバーだけでなくクライアント実装も提供 |
| 認証・認可 | Bearer Token・OAuth・タグベースフィルタリング |
| サーバー合成 | Mount / Import で大規模サーバーを構築 |
| テスト支援 | インメモリでサーバーをテスト可能 |
| CLI ツール |
fastmcp run / inspect でデバッグが楽 |
FastMCP を使えば、「LLM に何かさせたい」というアイデアを数分でMCPサーバーとして実装できます。ぜひ試してみてください。