FastMCP: MCPサーバーとクライアント構築の完全ガイド

FastMCP: MCPサーバーとクライアント構築の完全ガイド

FastMCP 2.0の全容を解説するチュートリアルです。このライブラリを使えば、Model Context Protocol (MCP) のサーバーとクライアントをPythonらしく、かつ迅速に構築できます。LLMアプリケーション向けのツール、リソース、プロンプトの作成方法を習得しましょう。

レッスン 4 2025-07-03 06:59

FastMCPとClaude Codeの連携

FastMCPとClaude Codeの統合

Claude Codeは、Anthropicが提供するClaudeの公式CLIツールで、MCPサーバーをサポートしています。これにより、独自のFastMCPサーバーを使ってClaudeの機能を拡張できます。この統合により、Claudeがあなたのツール、リソース、ドメイン固有の機能にアクセスできるようになり、強力な開発ワークフローが実現します。

Claude Codeと統合する理由

Claude Codeとの統合には、いくつかの魅力的な利点があります。

  • 開発ワークフローの強化: カスタムツールにClaudeから直接アクセス
  • コンテキストに応じたアシスタンス: Claudeがプロジェクトのリソースや設定を読み取り可能
  • タスクの自動化: MCPツールを通じて複雑な操作を実行
  • ローカルおよびリモートのサポート: ローカル開発サーバーとデプロイ済みサービスの両方に接続
  • リアルタイムインタラクション: ツールの動的な検出と実行

FastMCPサーバーのセットアップ

開発サーバーの作成

開発ワークフロー用に設計されたFastMCPサーバーを作成しましょう。

# dev_server.py
import os
import subprocess
import json
from pathlib import Path
from datetime import datetime
from fastmcp import FastMCP, Context

mcp = FastMCP(
    name="Development Assistant",
    instructions="""
    このサーバーは以下のような開発ツールを提供します:
    - プロジェクト分析とファイル操作
    - Git操作とリポジトリ管理
    - コード品質チェックとテスト
    - 環境と依存関係の管理
    """
)

@mcp.tool
def run_command(command: str, cwd: str = ".") -> dict:
    """シェルコマンドを実行し、結果を返します。"""
    try:
        result = subprocess.run(
            command.split(),
            cwd=cwd,
            capture_output=True,
            text=True,
            timeout=30
        )
        return {
            "command": command,
            "returncode": result.returncode,
            "stdout": result.stdout,
            "stderr": result.stderr,
            "success": result.returncode == 0
        }
    except subprocess.TimeoutExpired:
        return {"error": "コマンドが30秒でタイムアウトしました"}
    except Exception as e:
        return {"error": str(e)}

@mcp.tool
def analyze_project_structure(path: str = ".") -> dict:
    """プロジェクトディレクトリの構造を分析します。"""
    project_path = Path(path)

    if not project_path.exists():
        return {"error": f"パス {path} は存在しません"}

    structure = {
        "root": str(project_path.absolute()),
        "files": [],
        "directories": [],
        "python_files": [],
        "config_files": [],
        "total_size": 0
    }

    config_extensions = {".json", ".yaml", ".yml", ".toml", ".ini", ".cfg"}

    for item in project_path.rglob("*"):
        if item.is_file():
            size = item.stat().st_size
            structure["total_size"] += size

            file_info = {
                "path": str(item.relative_to(project_path)),
                "size": size,
                "modified": datetime.fromtimestamp(item.stat().st_mtime).isoformat()
            }

            structure["files"].append(file_info)

            if item.suffix == ".py":
                structure["python_files"].append(file_info)
            elif item.suffix in config_extensions:
                structure["config_files"].append(file_info)

        elif item.is_dir():
            structure["directories"].append(str(item.relative_to(project_path)))

    return structure

@mcp.tool
async def check_code_quality(file_path: str, ctx: Context) -> dict:
    """様々なリンティングツールを使用してコードの品質をチェックします。"""
    await ctx.info(f"{file_path} のコード品質をチェック中...")

    if not os.path.exists(file_path):
        return {"error": f"ファイル {file_path} が見つかりません"}

    results = {"file": file_path, "checks": {}}

    # 異なるツールでチェック
    tools = {
        "flake8": ["flake8", file_path],
        "pylint": ["pylint", file_path, "--output-format=json"],
        "mypy": ["mypy", file_path]
    }

    for tool, command in tools.items():
        try:
            result = subprocess.run(
                command,
                capture_output=True,
                text=True,
                timeout=15
            )
            results["checks"][tool] = {
                "returncode": result.returncode,
                "output": result.stdout,
                "errors": result.stderr
            }
        except (subprocess.TimeoutExpired, FileNotFoundError):
            results["checks"][tool] = {"error": f"{tool} が利用できないか、タイムアウトしました"}

    return results

@mcp.resource("project://config")
def get_project_config() -> dict:
    """様々な設定ファイルからプロジェクトの設定を取得します。"""
    config = {"found_files": [], "configurations": {}}

    config_files = [
        "pyproject.toml", "setup.py", "requirements.txt", 
        "package.json", "Dockerfile", ".env", "config.yaml"
    ]

    for config_file in config_files:
        if os.path.exists(config_file):
            config["found_files"].append(config_file)
            try:
                with open(config_file, 'r') as f:
                    content = f.read()
                config["configurations"][config_file] = {
                    "content": content[:1000] + ("..." if len(content) > 1000 else ""),
                    "size": len(content)
                }
            except Exception as e:
                config["configurations"][config_file] = {"error": str(e)}

    return config

@mcp.resource("git://status")
def get_git_status() -> dict:
    """現在のGitリポジトリの状態を取得します。"""
    try:
        # Gitリポジトリ内にいるか確認
        subprocess.run(["git", "rev-parse", "--git-dir"], 
                      check=True, capture_output=True)

        # 様々なGit情報を取得
        commands = {
            "branch": ["git", "branch", "--show-current"],
            "status": ["git", "status", "--porcelain"],
            "last_commit": ["git", "log", "-1", "--pretty=format:%H|%an|%ad|%s"],
            "remote": ["git", "remote", "-v"]
        }

        result = {}
        for key, command in commands.items():
            try:
                output = subprocess.run(
                    command, capture_output=True, text=True, check=True
                ).stdout.strip()
                result[key] = output
            except subprocess.CalledProcessError:
                result[key] = "unavailable"

        return result

    except subprocess.CalledProcessError:
        return {"error": "Gitリポジトリではありません"}

if __name__ == "__main__":
    mcp.run(transport="http", host="127.0.0.1", port=8000)

Claude Codeへの接続

サーバーの起動

まず、FastMCPサーバーを起動します。

# HTTPモードでサーバーを起動
python dev_server.py

サーバーは http://127.0.0.1:8000/mcp/ で稼働します。

Claude Codeへのサーバーの追加

Claude CodeのMCP管理コマンドを使用してサーバーを追加します。

# FastMCPサーバーを追加
claude mcp add dev-assistant --transport http http://localhost:8000/mcp/

# サーバーが接続されていることを確認
claude mcp list

代替: STDIOトランスポート

ローカル開発では、STDIOトランスポートの方が都合が良い場合があります。

# dev_server_stdio.py
# ... (上記と同じサーバーコード)

if __name__ == "__main__":
    mcp.run()  # デフォルトでSTDIOを使用

これをClaude Codeに追加します。

claude mcp add dev-assistant --transport stdio "python dev_server_stdio.py"

Claude Codeでのサーバーの使用

接続が完了すると、Claude Codeを通じてMCPサーバーとやり取りできます。

基本的なツール使用法

# Claudeは自動的にツールを検出し、使用します
claude "現在のプロジェクトの構造を分析してください"

# Claudeは analyze_project_structure ツールを呼び出します
# 応答にはファイル数、サイズ、構成が含まれます

リソースアクセス

Claudeは @ メンションを使用してサーバーのリソースにアクセスできます。

# プロジェクト設定にアクセス
claude "このプロジェクトにはどのような設定ファイルがありますか? @dev-assistant:project://config を使用してください"

# Gitステータスを確認
claude "現在のgitステータスは何ですか? @dev-assistant:git://status を使用してください"

複雑なワークフロー

複数のツールを組み合わせて高度な操作が可能です。

claude "src/ディレクトリ内のすべてのPythonファイルのコード品質をチェックし、問題点を要約してください"

# Claudeは以下を実行します:
# 1. analyze_project_structure を使用してPythonファイルを検索
# 2. 各ファイルに対して check_code_quality を実行
# 3. 結果を要約

高度な設定例

データベースツールを備えた開発サーバー

# db_server.py
import sqlite3
from fastmcp import FastMCP

mcp = FastMCP("Database Tools")

@mcp.tool
def query_database(query: str, db_path: str = "app.db") -> dict:
    """データベースに対してSQLクエリを実行します。"""
    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()
        cursor.execute(query)

        if query.strip().upper().startswith("SELECT"):
            results = cursor.fetchall()
            columns = [description[0] for description in cursor.description]
            return {
                "columns": columns,
                "rows": results,
                "count": len(results)
            }
        else:
            conn.commit()
            return {"affected_rows": cursor.rowcount}

    except Exception as e:
        return {"error": str(e)}
    finally:
        conn.close()

@mcp.resource("db://schema")
def get_database_schema() -> dict:
    """データベースのスキーマ情報を取得します。"""
    # データベーススキーマを読み取る実装
    pass

if __name__ == "__main__":
    mcp.run(transport="http", port=8001)

APIテストサーバー

# api_server.py
import requests
from fastmcp import FastMCP, Context

mcp = FastMCP("API Testing Tools")

@mcp.tool
async def test_api_endpoint(url: str, method: str = "GET", 
                          headers: dict = None, data: dict = None,
                          ctx: Context = None) -> dict:
    """APIエンドポイントをテストし、詳細な結果を返します。"""
    await ctx.info(f"{method} {url} をテスト中...")

    try:
        response = requests.request(
            method=method.upper(),
            url=url,
            headers=headers or {},
            json=data,
            timeout=10
        )

        return {
            "url": url,
            "method": method.upper(),
            "status_code": response.status_code,
            "headers": dict(response.headers),
            "response_time": response.elapsed.total_seconds(),
            "body": response.text[:1000] if response.text else None,
            "success": 200 <= response.status_code < 300
        }

    except Exception as e:
        return {
            "url": url,
            "method": method.upper(),
            "error": str(e),
            "success": False
        }

if __name__ == "__main__":
    mcp.run(transport="http", port=8002)

複数サーバーの設定

複数のFastMCPサーバーをClaude Codeに接続できます。

# 複数のサーバーを追加
claude mcp add dev-tools --transport http http://localhost:8000/mcp/
claude mcp add db-tools --transport http http://localhost:8001/mcp/
claude mcp add api-tools --transport http http://localhost:8002/mcp/

# Claudeはこれで全てのサーバーにアクセスできます
claude "プロジェクトの構造をチェックし、次に設定に定義されているAPIエンドポイントをテストしてください"

一般的な問題のトラブルシューティング

サーバーが応答しない

# サーバーが稼働しているか確認
curl http://localhost:8000/mcp/

# サーバーを削除して再追加
claude mcp remove dev-assistant
claude mcp add dev-assistant --transport http http://localhost:8000/mcp/

ツールが見つからない

サーバーが適切に登録され、ツールがデコレートされていることを確認してください。

# ツールが適切にデコレートされていることを確認
@mcp.tool  # このデコレータが必要です
def my_tool():
    pass

リソースアクセスに関する問題

リソースURIの形式を確認してください。

# 正しい形式
claude "Use @server-name:resource://uri"

# 誤り: @server-name/resource://uri

Claude Code統合のベストプラクティス

  1. 明確なツール名: 説明的で分かりやすいツール名を使用する
  2. 適切なドキュメント: 包括的なdocstringを含める
  3. エラーハンドリング: 意味のあるエラーメッセージを提供する
  4. リソースの整理: リソースに論理的なURIスキームを使用する
  5. パフォーマンス: ツールの実行時間を合理的に保つ
  6. セキュリティ: 入力を検証し、危険な操作を制限する

次のステップ

FastMCPサーバーをClaude Codeに統合することで、以下のことが可能になります。

  • ドメイン固有の開発アシスタントの構築
  • 複雑なプロジェクト管理タスクの自動化
  • カスタムコード分析および品質ツールの作成
  • 特殊なテストおよびデプロイメントワークフローの開発

最後のセクションでは、FastMCPサーバーの高度なパターン、本番環境へのデプロイ、スケーリング戦略について掘り下げていきます。