Qainsights Locust MCP Server

🚀 ⚡️ Locust-MCP-Server

このサーバーは、モデルコンテキストプロトコル（MCP）をサポートするLocust負荷テストサーバーの実装です。Locustの負荷テスト機能をAI駆動の開発環境とシームレスに統合します。

✨ 主な機能

• モデルコンテキストプロトコルフレームワークへの簡単な統合
• ヘッドレスモードとUIモードのサポート
• テストパラメータ（ユーザー数、起動率、実行時間）の設定可能
• Locust負荷テストを実行するための使いやすいAPI
• リアルタイムのテスト実行出力
• HTTP/HTTPSプロトコルのネイティブサポート
• カスタムタスクシナリオのサポート

🔧 前提条件

始める前に、以下をインストールしてください。

• Python 3.13以上
• uvパッケージマネージャー（インストールガイド）

📦 インストール

1. リポジトリをクローンします。

git clone https://github.com/yourusername/locust-mcp-server.git

2. 必要な依存関係をインストールします。

uv pip install -r requirements.txt

3. 環境変数を設定します（オプション）。プロジェクトのルートディレクトリに.envファイルを作成します。

LOCUST_HOST=http://localhost:8089  # テストのデフォルトホスト
LOCUST_USERS=3                     # デフォルトのユーザー数
LOCUST_SPAWN_RATE=1               # ユーザーの起動率のデフォルト値
LOCUST_RUN_TIME=10s               # デフォルトのテスト継続時間

🚀 クイックスタート

1. Locustテストスクリプトを作成します（例：hello.py）。

from locust import HttpUser, task, between

class QuickstartUser(HttpUser):
    wait_time = between(1, 5)

    @task
    def hello_world(self):
        self.client.get("/hello")
        self.client.get("/world")

    @task(3)
    def view_items(self):
        for item_id in range(10):
            self.client.get(f"/item?id={item_id}", name="/item")
            time.sleep(1)

    def on_start(self):
        self.client.post("/login", json={"username":"foo", "password":"bar"})

2. お好きなMCPクライアントでMCPサーバーを設定します（Claude Desktop、Cursor、Windsurfなど）。

{
  "mcpServers": {
    "locust": {
      "command": "/Users/naveenkumar/.local/bin/uv",
      "args": [
        "--directory",
        "/Users/naveenkumar/Gits/locust-mcp-server",
        "run",
        "locust_server.py"
      ]
    }
  }
}

3. これでLLMを使ってテストを実行できます。例えば、hello.pyを実行するためにLocustテストを実行と入力します。Locust MCPサーバーは以下のツールを使ってテストを開始します。

• run_locust：設定オプション付きのテスト実行（ヘッドレスモード、ホスト、継続時間、ユーザー数、起動率）

💻 使用例

基本的な使用法

from locust import HttpUser, task, between

class QuickstartUser(HttpUser):
    wait_time = between(1, 5)

    @task
    def hello_world(self):
        self.client.get("/hello")
        self.client.get("/world")

    @task(3)
    def view_items(self):
        for item_id in range(10):
            self.client.get(f"/item?id={item_id}", name="/item")
            time.sleep(1)

    def on_start(self):
        self.client.post("/login", json={"username":"foo", "password":"bar"})

高度な使用法

# 高度なシナリオでのLocustテストの設定
# 例えば、特定のユーザー数や起動率、実行時間を設定する場合
from locust import HttpUser, task, between

class AdvancedUser(HttpUser):
    wait_time = between(1, 5)

    @task
    def custom_task(self):
        # カスタムのタスクを定義
        pass

    def on_start(self):
        self.client.post("/login", json={"username":"advanced", "password":"user"})

📚 ドキュメント

実行Locustテスト

run_locust(
    test_file: str,
    headless: bool = True,
    host: str = "http://localhost:8089",
    runtime: str = "10s",
    users: int = 3,
    spawn_rate: int = 1
)

パラメータ：

• test_file：Locustテストスクリプトのパス
• headless：ヘッドレスモード（True）またはUI付き（False）
• host：負荷テストの対象ホスト
• runtime：テストの継続時間（例：“30s”、“1m”、“5m”）
• users：シミュレートする同時ユーザー数
• spawn_rate：ユーザーの起動率

🤝 貢献

貢献を歓迎します！自由にプルリクエストを送信してください。

📄 ライセンス

このプロジェクトはMITライセンスの下で提供されています。詳細については、LICENSEファイルを参照してください。