Druid MCP Server

Druid MCP

このリポジトリは、アドホックなデータ分析、監視、トラブルシューティング、比較のために、1 つ以上の Apache Druid クラスターに包括的な読み取り専用アクセスを提供する MCP（Model Context Protocol）サーバーです。

たとえば、それぞれに独自の Druid クラスターがあり、同じソースからデータを取り込んでいる複数の環境がある場合、次のようなシンプルなプロンプトを使って、複雑なクラスター間比較を実行できます。

すべての Druid クラスターで SumUp のデータソースを確認して比較してください。

AI エージェントは、Druid クラスターとそのデータソースを一覧表示し、スキーマ、セグメント、タスク、データ分布をすべて自動で調査して、最終的に分かりやすい概要を提示します。

概要

Druid MCP サーバーを使うと、AI アプリケーションは Model Context Protocol（MCP）を通じて Apache Druid とやり取りできます。Druid データソースのクエリと調査のための標準インターフェイスを提供し、Druid データを AI ワークフローに簡単に統合できます。

機能

• マルチクラスター対応：複数の Druid クラスターに同時に接続してクエリを実行
• クエリ実行：SQL およびネイティブ JSON クエリ
• データソース管理：データソースのスキーマを一覧表示、調査、確認
• 取り込み監視：スーパーバイザーとタスクのステータスを追跡
• クラスター操作：サービスの健全性を監視し、セグメントを分析
• ルックアップ管理：ルックアップテーブルをクエリして確認
• スマート分析プロンプト：一般的なデータ分析タスク向けの事前作成済みプロンプト
• スキーマキャッシュ：リソースのみを対象に、パフォーマンス向上のため 5 分の TTL で効率的にキャッシュ

前提条件

• Python 3.11+
• Apache Druid クラスターへのアクセス
• MCP 対応クライアント（例：Claude Desktop）

インストール

uv を使って依存関係をインストールします。

uv install

または pip を使います。

pip install -e .

設定

マルチクラスター対応

空白区切りの key=value ペアを使って、複数の Druid クラスターを設定します。

# Define available clusters (optional, defaults to "localhost=http://localhost:8088")
export DRUID_CLUSTER_URLS="localhost=http://localhost:8088 dev=https://druid.dev.example.com prod=https://druid.prod.example.com"

クラスター名には任意の有効な文字列を指定でき、次の用途で使用されます。

• リソース URI（例：druid://dev/v2/datasources）
• ツールのパラメーター（例：execute_sql_query("dev", "SELECT ...")）
• クラスター管理コマンド

使用方法

開発モード

開発用に自動リロードを有効にして実行します。

mcp dev main.py

本番モード

本番環境でサーバーを実行します。

mcp run main.py

Claude Desktop との連携

Claude Desktop で使用するためにサーバーをインストールします。

mcp install main.py

利用可能なツール

MCP サーバーは、機能別に整理された 16 個のツールを公開します。

重要：クラスター管理ツールを除き、すべてのツールで最初の引数として明示的な cluster パラメーターが必要です。これにより、状態管理なしで効率的なマルチクラスター操作が可能になります。

クエリ実行

• execute_sql_query(cluster, query, context)：SQL クエリを実行
• execute_native_query(cluster, query)：ネイティブ JSON クエリ（timeseries、topN、groupBy など）を実行

データソース操作

• list_datasources(cluster, include_details)：利用可能なすべてのデータソースを一覧表示
• get_datasource_schema(cluster, datasource)：データソースのディメンションとメトリクスを取得

取り込み監視

• list_supervisors(cluster, include_state)：オプションで状態情報を含め、すべてのスーパーバイザーを一覧表示
• get_supervisor_status(cluster, supervisor_id)：スーパーバイザーの詳細なステータスと健全性を取得
• list_tasks(cluster, ...)：フィルターオプションを使ってタスクを一覧表示
• get_task_status(cluster, task_id)：指定したタスクのステータスを取得

クラスター管理

• get_cluster_status(cluster)：クラスター全体の健全性ステータス
• list_services(cluster, service_type)：タイプ別にアクティブなサービスを一覧表示
• list_segments(cluster, datasource, full)：データソースのセグメントを一覧表示
• get_segments_info(cluster, datasource)：セグメントの集計統計を取得

ルックアップ管理

• list_lookups(cluster, tier)：ティア別にルックアップを一覧表示
• get_lookup(cluster, lookup_id, tier)：指定したルックアップの設定を取得
• get_lookup_status(cluster, lookup_id, tier)：ノード全体でのルックアップ読み込みステータスを取得

マルチクラスター管理

• list_clusters(): 設定済みのすべての Druid クラスター名を一覧表示します

分析プロンプト

スマートなデフォルト設定（先月から現在まで）を備えた、事前構築済みの 5 つのプロンプトです。

• analyze_time_range: 期間内のデータを分析します
• explore_datasource: データソースの構造と内容を調査します
• monitor_ingestion: 取り込みの健全性と進捗を監視します
• compare_periods: 期間ごとのメトリクスを比較します
• data_quality_check: 包括的なデータ品質チェックを実行します

例

基本的な使い方

# List datasources from production cluster
list_datasources("prod")

# Execute SQL query on development cluster
execute_sql_query("dev", "SELECT COUNT(*) FROM wikipedia WHERE __time >= CURRENT_TIMESTAMP - INTERVAL '1' DAY")

# Get datasource schema from staging
get_datasource_schema("stage", "wikipedia")

マルチクラスター操作

# Compare data across environments
dev_count = execute_sql_query("dev", "SELECT COUNT(*) FROM orders")
prod_count = execute_sql_query("prod", "SELECT COUNT(*) FROM orders")

# Check cluster health across all environments
for cluster in ["dev", "stage", "prod"]:
    status = get_cluster_status(cluster)
    print(f"{cluster}: {status}")

# Monitor ingestion across clusters
list_supervisors("dev", include_state=True)
list_supervisors("prod", include_state=True)

高度なクエリ

# Native query with explicit cluster
execute_native_query("prod", {
    "queryType": "timeseries",
    "dataSource": "wikipedia",
    "intervals": ["2024-01-01/2024-01-02"],
    "granularity": "hour",
    "aggregations": [
        {"type": "count", "name": "edits"},
        {"type": "longSum", "name": "added", "fieldName": "added"}
    ]
})

# Complex filtering with multiple parameters
list_tasks("prod", 
    datasource="wikipedia",
    state="running",
    max_tasks=10
)

リソース

Druid API エンドポイントは、キャッシュ機能とマルチクラスター対応を備えた MCP リソースとして公開されます。

リソース URI の形式

druid://{cluster}/{path}

利用可能なリソース

• druid://localhost/v2/datasources - すべてのデータソースを一覧表示します
• druid://localhost/v2/datasources/{name} - データソースのスキーマを取得します
• druid://localhost/coordinator/v1/datasources/{name} - Coordinator の情報
• druid://localhost/coordinator/v1/metadata/datasources/{name}/segments - セグメント
• druid://localhost/overlord/v1/tasks - タスクを一覧表示します
• druid://localhost/overlord/v1/supervisors - スーパーバイザーを一覧表示します
• druid://localhost/status/health - ヘルスチェック

異なる環境にアクセスするには、localhost を設定済みの任意のクラスター名に置き換えてください。

セキュリティ

この MCP サーバーは、Druid クラスターへの読み取り専用アクセスを提供します。書き込み操作はサポートされておらず、許可もされていません。

ライセンス

MIT