MCPサーバーを実装する：最小構成

本レッスンでは、Anthropic 公式 SDK を使って 最小の MCP サーバー を実装します。動かして手で触ってこそ分かるのがプロトコルです。

環境準備

# Python 3.10 以上が必要
python --version

# 仮想環境作成
python -m venv .venv
source .venv/bin/activate  # macOS/Linux
# .venvScriptsactivate  # Windows

# 公式SDKインストール
pip install mcp

最小のMCPサーバー：Hello World

# server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("hello-mcp")

@mcp.tool()
def greet(name: str) -> str:
    """ユーザーに挨拶する。

    Args:
        name: 挨拶する相手の名前
    """
    return f"こんにちは、{name}さん！"

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

動作確認（コマンドライン）

# MCP Inspector でテスト
npx @modelcontextprotocol/inspector python server.py

# ブラウザで http://localhost:5173 が開く
# 「greet」ツールを呼び出して動作確認

Claude Desktopへの登録

# Claude Desktop の設定ファイル
# macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
# Windows: %APPDATA%Claudeclaude_desktop_config.json

{
  "mcpServers": {
    "hello-mcp": {
      "command": "python",
      "args": ["/absolute/path/to/server.py"]
    }
  }
}

# Claude Desktop を再起動

複数ツールを持つMCPサーバー

# tools_server.py
from mcp.server.fastmcp import FastMCP
import json

mcp = FastMCP("multi-tools")

@mcp.tool()
def calculate(expression: str) -> str:
    """数式を計算する。

    Args:
        expression: 数式（例: '2 + 3 * 4'）
    """
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"結果: {result}"
    except Exception as e:
        return f"エラー: {str(e)}"


@mcp.tool()
def search_db(query: str, limit: int = 10) -> str:
    """データベースを検索する。

    Args:
        query: 検索クエリ
        limit: 最大取得件数
    """
    # ダミー実装
    results = [
        {"id": 1, "title": f"{query}に関する記事1"},
        {"id": 2, "title": f"{query}に関する記事2"},
    ]
    return json.dumps(results[:limit], ensure_ascii=False)


@mcp.tool()
def send_notification(channel: str, message: str) -> str:
    """通知を送信する。

    Args:
        channel: 送信先チャネル（例: 'slack', 'email'）
        message: メッセージ本文
    """
    # 実際は Slack API 等を呼ぶ
    return f"{channel}に送信しました: {message}"


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

Resources の追加

Resources は AI が参照できるデータソース。ファイルやURLなどを公開できます。

@mcp.resource("config://app")
def get_app_config() -> str:
    """アプリケーションの現在の設定を返す。"""
    return json.dumps({
        "version": "1.0.0",
        "features": ["chat", "search", "memory"],
        "user_count": 12345,
    })


@mcp.resource("docs://{topic}")
def get_documentation(topic: str) -> str:
    """指定トピックのドキュメントを返す。

    Args:
        topic: トピック名
    """
    docs = {
        "billing": "請求は月末締め、翌月20日支払い...",
        "api": "API エンドポイントは...",
    }
    return docs.get(topic, "ドキュメントが見つかりません")

Prompts の追加

Prompts は テンプレート化されたプロンプト。複雑なプロンプトをサーバー側で管理できる。

@mcp.prompt()
def analyze_logs(logs: str, severity: str = "ERROR") -> str:
    """ログを分析するプロンプト。"""
    return f"""以下のログから {severity} レベル以上のエラーを抽出し、
原因分析と対処法を提案してください。

【ログ】
{logs}

【出力形式】
1. エラー一覧（時刻・種別・メッセージ）
2. 共通する原因（推定）
3. 対処の優先順位
4. 再発防止策"""

ロギングとデバッグ

import logging
import sys

# stderr にログ出力（stdoutはMCP通信に使用）
logging.basicConfig(
    level=logging.INFO,
    stream=sys.stderr,
    format='%(asctime)s [%(levelname)s] %(message)s',
)
logger = logging.getLogger(__name__)


@mcp.tool()
def search_db(query: str, limit: int = 10) -> str:
    logger.info(f"search_db called: query={query}, limit={limit}")
    # ...
    logger.info(f"search_db result: {len(results)} items")
    return json.dumps(results)

エラーハンドリング

from mcp.types import TextContent

@mcp.tool()
def fetch_user(user_id: int) -> str:
    """ユーザー情報を取得する。"""
    try:
        # DB 呼び出し
        user = db.fetch_user(user_id)
        if not user:
            return f"ユーザー {user_id} が見つかりません"
        return json.dumps(user, default=str, ensure_ascii=False)
    except DatabaseError as e:
        logger.error(f"DB error: {e}")
        return f"データベースエラー: {str(e)}"
    except Exception as e:
        logger.exception("Unexpected error")
        return f"予期しないエラーが発生しました"

テストの書き方

# test_server.py
import pytest
from mcp.client.session import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters

@pytest.mark.asyncio
async def test_greet():
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool("greet", {"name": "太郎"})
            assert "太郎" in result.content[0].text

このレッスンのまとめ

FastMCP を使えば、最小10行 でMCPサーバーが立ち上がります。Tools/Resources/Prompts の3つを使い分けて、機能を整理。次のレッスンでは、DB・SaaS との接続パターンを学びます。