AIエージェント開発

【2026年最新】MCPサーバーの作り方|Python FastMCPで実践開発

【2026年最新】MCPサーバーの作り方|Python FastMCPで実践開発

この記事の結論

Python FastMCPを使ったMCPサーバーの構築方法を徹底解説。ツール・リソース・プロンプトの実装からClaude Desktopでのテスト、本番デプロイまで網羅。

AIエージェントが外部ツールやデータベースと連携する際の標準規格として注目されているMCP(Model Context Protocol)。その中核となるのがMCPサーバーです。

本記事では、Python向けMCPフレームワークFastMCPを使って、MCPサーバーをゼロから構築する方法を実践的なコード例とともに解説します。FastMCPは全MCPサーバーの約70%で採用されており、デコレータベースのシンプルなAPIでプロトコルの複雑さを抽象化してくれます。

MCPの基本概念をまだ理解していない方は、先にMCPとは何か?Model Context Protocolの全体像をご覧ください。

MCPサーバーとは何か? ─ AIエージェントの「手足」を作る仕組み

MCPサーバーは、LLM(大規模言語モデル)がデータの取得や外部システムの操作を行うための標準化されたインターフェースを提供するプログラムです。Web APIがブラウザやアプリにデータを提供するのと同様に、MCPサーバーはAIエージェントに対して機能を公開します。

MCPの3つのプリミティブ

MCPサーバーは以下の3種類の機能を公開できます。

  • Tools(ツール):LLMが呼び出せる関数。データベースへの問い合わせ、API呼び出し、ファイル操作など副作用を伴う処理を実行します。
  • Resources(リソース):URIベースで公開される読み取り専用のデータ。設定ファイル、ユーザープロフィール、ドキュメントなどを提供します。
  • Prompts(プロンプト):再利用可能なプロンプトテンプレート。LLMとの対話パターンを標準化します。

MCPサーバーが解決する課題

従来、AIエージェントに外部機能を追加するには、各LLMプロバイダー固有のFunction Calling仕様に合わせた実装が必要でした。MCPはこれを標準化し、一度作ったサーバーをClaude、ChatGPT、Geminiなど複数のLLMクライアントから利用できるようにします。

FastMCPのセットアップと最初のサーバー構築

環境構築

FastMCPはPython 3.10以上が必要です。uv(推奨)またはpipでインストールします。

# uvを使う場合(推奨)
uv init mcp-server-demo
cd mcp-server-demo
uv add fastmcp

# pipを使う場合
pip install fastmcp

2026年3月時点での最新バージョンはFastMCP 3.1.0です。バージョン3.0で導入されたコンポーネントバージョニング、認可制御、OpenTelemetryインストルメンテーションなどの機能が利用できます。

最小構成のMCPサーバー

わずか数行でMCPサーバーを作成できます。

from fastmcp import FastMCP

# MCPサーバーのインスタンスを作成
mcp = FastMCP("demo-server")

@mcp.tool()
def add(a: int, b: int) -> int:
    """2つの数値を足し算します。"""
    return a + b

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

これだけで、LLMから呼び出せる「足し算ツール」を持つMCPサーバーが完成です。FastMCPは関数のシグネチャ(型ヒント)とdocstringから、MCPプロトコルに必要なスキーマ定義を自動生成します。

ツール・リソース・プロンプトの実装パターン

ツールの実装 ─ LLMに「できること」を提供する

ツールはMCPサーバーの最も重要な要素です。外部API呼び出し、データベース操作、ファイル処理など、実際のアクションを実行します。

import httpx
from fastmcp import FastMCP, Context

mcp = FastMCP("weather-server")

@mcp.tool()
async def get_weather(city: str, units: str = "metric") -> dict:
    """指定した都市の現在の天気を取得します。

    Args:
        city: 都市名(例: "Tokyo", "New York")
        units: 温度の単位。metric(摂氏)またはimperial(華氏)
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.weatherapi.com/v1/current.json",
            params={"q": city, "key": "YOUR_API_KEY"}
        )
        data = response.json()
        return {
            "city": data["location"]["name"],
            "temperature": data["current"]["temp_c"],
            "condition": data["current"]["condition"]["text"],
            "humidity": data["current"]["humidity"]
        }

@mcp.tool()
async def search_database(
    query: str,
    table: str,
    limit: int = 10,
    ctx: Context = None
) -> list[dict]:
    """データベースを検索し、条件に合うレコードを返します。

    Args:
        query: 検索クエリ文字列
        table: 検索対象のテーブル名
        limit: 返却する最大レコード数
    """
    # Contextを使ってログを送信
    await ctx.info(f"Searching table '{table}' with query: {query}")

    # データベース検索のシミュレーション
    results = [
        {"id": 1, "name": "Sample Record", "status": "active"}
    ]

    await ctx.info(f"Found {len(results)} results")
    return results

ポイントContextパラメータを追加すると、FastMCPが自動的にコンテキストオブジェクトを注入します。Contextはクライアントへのログ送信やプログレス通知に使えます。ContextパラメータはMCPスキーマには含まれないため、クライアント側からは見えません。

リソースの実装 ─ データを公開する

リソースはURIテンプレートを使って、LLMに読み取り専用のデータを提供します。

import json

@mcp.resource("config://app/settings")
def get_app_settings() -> str:
    """アプリケーションの設定情報を返します。"""
    settings = {
        "app_name": "MyApp",
        "version": "2.1.0",
        "debug": False,
        "max_connections": 100
    }
    return json.dumps(settings, ensure_ascii=False, indent=2)

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """指定されたユーザーのプロフィール情報を返します。"""
    # 実際にはデータベースから取得
    profiles = {
        "u001": {"name": "田中太郎", "role": "engineer", "team": "backend"},
        "u002": {"name": "佐藤花子", "role": "designer", "team": "frontend"},
    }
    profile = profiles.get(user_id, {"error": "User not found"})
    return json.dumps(profile, ensure_ascii=False, indent=2)

@mcp.resource("docs://api/{version}/endpoints")
def get_api_docs(version: str) -> str:
    """APIドキュメントのエンドポイント一覧を返します。"""
    return f"API v{version} のエンドポイント一覧: /users, /products, /orders"

リソースURIはRFC 6570 URIテンプレート構文に従います。{user_id}のようなパラメータは自動的に関数の引数にマッピングされます。

プロンプトの実装 ─ 対話パターンを標準化する

プロンプトは、LLMとの対話における再利用可能なテンプレートを定義します。

from fastmcp.prompts import UserMessage, AssistantMessage

@mcp.prompt()
def code_review(code: str, language: str = "python") -> list:
    """コードレビューを依頼するプロンプトを生成します。

    Args:
        code: レビュー対象のコード
        language: プログラミング言語
    """
    return [
        UserMessage(f"""以下の{language}コードをレビューしてください。
セキュリティ、パフォーマンス、可読性の観点から改善点を指摘してください。

```{language}
{code}
```"""),
    ]

@mcp.prompt()
def sql_generator(description: str, tables: str) -> str:
    """自然言語の説明からSQLクエリを生成するプロンプトです。

    Args:
        description: 取得したいデータの説明
        tables: 使用可能なテーブル定義
    """
    return f"""以下のテーブル定義を参考に、説明に合致するSQLクエリを生成してください。

テーブル定義:
{tables}

要件:
{description}

注意事項:
- JOINを使う場合は適切なインデックスを考慮してください
- LIMITを付けて大量データの取得を防いでください"""

フレームワーク比較 ─ FastMCP vs TypeScript SDK vs その他

MCPサーバーを構築するフレームワークはいくつかあります。プロジェクトの要件に合わせて選択してください。

項目 Python FastMCP TypeScript SDK(公式) TypeScript FastMCP
言語 Python 3.10+ TypeScript / Node.js TypeScript / Node.js
API スタイル デコレータ(@mcp.tool) コールバックベース デコレータ風メソッド
バリデーション Pydantic(ランタイム検証) Zod(コンパイル時+ランタイム) Zod
スキーマ自動生成 型ヒント + docstringから自動 手動定義が必要 部分的に自動
非同期サポート async/await ネイティブ async/await ネイティブ async/await ネイティブ
トランスポート STDIO / Streamable HTTP STDIO / Streamable HTTP STDIO / Streamable HTTP
認証・認可 3.0でグラニュラー認可対応 カスタム実装が必要 組み込みの認証機能
テレメトリ OpenTelemetry組み込み プラグインで対応 なし
開発効率 高(ボイラープレート最小) 中(手動設定が多い)
採用率 全MCP実装の約70% 公式SDK 新興(成長中)
推奨ユースケース データ分析、ML連携、API統合 フロントエンド連携、既存Node.jsプロジェクト TypeScript環境での迅速な開発

結論:Pythonエコシステム(pandas、scikit-learn、SQLAlchemy等)を活用したい場合や、ボイラープレートを最小限にしたい場合はFastMCPが最適です。既存のNode.jsプロジェクトとの統合が必要な場合はTypeScript SDKを選択してください。

Claude Desktopでのテストとデバッグ

Claude Desktop への接続設定

作成したMCPサーバーをClaude Desktopに登録してテストします。

方法1:fastmcp CLIを使う(推奨)

# FastMCPのCLIで自動設定
fastmcp install claude-desktop server.py 
  --server-name "Demo Server" 
  --env API_KEY=your-api-key

方法2:設定ファイルを直接編集する

Claude Desktopの設定ファイルを開きます。

  • macOS~/Library/Application Support/Claude/claude_desktop_config.json
  • Windows%APPDATA%Claudeclaude_desktop_config.json
{
  "mcpServers": {
    "demo-server": {
      "command": "uv",
      "args": ["run", "--directory", "/path/to/mcp-server-demo", "server.py"],
      "env": {
        "API_KEY": "your-api-key"
      }
    }
  }
}

設定ファイルを保存したら、Claude Desktopを再起動します。正常に接続されると、チャット入力欄にツールアイコンが表示され、利用可能なツール一覧を確認できます。

MCP Inspectorでデバッグする

FastMCPにはデバッグ用のインスペクターが組み込まれています。

# MCP Inspectorを起動
fastmcp dev server.py

ブラウザでインスペクターが開き、ツール・リソース・プロンプトの一覧確認、個別のテスト実行、リクエスト/レスポンスの詳細ログ確認ができます。本番デプロイ前のテストに活用してください。

テストコードの記述

FastMCPはプログラム的なテストもサポートしています。

import pytest
from fastmcp import Client

@pytest.mark.anyio
async def test_add_tool():
    """addツールが正しく動作することをテスト"""
    async with Client("server.py") as client:
        result = await client.call_tool("add", {"a": 3, "b": 5})
        assert result[0].text == "8"

@pytest.mark.anyio
async def test_user_profile_resource():
    """ユーザープロフィールリソースのテスト"""
    async with Client("server.py") as client:
        result = await client.read_resource("users://u001/profile")
        import json
        profile = json.loads(result[0].text)
        assert profile["name"] == "田中太郎"

本番デプロイとセキュリティ対策

Streamable HTTPトランスポートでのデプロイ

本番環境では、STDIOではなくStreamable HTTPトランスポートを使用します。SSE(Server-Sent Events)方式は非推奨となっているため、新規デプロイではStreamable HTTPを選択してください。

from fastmcp import FastMCP

mcp = FastMCP(
    "production-server",
    host="0.0.0.0",
    port=8000,
    # 本番向け設定
    log_level="WARNING",
)

@mcp.tool()
async def process_data(input_data: str) -> dict:
    """データを処理して結果を返します。"""
    # 処理ロジック
    return {"status": "processed", "result": input_data.upper()}

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

Dockerでのコンテナ化

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# uv をインストール
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv

# 依存関係をインストール
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev

# アプリケーションコードをコピー
COPY server.py .

EXPOSE 8000

CMD ["uv", "run", "server.py"]
# ビルドと実行
docker build -t mcp-server .
docker run -p 8000:8000 -e API_KEY=your-key mcp-server

セキュリティのベストプラクティス

MCPサーバーは外部からの入力を処理するため、セキュリティ対策が不可欠です。AIエージェントのセキュリティリスクについてはOWASP準拠のAIエージェントセキュリティガイドも参考にしてください。

  • 入力バリデーション:Pydanticモデルを活用して、すべての入力を厳密に検証する
  • 認証・認可:FastMCP 3.0のグラニュラー認可機能でツールごとにアクセス制御を設定する
  • 環境変数管理:APIキーやシークレットはコードにハードコードせず、環境変数経由で注入する
  • レート制限:過度なリクエストを防ぐためにレート制限を実装する
  • ログとモニタリング:OpenTelemetryでツール呼び出しをトレースし、異常な利用パターンを検出する
from pydantic import BaseModel, Field, field_validator

class DatabaseQuery(BaseModel):
    """安全なデータベースクエリのバリデーションモデル"""
    table: str = Field(..., pattern=r"^[a-zA-Z_][a-zA-Z0-9_]*$")
    limit: int = Field(default=10, ge=1, le=100)
    query: str = Field(..., max_length=500)

    @field_validator("table")
    @classmethod
    def validate_table(cls, v: str) -> str:
        allowed_tables = {"users", "products", "orders"}
        if v not in allowed_tables:
            raise ValueError(f"Table '{v}' is not allowed")
        return v

@mcp.tool()
async def safe_query(params: DatabaseQuery) -> list[dict]:
    """バリデーション済みのパラメータでデータベースを検索します。"""
    # params は Pydantic によって自動的に検証済み
    return [{"table": params.table, "results": "..."}]

スケーリングと運用

本番環境でのスケーリングに関する注意点です。

  • セッション管理:デフォルトではセッションはインメモリに保存されます。水平スケーリングする場合は、stateless_http=Trueを設定し、Redisなどの外部ストアにセッションを保存してください。
  • 同時接続数:Streamable HTTPトランスポートでは、現代的なCPUで最大1,200の同時接続を低レイテンシーで処理できます。
  • ヘルスチェック:ロードバランサー用のヘルスチェックエンドポイントを用意し、コンテナオーケストレーションと連携させてください。

実践的なMCPサーバー構築例 ─ 社内ナレッジベース

ここまでの知識を統合して、実用的なMCPサーバーの全体像を示します。

"""
社内ナレッジベースMCPサーバー
- ドキュメント検索ツール
- 部署別リソース
- Q&Aプロンプトテンプレート
"""
from fastmcp import FastMCP, Context
import json

mcp = FastMCP(
    "knowledge-base",
    version="1.0.0",
)

# ===== ツール =====

@mcp.tool()
async def search_docs(
    query: str,
    department: str = "all",
    max_results: int = 5,
    ctx: Context = None
) -> list[dict]:
    """社内ドキュメントを全文検索します。

    Args:
        query: 検索キーワード
        department: 部署フィルタ(all, engineering, sales, hr)
        max_results: 返却する最大件数(1-20)
    """
    await ctx.info(f"Searching docs: query='{query}', dept='{department}'")

    # 実際にはElasticsearchやベクトルDBに接続
    results = [
        {
            "title": "リモートワーク規定 2026年版",
            "department": "hr",
            "summary": "リモートワークの申請手順と注意事項...",
            "url": "https://wiki.example.com/remote-work-2026"
        }
    ]

    await ctx.info(f"Found {len(results)} documents")
    return results[:max_results]

@mcp.tool()
async def submit_inquiry(
    subject: str,
    body: str,
    category: str,
    priority: str = "normal"
) -> dict:
    """社内問い合わせを起票します。

    Args:
        subject: 問い合わせ件名
        body: 問い合わせ内容
        category: カテゴリ(IT, HR, General)
        priority: 優先度(low, normal, high)
    """
    # 実際にはチケットシステムのAPIを呼び出す
    ticket_id = "TKT-2026-0042"
    return {
        "ticket_id": ticket_id,
        "status": "created",
        "message": f"問い合わせ {ticket_id} を作成しました"
    }

# ===== リソース =====

@mcp.resource("kb://departments/{dept}/faq")
def get_department_faq(dept: str) -> str:
    """部署別のFAQを返します。"""
    faqs = {
        "engineering": [
            {"q": "VPNの接続方法は?", "a": "社内Wiki参照"},
            {"q": "開発環境のセットアップ手順は?", "a": "README.mdを確認"},
        ],
        "hr": [
            {"q": "有給休暇の申請方法は?", "a": "勤怠システムから申請"},
            {"q": "健康診断の予約は?", "a": "年1回、人事部から案内"},
        ],
    }
    return json.dumps(
        faqs.get(dept, [{"q": "該当部署のFAQが見つかりません", "a": ""}]),
        ensure_ascii=False,
        indent=2
    )

@mcp.resource("kb://company/policies")
def get_company_policies() -> str:
    """全社共通ポリシーの一覧を返します。"""
    return json.dumps({
        "policies": [
            "情報セキュリティポリシー",
            "リモートワーク規定",
            "経費精算ルール",
            "ハラスメント防止ガイドライン"
        ]
    }, ensure_ascii=False, indent=2)

# ===== プロンプト =====

@mcp.prompt()
def onboarding_guide(role: str, department: str) -> str:
    """新入社員向けのオンボーディング手順を案内するプロンプト。

    Args:
        role: 職種(engineer, designer, sales等)
        department: 配属先部署
    """
    return f"""新入社員のオンボーディングをサポートしてください。

職種: {role}
配属先: {department}

以下の手順に沿って案内してください:
1. 社内システムのアカウント設定
2. 部署固有のツール・環境セットアップ
3. 必読ドキュメントの案内
4. メンター・チームメンバーの紹介
5. 最初の1週間のスケジュール確認

社内ナレッジベースの情報を活用して、具体的に回答してください。"""

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

あわせて読みたいMastra TypeScript AIフレームワーク完全ガイド

次のステップ ─ MCPサーバー開発をさらに進めるために

本記事で解説したFastMCPの基礎を踏まえ、さらにスキルを高めるための次のステップを紹介します。

1. サーバー構成(Server Composition)を活用する

FastMCP 3.0では、複数のMCPサーバーを1つに統合するサーバー構成機能が使えます。マイクロサービス的にドメインごとのサーバーを作り、ゲートウェイサーバーで統合するアーキテクチャが推奨されます。

from fastmcp import FastMCP

# 個別のサーバーをインポート
from knowledge_server import mcp as knowledge
from ticket_server import mcp as tickets

# ゲートウェイサーバーで統合
gateway = FastMCP("company-gateway")
gateway.mount("knowledge", knowledge)
gateway.mount("tickets", tickets)

if __name__ == "__main__":
    gateway.run(transport="streamable-http")

2. OpenAPIプロバイダーで既存APIを即座にMCP化する

FastMCP 3.0のOpenAPIプロバイダーを使えば、OpenAPI仕様書(Swagger)から自動的にMCPツールを生成できます。既存のREST APIを持つシステムとの統合が格段に楽になります。

3. Claude Agent SDKとの連携

MCPサーバーはClaude Agent SDKと組み合わせることで、より高度なAIエージェントを構築できます。Agent SDKのツールとしてMCPサーバーを接続し、マルチステップの推論と外部システム操作を組み合わせたエージェントを実装してみてください。

4. 学習リソース

MCPは2026年のAIエージェント開発において不可欠な技術です。まずは小さなツールサーバーから始めて、段階的に本番レベルのサーバーへと発展させていきましょう。

AIエージェント導入・開発のご相談

株式会社Uravationでは、AIエージェントの導入支援・カスタム開発・研修を提供しています。

無料相談はこちら →

あわせて読みたい: MCPプロトコル最新ガイド

Need help moving from reading to rollout?

この記事を読んで導入イメージが固まってきた方へ

Uravationでは、AIエージェントの要件整理、PoC設計、社内導入、研修まで一気通貫で支援しています。

この記事をシェア

X Facebook LINE

※ 本記事の情報は2026年3月時点のものです。サービスの料金・仕様は変更される可能性があります。最新情報は各サービスの公式サイトをご確認ください。

関連記事