MCPサーバーの構築方法：コード例付きステップバイステップガイド

モデルコンテキストプロトコル（MCP）は、AIモデルを実世界のツールやサービスに接続するための新しい標準になりつつあります。MCPサーバーを構築することで、シンプルで標準化されたインターフェースを通じて、データ、アクション、リソースをClaudeのようなLLMに公開することができます。

このガイドでは、Pythonで基本的なMCPサーバーをセットアップし、リソースとツールを定義し、MCPクライアントに接続する方法をステップバイステップで学びます。

MCPサーバーとは

MCPサーバーは、LLMとデータベース、ファイルストレージ、APIなどの外部システムの間の橋渡しとして機能します。リソース（読み取り可能なデータ）、ツール（アクション）、プロンプト（指示）をLLMがタスク中に安全に使用できる方法で定義します。

各モデルやツールごとにカスタム統合を作成する代わりに、MCPはプロトコルバージョン0.1（2025年4月現在）で動作する普遍的な標準を提供します。

開始前に必要なもの

• Python 3.8以降
• Pythonスクリプトの基本的な経験
• Python用MCP SDK（pipで入手可能）
• Claude DesktopやCursorなどのMCP互換クライアント（テスト用にオプション）
• バージョン管理用のGit（推奨）
• テキストエディタまたはIDE（Visual Studio Code推奨）

コア構造の理解

MCPでは：

• サーバー：LLMにリソースとツールを提供します。
• クライアント：LLMをサーバーに接続します。
• プロトコル：クライアントとサーバー間の通信を管理します。

次の2つの重要なプリミティブを定義します：

• リソース：LLMが読み取れる静的または動的な情報。
• ツール：LLMが実行できる呼び出し可能な関数。

通信フローは次のように機能します：

1. LLM（クライアント経由）がサーバーからデータやアクションをリクエスト
2. サーバーがこれらのリクエストを処理し、標準化された応答を返す
3. LLMはこの情報を推論や応答に使用できる

1. Pythonプロジェクトのセットアップ

プロジェクトディレクトリとPython仮想環境を作成することから始めます。

mkdir my_mcp_server
cd my_mcp_server
python -m venv venv
source venv/bin/activate  # Linux/Mac
venvScriptsactivate     # Windows

基本的なプロジェクト構造を作成します：

mkdir -p src/resources src/tools tests
touch src/__init__.py src/resources/__init__.py src/tools/__init__.py
touch requirements.txt README.md

requirements.txtに以下を追加します：

mcp-server>=0.1.0
pydantic>=2.0.0
pytest>=7.0.0

2. MCP SDKのインストール

Python用MCP Server SDKとその他の依存関係をインストールします：

pip install -r requirements.txt

公式SDKがまだ公開されていない場合は、GitHubリポジトリからインストールする必要があるかもしれません：

pip install git+https://github.com/anthropic/mcp-server-python.git

3. 基本的なMCPサーバーの作成

src/server.pyというファイルを作成します：

from typing import Dict, Any
from mcp_server import MCPServer
import logging

# ロギングの設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("mcp_server")

def main() -> None:
    """MCPサーバーを初期化して起動します。"""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="シンプルなMCPサーバーの例"
        )
        
        # リソースとツールはここに追加されます
        
        logger.info("MCPサーバーを起動しています...")
        server.start()
    except Exception as e:
        logger.error(f"MCPサーバーの起動に失敗しました: {e}")
        raise

if __name__ == "__main__":
    main()

これで適切なロギングを備えた基本的なMCPサーバーがセットアップされます。

4. リソースの定義

リソースはモデルが読み取れるデータを公開します。src/resources/user_profiles.pyファイルを作成しましょう：

from typing import List, Dict, Any
from pydantic import BaseModel
import logging

logger = logging.getLogger("mcp_server.resources")

class UserProfile(BaseModel):
    """ユーザープロファイルのデータモデル。"""
    name: str
    role: str
    department: str = "General"
    years_experience: int = 0

def fetch_user_profiles() -> List[Dict[str, Any]]:
    """
    データベースからユーザープロファイルを取得します。
    
    Returns:
        List[Dict[str, Any]]: ユーザープロファイル辞書のリスト。
    """
    try:
        # 実際の実装ではデータベースにクエリを実行します
        # この例では、モックデータを返します
        users = [
            UserProfile(name="Alice", role="Engineer", department="Engineering", years_experience=5),
            UserProfile(name="Bob", role="Product Manager", department="Product", years_experience=3),
            UserProfile(name="Charlie", role="Designer", department="Design", years_experience=7)
        ]
        
        logger.info(f"{len(users)}件のユーザープロファイルの取得に成功しました")
        return [user.model_dump() for user in users]
    except Exception as e:
        logger.error(f"ユーザープロファイルの取得中にエラーが発生しました: {e}")
        # 本番環境では、エラー処理戦略に応じて空のリストを返すか
        # 特定の例外を発生させることもあります
        return []

次に、このリソースを含めるためにsrc/server.pyを更新します：

from typing import Dict, Any
from mcp_server import MCPServer, Resource
import logging
from src.resources.user_profiles import fetch_user_profiles

# ... 既存のコード ...

def main() -> None:
    """MCPサーバーを初期化して起動します。"""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="シンプルなMCPサーバーの例"
        )
        
        # ユーザープロファイルリソースを追加
        user_profiles = Resource(
            name="user_profiles",
            description="会社データベースからのユーザープロファイルのリスト。",
            fetch_fn=fetch_user_profiles
        )
        
        server.add_resource(user_profiles)
        
        logger.info("MCPサーバーを起動しています...")
        server.start()
    except Exception as e:
        logger.error(f"MCPサーバーの起動に失敗しました: {e}")
        raise

if __name__ == "__main__":
    main()

これでLLMはMCPクライアントを通じてuser_profilesをクエリできるようになりました。

5. ツールの定義

ツールを使用すると、LLMがアクションを実行できます。src/tools/user_management.pyファイルを作成します：

from typing import Dict, Any, Optional
from pydantic import BaseModel, Field, ValidationError
import logging

logger = logging.getLogger("mcp_server.tools")

class CreateUserRequest(BaseModel):
    """ユーザー作成リクエストの検証モデル。"""
    name: str = Field(..., min_length=2, description="ユーザーのフルネーム")
    role: str = Field(..., min_length=2, description="ユーザーの職務")
    department: Optional[str] = Field("General", description="ユーザーの部署")
    years_experience: Optional[int] = Field(0, ge=0, description="専門的な経験年数")

def create_user_profile(request_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    データベースに新しいユーザープロファイルを作成します。
    
    Args:
        request_data (Dict[str, Any]): 名前、役割などを含むユーザーデータ。
        
    Returns:
        Dict[str, Any]: ステータスとユーザー情報を含むレスポンス
    """
    try:
        # 入力データを検証
        user_data = CreateUserRequest(**request_data)
        
        # 実際の実装ではデータベースに挿入します
        # この例では、アクションをログに記録するだけです
        logger.info(f"新しいユーザーを作成: {user_data.name} - {user_data.role} in {user_data.department}")
        
        # 作成されたユーザーデータを含む成功レスポンスを返す
        return {
            "status": "success",
            "message": f"ユーザー {user_data.name} が正常に作成されました",
            "user": user_data.model_dump()
        }
    except ValidationError as e:
        # 検証エラーの処理
        logger.error(f"検証エラー: {e}")
        return {
            "status": "error",
            "message": "無効なユーザーデータが提供されました",
            "details": str(e)
        }
    except Exception as e:
        # その他のエラーの処理
        logger.error(f"ユーザー作成中にエラーが発生しました: {e}")
        return {
            "status": "error",
            "message": "ユーザーの作成に失敗しました",
            "details": str(e)
        }

次に、このツールを含めるためにsrc/server.pyを更新します：

from typing import Dict, Any
from mcp_server import MCPServer, Resource, Tool
import logging
from src.resources.user_profiles import fetch_user_profiles
from src.tools.user_management import create_user_profile

# ... 既存のコード ...

def main() -> None:
    """MCPサーバーを初期化して起動します。"""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="シンプルなMCPサーバーの例"
        )
        
        # ユーザープロファイルリソースを追加
        user_profiles = Resource(
            name="user_profiles",
            description="会社データベースからのユーザープロファイルのリスト。",
            fetch_fn=fetch_user_profiles
        )
        
        # ユーザー作成ツールを追加
        create_user = Tool(
            name="create_user_profile",
            description="データベースに新しいユーザープロファイルを作成します。",
            parameters={
                "name": {"type": "string", "description": "ユーザーのフルネーム"},
                "role": {"type": "string", "description": "ユーザーの職務"},
                "department": {"type": "string", "description": "ユーザーの部署（オプション）"},
                "years_experience": {"type": "integer", "description": "経験年数（オプション）"}
            },
            execute_fn=create_user_profile
        )
        
        server.add_resource(user_profiles)
        server.add_tool(create_user)
        
        logger.info("MCPサーバーを起動しています...")
        server.start()
    except Exception as e:
        logger.error(f"MCPサーバーの起動に失敗しました: {e}")
        raise

6. エラー処理と検証

検証ロジックを集中化するためにsrc/utils/validation.pyファイルを作成します：

from typing import Dict, Any, List, Optional, Type
from pydantic import BaseModel, ValidationError
import logging

logger = logging.getLogger("mcp_server.validation")

def validate_request(
    data: Dict[str, Any],
    model_class: Type[BaseModel]
) -> tuple[Optional[BaseModel], Optional[Dict[str, Any]]]:
    """
    リクエストデータをPydanticモデルに対して検証します。
    
    Args:
        data: 検証する入力データ
        model_class: 検証に使用するPydanticモデルクラス
        
    Returns:
        tuple: (validated_model, error_dict)
            - 有効な場合: (モデルインスタンス, None)
            - 無効な場合: (None, エラー辞書)
    """
    try:
        validated_data = model_class(**data)
        return validated_data, None
    except ValidationError as e:
        errors = e.errors()
        error_dict = {
            "status": "error",
            "message": "検証に失敗しました",
            "errors": errors
        }
        logger.error(f"検証エラー: {errors}")
        return None, error_dict

この汎用関数は、すべてのツールで入力データを一貫して検証するために使用できます。

7. MCPサーバーの実行とテスト

サーバーが正常に動作することを確認するための簡単なテストスクリプトtest_server.pyを作成します：

import requests
import json
import time
import subprocess
import sys
from pathlib import Path

def test_server():
    """MCPサーバーが正しく実行されていることを確認するための簡単なテスト。"""
    # 別のプロセスでサーバーを起動
    server_process = subprocess.Popen([sys.executable, "src/server.py"])
    
    try:
        # サーバーが起動するのを待つ
        time.sleep(2)
        
        # MCPクライアントを使用してサーバーをテスト
        # 実際のテストでは、MCP Client SDKを使用します
        # この例では、HTTPリクエストを使用してクライアントをシミュレートします
        
        # サーバーがlocalhost:8000で実行されていると仮定
        base_url = "http://localhost:8000"
        
        # リソースの取得をテスト
        response = requests.get(f"{base_url}/resources/user_profiles")
        assert response.status_code == 200
        data = response.json()
        print("リソースレスポンス:", json.dumps(data, indent=2))
        
        # ツールの実行をテスト
        tool_data = {
            "name": "Test User",
            "role": "Tester",
            "department": "QA"
        }
        response = requests.post(
            f"{base_url}/tools/create_user_profile",
            json=tool_data
        )
        assert response.status_code == 200
        data = response.json()
        print("ツールレスポンス:", json.dumps(data, indent=2))
        
        print("すべてのテストに合格しました！")
        
    finally:
        # クリーンアップ: サーバープロセスを終了
        server_process.terminate()
        server_process.wait()

if __name__ == "__main__":
    test_server()

サーバーを実行します：

python src/server.py

サーバーが実行されているとき、別のターミナルでは次のような出力が表示されるかもしれません：

2025-04-28 10:15:23 - mcp_server - INFO - MCPサーバーを起動しています...
2025-04-28 10:15:23 - mcp_server - INFO - サーバーが0.0.0.0:8000でリッスンしています
2025-04-28 10:15:30 - mcp_server.resources - INFO - 3件のユーザープロファイルの取得に成功しました
2025-04-28 10:15:35 - mcp_server.tools - INFO - 新しいユーザーを作成: Test User - Tester in QA

次に、サーバーURLまたはサーバーを起動するコマンドを提供することで、MCPクライアント（Claude Desktopなど）をローカルMCPサーバーに接続するように設定します。

セキュリティに関する考慮事項

MCPサーバーをデプロイする際は、次のセキュリティのベストプラクティスを検討してください：

1. 認証: APIキーやOAuthを実装してクライアントを認証します。

def authenticate_request(request):
    api_key = request.headers.get("X-API-Key")
    if not api_key or api_key != os.environ.get("MCP_API_KEY"):
        raise ValueError("無効なAPIキー")

2. 入力検証: Pydanticモデルを使用してすべての入力を常に検証します。
3. レート制限: 乱用を防ぐためにレート制限を実装します。
4. HTTPS: 本番環境では常にHTTPSを使用します。
5. 制限されたアクション: ツールができることの明確な境界を定義します。

パフォーマンス最適化

1. キャッシング: 高コストのリソース取得をキャッシュします：

from functools import lru_cache

@lru_cache(maxsize=128, ttl=300)  # 5分間キャッシュ
def fetch_user_profiles():
    # 高コストのデータベースクエリ
    pass

2. 非同期処理: I/Oバウンドの操作には非同期を使用します：

async def fetch_user_profiles():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com/users") as response:
            data = await response.json()
            return data

3. 接続プーリング: データベースアクセスには接続プールを使用します。

デプロイメント

ローカル開発

ローカル開発では、次のように実行します：

python src/server.py

Dockerデプロイメント

Dockerfileを作成します：

FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["python", "src/server.py"]

ビルドして実行：

docker build -t mcp-server .
docker run -p 8000:8000 mcp-server

クラウドデプロイメント（AWS）

1. EC2インスタンスを作成するか、AWS App Runnerを使用
2. Dockerコンテナをデプロイ
3. Application Load Balancerをセットアップ
4. セキュリティグループを設定してアクセスを制限

MCPサーバーのテスト

テストファイルtests/test_resources.pyを作成します：

import pytest
from src.resources.user_profiles import fetch_user_profiles

def test_fetch_user_profiles():
    """ユーザープロファイルが正常に取得できることをテストします。"""
    profiles = fetch_user_profiles()
    
    # 構造を確認
    assert isinstance(profiles, list)
    assert len(profiles) > 0
    
    # コンテンツを確認
    first_profile = profiles[0]
    assert "name" in first_profile
    assert "role" in first_profile
    assert isinstance(first_profile["name"], str)

テストを実行：

pytest

一般的なエラーとトラブルシューティング

問題 解決策 例 MCPサーバーに接続できない サーバーが実行中でポートが正しいことを確認 netstat -tulpn | grep 8000 LLMがリソースを見つけられない nameとdescriptionフィールドが適切に設定されているか確認 リソースの初期化を確認 ツール実行中のエラー 入力パラメータが期待される型と一致するか検証 検証にPydanticを使用 クライアントが出力を解析できない 関数がJSON直列化可能なデータを返すことを確認 カスタムオブジェクトの代わりに.model_dump()を使用 起動時にサーバーがクラッシュする インポートと環境変数を確認 詳細なログのためにDEBUG=Trueを設定 ツールのタイムアウト 外部API呼び出しのタイムアウト処理を追加 タイムアウト付きのasyncio.wait_for()を使用 認証の失敗 APIキーと権限を確認 リクエストヘッダーを確認 XML/JSON解析エラー 適切なコンテンツタイプヘッダーを使用 Content-Type: application/jsonを設定

次のステップ

基本的なMCPサーバーを構築した後、次の高度な拡張を検討してください：

1. データベース統合: PostgreSQL、MongoDB、その他のデータベースに接続。
2. ファイル操作: ファイルの読み取り、書き込み、変換のためのツールを追加。
3. 外部API: GitHub、Slack、Google Driveなどの人気サービスと統合。
4. Webhook: LLMが他のシステムでイベントをトリガーできるようにする。
5. ストリーミングリソース: 大規模なデータセットのストリーミングをサポート。
6. コンテキスト対応アクション: LLMの現在のコンテキストを理解するツールを追加。

例：データベース接続の追加

import psycopg2
from contextlib import contextmanager

@contextmanager
def get_db_connection():
    """データベース接続コンテキストマネージャーを作成します。"""
    conn = None
    try:
        conn = psycopg2.connect(
            host=os.environ.get("DB_HOST"),
            database=os.environ.get("DB_NAME"),
            user=os.environ.get("DB_USER"),
            password=os.environ.get("DB_PASSWORD")
        )
        yield conn
    finally:
        if conn is not None:
            conn.close()

def fetch_user_profiles_from_db():
    """PostgreSQLデータベースからユーザープロファイルを取得します。"""
    with get_db_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT name, role, department FROM users")
            columns = [desc[0] for desc in cur.description]
            return [dict(zip(columns, row)) for row in cur.fetchall()]

結論

Pythonで簡単なMCPサーバーを構築することは、LLMをより強力にするための扉を開きます。標準化されたリソースとツールを通じてデータとアクションを公開することで、AIシステムが外部サービスと安全かつ有意義に対話することが容易になります。

小さく始め、1つのリソースと1つのツールに焦点を当て、時間をかけてデータベース、クラウドストレージ、内部APIなどのより高度なユースケースに拡張することができます。

MCPエコシステムは急速に成長しており、今これらの標準を実装することで、プロトコルとそれを使用するLLMの両方の改善からアプリケーションが恩恵を受ける態勢が整います。

よくある質問

コーディング経験なしでMCPサーバーを構築できますか？

Pythonの経験が多少必要です。MCPサーバーはリソースとツールの正確な定義が必要なソフトウェアプロセスです。

Python用の公式SDKはありますか？

はい。AnthropicとコントリビューターがPythonやTypeScriptなど複数の言語用のSDKをリリースしています。

MCPサーバーを本番環境にデプロイできますか？

はい。MCPサーバーをクラウドプラットフォームにホスティングし、ファイアウォールの背後に配置し、LLMクライアントに安全に利用可能にすることができます。