FastMCPの負荷テスト｜同時接続・p95・エラー率を測る

FastMCPサーバーを公開すると、「何ユーザーまで耐えられるのか」「同時接続が増えたとき、どこから遅くなるのか」を確認したくなります。

しかし、MCPは単純なREST APIではありません。Streamable HTTPには初期化、セッション、ツール呼び出しなどのライフサイクルがあります。curlで同じJSONを大量送信するだけでは、実際のMCPクライアントに近い負荷にならないことがあります。

この記事ではFastMCP公式Clientを使い、複数の仮想ユーザーからツールを呼び出す負荷テストを作ります。p50・p95・p99レイテンシ、スループット、エラー率を集計し、性能の頭打ちとボトルネックを見つける方法まで解説します。

この記事の結論
1ユーザーから始め、同時接続を5、10、25、50と段階的に増やします。p95が急増する直前を探し、その値より十分低い位置を運用上限にします。初回は必ずステージング環境と読み取り専用ツールで実施してください。

FastMCP負荷テストで測る4つの指標

最初に、負荷テストの合否を決める指標を固定します。

平均値だけでは、一部の極端に遅い呼び出しが見えにくくなります。ユーザー体験と運用上限を考えるときは、平均よりp95を中心に確認します。

合格条件の例は次の通りです。

• p95が500ミリ秒以下
• エラー率が1%未満
• CPU使用率が継続して80%を超えない
• データベース接続数が上限の70%以内
• 負荷終了後にレイテンシとメモリ使用量が元へ戻る

数値はサービスごとに異なります。テストを始める前に、利用者が許容できる応答時間とインフラ上限から決めてください。

生のHTTPリクエストだけでは不十分な理由

FastMCPのHTTPサーバーはMCPのStreamable HTTP transportを使います。クライアントはサーバーと初期化を行い、構成によっては同じセッションIDを後続リクエストへ引き継ぎます。

そのため、JSON-RPCのtools/callだけを直接POSTすると、次の要素を再現できないことがあります。

• MCP初期化と機能ネゴシエーション
• セッションの作成・維持・終了
• 認証ヘッダーの更新
• SSEやJSON responseの扱い
• FastMCP Client側のシリアライズ

プロトコル自体の限界を調べる目的でなければ、公式Clientを使う方が実利用に近くなります。

テスト対象を準備する

負荷をかける前に、FastMCPサーバーをHTTPで起動します。

fastmcp run server.py:mcp --transport http --port 8000

既定のMCP Endpointは、通常http://127.0.0.1:8000/mcpです。別端末やコンテナから実行するときは、テスト実行元から到達できるURLへ変更します。

最初の対象ツールは、次の条件を満たすものがおすすめです。

• 読み取り専用である
• 同じ引数で何度呼んでも安全である
• 課金API、メール送信、購入、削除を行わない
• 本番データを書き換えない
• 処理結果のサイズが極端に変動しない

LLM APIや検索APIを呼ぶツールは、FastMCPではなく外部サービスの速度・レート制限・費用を測るテストになりがちです。まずローカル処理やテスト用データベースを使うツールで、サーバー自体の基準性能を測ります。

FastMCP Clientで負荷テストを作る

次のスクリプトは、仮想ユーザーごとにFastMCP Clientを1つ作り、同じ接続内で複数回ツールを呼び出します。これにより、実際のクライアントに近いセッション利用を再現できます。

import asyncio
import json
import math
import os
import time
from collections import Counter
from dataclasses import dataclass

from fastmcp import Client
from fastmcp.client.auth import BearerAuth

MCP_URL = os.getenv("MCP_URL", "http://127.0.0.1:8000/mcp")
MCP_TOOL = os.getenv("MCP_TOOL", "search")
MCP_ARGUMENTS = json.loads(
    os.getenv("MCP_ARGUMENTS", '{"query":"fastmcp","limit":5}')
)
USERS = min(int(os.getenv("LOAD_USERS", "5")), 200)
CALLS_PER_USER = int(os.getenv("CALLS_PER_USER", "10"))
RAMP_SECONDS = float(os.getenv("RAMP_SECONDS", "10"))
TOOL_TIMEOUT = float(os.getenv("TOOL_TIMEOUT", "5"))
AUTH_TOKEN = os.getenv("MCP_AUTH_TOKEN")

@dataclass
class Sample:
    latency_ms: float
    ok: bool
    error: str | None = None

def percentile(values: list[float], percent: int) -> float:
    if not values:
        return 0.0
    ordered = sorted(values)
    index = max(0, math.ceil(len(ordered) * percent / 100) - 1)
    return ordered[index]

def make_client() -> Client:
    auth = BearerAuth(AUTH_TOKEN) if AUTH_TOKEN else None
    return Client(MCP_URL, auth=auth)

async def call_once(client: Client, user_id: int, sequence: int) -> Sample:
    started = time.perf_counter()
    try:
        result = await client.call_tool(
            MCP_TOOL,
            MCP_ARGUMENTS,
            timeout=TOOL_TIMEOUT,
            raise_on_error=False,
            meta={
                "load_test": True,
                "virtual_user": user_id,
                "sequence": sequence,
            },
        )
        latency_ms = (time.perf_counter() - started) * 1000
        if result.is_error:
            return Sample(latency_ms, False, "tool_error")
        return Sample(latency_ms, True)
    except Exception as exc:
        latency_ms = (time.perf_counter() - started) * 1000
        return Sample(latency_ms, False, type(exc).__name__)

async def virtual_user(user_id: int) -> list[Sample]:
    delay = user_id * RAMP_SECONDS / max(USERS - 1, 1)
    await asyncio.sleep(delay)

    samples: list[Sample] = []
    async with make_client() as client:
        for sequence in range(CALLS_PER_USER):
            samples.append(await call_once(client, user_id, sequence))
    return samples

async def warm_up() -> None:
    async with make_client() as client:
        await client.call_tool(
            MCP_TOOL,
            MCP_ARGUMENTS,
            timeout=TOOL_TIMEOUT,
        )

async def main() -> None:
    if os.getenv("LOAD_TEST_CONFIRM") != "staging":
        raise SystemExit(
            "停止: LOAD_TEST_CONFIRM=staging を明示してください"
        )

    print(
        f"target={MCP_URL} tool={MCP_TOOL} "
        f"users={USERS} calls_per_user={CALLS_PER_USER}"
    )

    await warm_up()
    started = time.perf_counter()
    groups = await asyncio.gather(
        *(virtual_user(user_id) for user_id in range(USERS))
    )
    elapsed = time.perf_counter() - started
    samples = [sample for group in groups for sample in group]
    successful = [sample.latency_ms for sample in samples if sample.ok]
    failures = [sample for sample in samples if not sample.ok]

    total = len(samples)
    error_rate = len(failures) / total * 100 if total else 0.0
    throughput = total / elapsed if elapsed else 0.0
    errors = Counter(sample.error for sample in failures)

    print(f"requests={total} elapsed={elapsed:.2f}s")
    print(f"throughput={throughput:.2f} req/s")
    print(f"error_rate={error_rate:.2f}% errors={dict(errors)}")
    print(
        f"p50={percentile(successful, 50):.1f}ms "
        f"p95={percentile(successful, 95):.1f}ms "
        f"p99={percentile(successful, 99):.1f}ms"
    )

if __name__ == "__main__":
    asyncio.run(main())

FastMCPとテスト用スクリプトを同じ仮想環境へ入れます。

pip install fastmcp

最初は5ユーザー、1ユーザーあたり10回で実行します。

LOAD_TEST_CONFIRM=staging \
MCP_URL=http://127.0.0.1:8000/mcp \
MCP_TOOL=search \
LOAD_USERS=5 \
CALLS_PER_USER=10 \
RAMP_SECONDS=10 \
python load_test.py

認証付きサーバーでは、トークンをコードへ書かず環境変数で渡します。

MCP_AUTH_TOKEN="$YOUR_TEST_TOKEN" \
LOAD_TEST_CONFIRM=staging \
python load_test.py

テスト専用の権限が小さいアカウントを用意し、実行後にトークンを失効できるようにしてください。

負荷を段階的に増やす

いきなり100ユーザーで始めると、壊れた理由を切り分けにくくなります。次のように同じ条件で段階を分けます。

各段階で、ツール引数、呼び出し回数、ランプ時間を揃えます。前段階で停止条件を超えたら、それ以上の負荷へ進みません。

for users in 1 5 10 25 50; do
  LOAD_TEST_CONFIRM=staging \
  LOAD_USERS="$users" \
  CALLS_PER_USER=20 \
  RAMP_SECONDS=30 \
  python load_test.py
done

CIや共有環境でこのループを無条件実行しないでください。対象URLがステージングであることを別の仕組みでも検証し、実行権限を限定します。

「性能の膝」を見つける

同時接続を増やすと、最初はスループットも伸びます。しかし、CPU、データベース接続、外部API、ワーカースレッドなどが飽和すると、スループットが頭打ちになり、p95が急増します。

この変曲点が「性能の膝」です。たとえば80接続付近でp95が急増するなら、運用上限を80に設定するのではなく、予期しないスパイクに備えて50〜60程度へ下げます。

上限は次の式で考えられます。

安全な同時接続上限 = 性能の膝 × 0.6〜0.8

固定の正解ではありません。復旧時間、オートスケール速度、外部APIのレート制限を加味して余裕を決めます。

コールドスタートと定常性能を分ける

最初の1回だけ遅い場合は、次の処理が含まれている可能性があります。

• Pythonモジュールのimport
• モデルや設定ファイルの読み込み
• データベース接続プールの作成
• DNS解決とTLS接続
• キャッシュの初期化

記事のスクリプトでは、集計前に1回ウォームアップしています。コールドスタートも重要な要件なら、ウォームアップを除いたテストを別シナリオとして保存してください。

コールドスタートと定常性能を同じ集計へ混ぜると、改善箇所を誤りやすくなります。

OpenTelemetryでボトルネックを特定する

負荷テストの結果だけでは、「遅い」という事実しか分かりません。前の記事で設定したOpenTelemetryのTraceを重ねると、ツール内部のどこで待ち時間が増えたか確認できます。

見るべき箇所は次の通りです。

• tools/call全体のp95
• 外部APIやデータベースの子Span
• 接続プール待ち時間
• 再試行回数とタイムアウト
• エラー種別ごとの発生数
• テレメトリ送信自体の負荷

監視設定は「FastMCPをOpenTelemetryで監視｜ログ・トレース入門」で解説しています。

よくあるボトルネックと改善策

同期I/Oがイベントループを止めている

async defツール内で同期HTTPクライアントや重いファイルI/Oを直接実行すると、他の呼び出しが待たされます。非同期対応クライアントへ変更するか、ブロッキング処理を適切なスレッドへ逃がします。

データベース接続プールが小さい

同時ユーザーを増やしてもスループットが伸びず、DB Spanの待ち時間だけが増える場合は、接続プールが候補です。接続数を無制限に増やすのではなく、データベース側の上限とクエリ負荷を確認します。

外部APIのレート制限に達している

429エラーや再試行増加が見える場合、FastMCPサーバーを増やしても改善しません。キャッシュ、バッチ化、キュー、クライアント別レート制限を検討します。

レスポンスが大きすぎる

大きなJSONや画像を毎回返すと、シリアライズとネットワーク転送が支配的になります。ページング、要約、リソース参照、圧縮などで転送量を抑えます。

ログが多すぎる

全リクエストの入力・出力を同期的に書き込むと、ディスクやログ送信がボトルネックになります。構造化ログの項目とレベルを見直し、入力本文やトークンは記録しません。

Uvicornワーカーを増やす前に確認すること

FastMCPをASGIアプリとして公開すると、Uvicornの複数ワーカーを利用できます。

from fastmcp import FastMCP

mcp = FastMCP("My Server")

@mcp.tool
def process(data: str) -> str:
    return f"Processed: {data}"

app = mcp.http_app(stateless_http=True)

uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4

ただし、複数ワーカーにはstateless HTTP modeが必要です。通常のStreamable HTTPはサーバー側セッションをメモリに保持するため、同じクライアントの次のリクエストが別ワーカーへ届くと失敗する可能性があります。

elicitationやsamplingなど、セッション状態を必要とする機能を使っている場合は、stateless化の影響を先に確認してください。単に--workers 4を追加するだけでは安全に水平スケールできません。

デプロイの基本は「FastMCPをDockerでデプロイ｜本番向け構成と設定」も参照してください。

負荷テストでやってはいけないこと

• 所有していないサーバーへ許可なく負荷をかける
• 初回から本番環境で実施する
• 削除、送信、購入などのツールを対象にする
• 外部API費用とレート制限を確認せず実行する
• 実ユーザーと同じ強い権限のトークンを使う
• 停止条件を決めず上限まで増やす
• テスト実行元のCPU不足をサーバー限界と誤認する

負荷テストはサービスを意図的に不安定化させる可能性があります。対象、時間帯、上限、連絡先、停止方法を記録し、関係者の承認を取ってください。

結果を比較できる形で保存する

改善前後を比較するには、結果だけでなく条件も保存します。

実行日時: 2026-08-15 14:00 JST
対象バージョン: abc1234
環境: staging / 2 CPU / 4 GB
同時ユーザー: 25
呼び出し回数: 500
ランプ時間: 30秒
ツール: search
p50: 84 ms
p95: 241 ms
p99: 390 ms
エラー率: 0.2%
スループット: 31.4 req/s

コードのコミットID、FastMCP・Pythonのバージョン、ワーカー数、データベース構成も残します。条件が違う結果を並べても、改善効果を判断できません。

実運用前のチェックリスト

• □ 対象環境と実施時間について許可を得た
• □ ステージングURLであることを確認した
• □ 読み取り専用で再実行可能なツールを選んだ
• □ 外部APIの費用とレート制限を確認した
• □ p95、エラー率、CPUなどの停止条件を決めた
• □ 1ユーザーの基準測定から始めた
• □ 負荷を段階的に増やした
• □ OpenTelemetryでボトルネックを確認した
• □ 改善前後を同じ条件で比較した
• □ トークンと個人情報をログへ残していない

まとめ

FastMCPの負荷テストでは、MCPのライフサイクルとセッションを扱える公式Clientを使うと、実際の利用に近い測定ができます。

1ユーザーの基準測定から始め、同時接続を段階的に増やし、p95・エラー率・スループットを記録します。スループットが伸びなくなりp95が急増する「性能の膝」を見つけ、その60〜80%を安全な運用上限の目安にします。

結果をOpenTelemetryのTraceと組み合わせれば、FastMCP本体、ツール処理、データベース、外部APIのどこが詰まっているかを切り分けられます。

負荷テストの前に正常系・異常系を整えたい場合は、「FastMCPをpytestでテスト｜ツール・リソース・認証の実践例」から進めてください。

参考資料

• FastMCP公式：Calling Tools
• FastMCP公式：HTTP Deployment
• FastMCP公式：Client Transports
• Model Context Protocol：Transports