building-python-mcp-servers

Python MCP サーバーの構築

MCP サーバーは、LLM クライアント（Claude Desktop、Claude Code など）にツールを公開します。 呼び出し元は LLM であるため、通常のライブラリとはエラーの発生パターンが異なります。エラーは機械可読でなければならず、すべての入力を信頼できないものとして扱う必要があり、テストスイートがすべて成功しても、ツールが実際に動作するとは限りません。このスキルでは、問題が起きた際に繰り返し現れるパターンをまとめています。

クイックスタート（FastMCP）

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-server")

@mcp.tool()
def read_config(path: str) -> dict:
    """Read a config file. `path` must be absolute."""
    p = Path(path)
    if not p.is_absolute():
        return {"error": "path must be absolute"}
    if not p.exists():
        return {"error": f"no such file: {path}"}
    return {"data": p.read_text()}

if __name__ == "__main__":
    mcp.run()

エラーの契約：例外を送出せずに返し、一貫性を保つ

捕捉されない例外は、LLM にとって推論できない不透明なプロトコルエラーとして表面化します。代わりに、予測可能な形式の構造化された結果を返し、呼び出し元がそれを確認するようにします。

• エラーの形式を一つ選び、すべてで統一します。 "error" キーを持つ辞書が一般的な慣例です。呼び出し元がこのキーを確認する必要があることを文書化してください。
• バッチツールでは、スキップした項目を報告し、握りつぶさないでください。 最もよくある不整合は、単一項目のツールでは項目ごとのエラーを収集する一方で、関連する「ディレクトリ全体に対して実行」するツールでは、読み込みに失敗したファイルを黙って continue することです。自動化では、これは見えないデータ損失につながります。すべてのバッチツールは、結果と項目ごとの skipped／errors リストの両方を返す必要があります。

@mcp.tool()
def validate_dir(path: str) -> dict:
    results, skipped = {}, []
    for f in Path(path).glob("*.md"):
        try:
            results[f.name] = _validate(f)
        except Exception as e:
            skipped.append({"file": f.name, "error": str(e)})  # never silently continue
    return {"results": results, "skipped": skipped}

型に関する落とし穴：YAML は引用符で囲まれていない ISO 形式の日付（date: 2025-06-15）を自動的に str や datetime.datetime ではなく、datetime.date に解析します。str／datetime だけを処理するバリデーターでは、YAML のネイティブな日付を誤って検証済みと判定してしまいます。解析済みの値を検証するときは、パーサーが実際に生成し得る型をすべて列挙してください。

サブプロセス／CLI のラップ

MCP サーバーの多くは、別のツールをシェル経由で実行します。繰り返し発生する問題は3つあります。

1. 終了コードがゼロ以外の場合でも、stdout を破棄しないでください。 多くの CLI は、仕様上、ゼロ以外で終了し（問題を報告するリンターやミューテーションテスターなど）、空の stderr とともに実際の出力を stdout に書き込みます。returncode != 0 の場合に常に f"Error: {result.stderr}" を返すラッパーでは、正常に実行された場合でも、LLM には空の "Error: " が報告されます。

def run_tool(args: list[str]) -> dict:
    r = subprocess.run(args, capture_output=True, text=True)
    return {                       # hand BOTH streams to the model; let it judge
        "returncode": r.returncode,
        "stdout": r.stdout,
        "stderr": r.stderr,
    }

2. 実際にラップするバージョンを固定し、サブコマンドの存在を確認してください。 プロジェクトでは 3.x を固定しているのに、ツールの 2.x CLI を前提にサーバーを作成すると、すでに存在しないサブコマンドやフラグが呼び出され、ラップしたツールはすべて実行時に壊れます。記憶に頼らず、インストール済みバージョンの --help を確認してください。

3. 引数を安全に解析してください。 追加引数の文字列を str.split() で分割すると、引用符で囲まれたスペースを含む引数が壊れます。shlex.split() を使用してください。クライアントが指定した文字列をシェルコマンドに埋め込んではいけません。subprocess.run には argv のリストを渡してください（shell=True は使用しません）。ツールが対象やパスを受け付ける場合、LLM（または LLM が読み取ったコンテンツ）がそれを選んだことを念頭に置き、検証してください。

モジュールレベルのグローバル状態を使わない（サーバーをテスト不能にします）

CLI 引数をインポート時に解析してモジュールのグローバル変数（WORKING_DIR、MAKEFILE_PATH、キャッシュなど）に格納すると、状態をリセットするだけのために、すべてのテストで sys.argv を差し替えて再インポートするための del sys.modules["server"] が必要になります。この方法は壊れやすく、間違いやすいものです。設定はオブジェクトに保持するか、引数として渡し、ファクトリーからツールを構築してください。

def build_server(config: Config) -> FastMCP:
    mcp = FastMCP("my-server")

    @mcp.tool()
    def do_thing(x: str) -> dict:
        return {"result": _work(x, config)}   # config captured, not global

    return mcp

これにより、二重登録も防げます。モジュールレベルの「すべてのツールを作成する」ループと、main() 内の同じループの両方があると、ファイルを直接実行した場合（Claude Desktop と同じく uv run server.py）と、コンソールエントリーポイント経由で実行した場合に、すべてのツールが二重に登録されます。登録箇所は必ず一か所にしてください。

ブロッキング処理をプロトコルのイベントループから切り離す

フレームワークが同期ツール関数をワーカースレッドに移すとは限りません。 一部の FastMCP ランタイムでは、プロトコルのイベントループ上でそのまま実行されます。そのため、5 秒かかる SQLite クエリ、ファイルシステムの走査、依存関係の探索、同期 HTTP 呼び出しは、同じ 5 秒間、ping や無関係なすべてのリクエストをブロックする可能性があります。

ツールを非同期にし、ブロッキング処理だけをスレッドに移してください。

import asyncio

@mcp.tool()
async def find_dependents(item_id: int) -> dict:
    rows = await asyncio.to_thread(repository.find_dependents, item_id)
    return {"items": [row.to_dict() for row in rows]}

接続の所有権に注意してください。イベントループのスレッドで SQLite 接続を作成し、その接続をワーカーに渡してはいけません。repository.find_dependents 内で接続を開いて閉じるか、ハンドオフを明示的に許可する並行処理契約を持つプールまたはドライバーを使用してください。共有され、特定のスレッドに紐づく接続をスレッドラッパーで扱うと、イベントループの停止が断続的なデータベースエラーに置き換わるだけです。

遅いツールの結果だけでなく、応答性もテストしてください。意図的にブロックさせたリポジトリ呼び出しを開始し、それを解放する前に軽量なツール（またはプロトコル ping）を呼び出し、軽量なリクエストが先に完了することを確認してください。

slow = asyncio.create_task(call_tool("find_dependents", {"item_id": 42}))
await entered_worker.wait()

healthy = await asyncio.wait_for(call_tool("health", {}), timeout=0.2)
assert healthy == {"ok": True}

release_worker.set()
await slow

遅い呼び出しだけに対する時間のアサーションでは、イベントループの停止を検出できません。回帰の兆候は、独立したプロトコル通信が進行しなくなることです。

サンプリングは任意のクライアント機能です — ツール内で失敗を処理する

ツール自体が正常に呼び出されたからといって、ctx.sample(...) が必ず動作するとは限りません。接続先のクライアントがサンプリングに対応していない場合もあれば、リクエストの処理中にサンプリングハンドラーが例外を発生させる場合もあります。これらはフレームワーク層では異なる失敗ですが、ツールレベルでは同じ結果です。要求された分析を生成できませんでした。

サンプリングの境界を囲む例外処理をツール内で行い、サーバーの通常のエラー形式に変換してください。フレームワークの外側の例外ラッパーに頼ってはいけません。その時点では、呼び出し元は文書化された契約ではなく、不透明なプロトコル／ツールエラーを受け取ることになります。

async def sample_or_error(ctx: Context, prompt: str) -> dict:
    try:
        response = await ctx.sample(prompt)
    except Exception as exc:
        return {"error": f"sampling failed: {exc}"}
    return {"result": response.text or ""}

無関係なプログラミングエラーがサンプリングの失敗として誤分類されないよう、try ブロックは狭い範囲に保ってください。両方のケースを明示的にテストしてください。サンプリングに対応していないクライアントと、設定済みのサンプリングハンドラーが例外を発生させるケースです。また、ツールが空文字列などの代替値を返す仕様なら、空のサンプリング応答もテストしてください。

配布：単一ファイルとパッケージ

MCP サーバーは単一ファイルとして起動されることがよくあります（uv run server.py）。そのため、次の 2 つのパッケージング上の落とし穴に気づかないまま出荷してしまいがちです。

• モジュール名とパッケージ名の衝突。 トップレベルに server.py があり、同時に server/ パッケージディレクトリもあると、import server はパッケージを解決先とするため（モジュールが隠される）、server:main のようなコンソールエントリーポイントで main が見つからず失敗します。ファイルを直接実行する場合に限って動作します。どちらか一方の名前を選んでください。
• ビルド対象の指定が狭すぎる。 only-include = ["server.py"] のようなビルド設定では、そのファイルだけを含む wheel が作られます。そのため、pip install したユーザーが import server.analyzers を実行すると ModuleNotFoundError が発生します。ローカルでは uv run server.py が動作していても同様です。パッケージを配布する場合は、パッケージとそのデータファイルを含めてください。

PEP 723 の単一ファイルサーバーでは、インラインの # /// script ヘッダーで明示的なバージョンを固定し、pyproject.toml の指定とも一致させてください。推移的依存関係にのみ依存している（インポートしているが宣言していない）と、その仲介依存関係が削除された時点で動作しなくなります。

テスト：ツールが実際に動作することを証明する

サブプロセス／トランスポート層をモックし、argv に特定のトークンが含まれていることを検証すると、ラップ対象のツールには存在しない可能性のあるコマンドを固定してしまいます。その結果、実行時にはすべてのツールが壊れていても、テストスイートは成功し続けます。ここで CI が成功しても、サーバーが動作するとは限りません。

• 少なくとも 1 つは、実際にラップ対象のツール（または実際のサンプルファイル）をエンドツーエンドで呼び出す統合テストを用意してください。
• 名前の衝突がある状況で、単に import server するだけのスモークテストには意味がありません（空のパッケージをインポートする可能性があります）。ツールが実行され、期待どおりの出力を返すことを検証してください。
• エラー時の契約をテストしてください。不正な入力では "error" 形式が返され、バッチツールでは skipped が設定されることを確認してください。

すべてのツールの I/O を信頼できないものとして扱う（プロンプトインジェクション）

ツールの入力、ファイルの内容、特に他のツールの説明は、攻撃者の影響を受ける可能性があり、モデルのコンテキストに渡されます。そのようなテキストを 2 回目の LLM 呼び出しに渡すサーバー自体が、プロンプトインジェクションの攻撃対象になります。出力は参考情報であり、信頼できる情報ではありません。ツールに必要以上のファイルシステム／ネットワークへのアクセス権を与えず、パスを検証し、ツールの出力を信頼できる指示として扱わないでください。

MCP サーバーのチェックリスト

Contract:
- [ ] One consistent error shape; documented that callers check it
- [ ] Batch tools return a per-item skipped/errors list (never silent continue)
- [ ] Inputs validated (absolute paths, allowed types) before use
- [ ] Sampling failures (unsupported client and handler exception) normalized inside the tool

Subprocess:
- [ ] Both stdout and stderr returned; non-zero exit not assumed to be failure
- [ ] Pinned to the wrapped tool's actual version; subcommands verified
- [ ] shlex.split for arg strings; argv list (no shell=True)

Structure:
- [ ] No module-level CLI parsing / global state
- [ ] Tools registered in exactly one place
- [ ] Blocking database/filesystem/network work moved off the protocol event loop
- [ ] Concurrency test proves a lightweight request completes while a slow tool is blocked

Distribution & tests:
- [ ] No server.py / server/ name collision; build includes the whole package
- [ ] PEP 723 header deps pinned and synced with pyproject
- [ ] An integration test exercises a real tool (not just mocked argv)

さらに詳しく

このスキルは、Will McGinnis による 高品質な Python ライブラリの開発ガイド を基にしています。