nifi-clickhouse-kafka-mcp

nifi-clickhouse-kafka-mcp

AI アシスタントが Apache NiFi、 ClickHouse、Kafka の内部を確認できる、読み取り専用の MCP サーバーです。サーバーは 1 つで、あらゆる環境に対応し、何かを壊すことはありません。

Claude、Cursor、その他の MCP クライアントに、次のようなことを尋ねられます。

• 「本番環境の orders フローが詰まっているのはなぜですか？」→ nifi_health が、キュー全体、停止したプロセッサー、エラーブリテンを、問題の深刻な順に特定します。
• 「そのキューで滞留している FlowFile と、その発生元を表示してください。」→ nifi_queue、nifi_flowfile、nifi_provenance。
• 「billing コンシューマーでデータが失われていますか？」→ kafka_consumer_lag が、すでに保持期間を超えて遅れているパーティションを検出します。
• 「直近 1 時間に送信された order 42 のメッセージを探してください。」→ キー、部分文字列、または JSON フィールドを指定して kafka_search。
• 「顧客データがあるテーブルと、そのサイズを教えてください。」→ clickhouse_search_columns、clickhouse_describe_table。

データプラットフォームチームが、認証情報や書き込みアクセスを渡さずに NiFi 2、ClickHouse、Strimzi Kafka クラスターへ AI がアクセスできるようにするゲートウェイから発展しました。

仕組み

flowchart LR
    clients["MCP clients<br/>Claude Code, Claude Desktop,<br/>Cursor, VS Code, ..."]
    idp["OIDC provider<br/>Keycloak, Entra ID, Okta, ..."]
    gw["<b>nifi-clickhouse-kafka-mcp</b><br/>lock 1<br/>───<br/>token + group check<br/>read-only tools<br/>request allowlist<br/>SQL parse check<br/>masking, size caps<br/>audit, metrics"]
    subgraph backends["Lock 2: read-only identities"]
        nifi["NiFi<br/>view-only policies"]
        ch["ClickHouse<br/>readonly user, SELECT grants"]
        kafka["Kafka<br/>Read/Describe ACLs"]
    end
    clients -- "stdio, or HTTP<br/>+ bearer token" --> gw
    clients -. "log in" .-> idp
    gw -. "verify token" .-> idp
    gw --> nifi
    gw --> ch
    gw --> kafka

各環境（prod、staging など）にはそれぞれ独自のバックエンドがあり、1 つのサーバーですべての環境に対応します。

このサーバーを選ぶ理由

	nifi-clickhouse-kafka-mcp
二重の読み取り専用	読み取りツールのみが存在し、許可リストに登録されたリクエストだけが送信されます。各バックエンドの ID も読み取り専用なので、一方のロックに不具合があっても、もう一方で検出できます。
リモート対応、複数ユーザー対応	任意の OIDC プロバイダー（Keycloak、Entra ID、Okta、Auth0、Dex）の OAuth トークンを使う Streamable HTTP、任意のグループ制限に対応し、ノート PC に認証情報を保存する必要はありません。または、1 人で使う場合は通常の stdio を利用できます。
監査可能	呼び出しごとに、ユーザー、ツール、引数、結果、サイズを 1 行の JSON で記録します。ClickHouse クエリでは、ユーザー情報が system.query_log に引き継がれます。Prometheus メトリクスはオプションです。
コンテキストに配慮	すべてのリストはページ分割されます。ClickHouse はストリーミングし、行数またはサイズの上限で停止して、その旨を通知します。結果全体の上限によって、モデルに大量のデータが流れ込むことを防ぎます。
シークレットをマスキング	JSON、YAML、NDJSON、URL、key=value テキストに含まれるパスワード、トークン、API キー、認証情報に加え、設定したフィールド名（メールアドレス、電話番号など）もツールの結果でマスキングされます。マスキングできないバイナリペイロードは、許可しない限り表示されません。
複数クラスター対応	名前付き環境（prod、staging など）ごとに、NiFi、ClickHouse、Kafka を任意に組み合わせて、1 つのコネクターで利用できます。
痕跡を残さない	Kafka の読み取りではパーティションを直接割り当てるため、コンシューマーグループに参加せず、オフセットもコミットしません。NiFi のキュー一覧とプロベナンスクエリは、使用後に削除されます。

他の NiFi、ClickHouse、Kafka MCP サーバーとの比較や、それらのユーザーから寄せられたどのような不満に応えるのかについては、docs/landscape.md をご覧ください。

クイックスタート

ローカルの NiFi 2、ClickHouse、Kafka に接続して試してみましょう。

git clone https://github.com/arogan178/nifi-clickhouse-kafka-mcp && cd nifi-clickhouse-kafka-mcp/examples/demo
docker compose up -d          # NiFi takes about a minute
uv run --with requests python seed_nifi.py
claude mcp add data -- uvx nifi-clickhouse-kafka-mcp --config "$PWD/config.yaml"

次に、Claude Code に「デモ用の NiFi フローで何が問題ですか？」と尋ねてください。examples/demo/README.md には、Kafka へのデータ投入を含む手順全体が記載されています。

独自のシステムで使うには、設定ファイルを作成してください（examples/config.example.yaml を参照）。

masked_fields: [email, phone]
environments:
  prod:
    nifi:
      url: https://nifi.example.com:8443
      auth: {type: mtls, cert: /certs/mcp.pem, key: /certs/mcp.key}
      ca_bundle: /certs/ca.pem
    clickhouse:
      url: https://clickhouse.example.com:8443
      user: mcp_reader
      password: ${CLICKHOUSE_PASSWORD}
    kafka:
      config:
        bootstrap.servers: kafka.example.com:9093
        security.protocol: SASL_SSL
        sasl.mechanism: SCRAM-SHA-512
        sasl.username: mcp-reader
        sasl.password: ${KAFKA_PASSWORD}
      schema_registry: {url: https://registry.example.com}

${VAR} と ${VAR:-default} では環境変数を読み込むため、シークレットをファイルに記載せずに済みます。 nifi-clickhouse-kafka-mcp --config config.yaml --check を実行すると、設定を検証し、提供されるツールを一覧表示します。

接続

1 人で使う場合は stdio。 サーバーは、お使いのマシン上で設定ファイルを使って実行します。

claude mcp add data -- uvx nifi-clickhouse-kafka-mcp --config /path/to/config.yaml

{ "mcpServers": { "data": { "command": "uvx", "args": ["nifi-clickhouse-kafka-mcp", "--config", "/path/to/config.yaml"] } } }

チームで使う場合は、OAuth を使った HTTP。 サーバーを 1 回起動（Docker または Kubernetes）すれば、全員が自分のアカウントでログインできます。

claude mcp add --transport http --client-id nifi-clickhouse-kafka-mcp --callback-port 47130 data https://mcp.example.com/mcp

ID プロバイダーがない場合は、すでに利用しているプロキシ（oauth2-proxy、Cloudflare Access、Tailscale など）の背後にサーバーを配置し、そのプロキシが追加するヘッダーを trusted_user_header に設定してください。監査ログには引き続きユーザーが記録されます。

docs/deploying.md では、ID プロバイダー、プロキシモード、読み取り専用のバックエンド ID、Docker、Kubernetes、監査、メトリクスについて説明しています。

ツール

システム	ツール
すべて	environments
NiFi	nifi_health, nifi_status, nifi_flow_tree, nifi_search, nifi_process_group, nifi_component, nifi_processor_state, nifi_controller_services, nifi_bulletins, nifi_parameters, nifi_parameter, nifi_queue, nifi_flowfile, nifi_provenance, nifi_provenance_event
ClickHouse	clickhouse_query, clickhouse_databases, clickhouse_tables, clickhouse_describe_table, clickhouse_search_columns, clickhouse_health
Kafka	kafka_cluster, kafka_topics, kafka_describe_topic, kafka_consumer_groups, kafka_consumer_lag, kafka_sample, kafka_search

ツールは、設定したシステムに対してのみ表示されます。disabled_tools を使って無効にできます（kafka_search や nifi_flow* などのパターンを指定できます）。

主な機能：

• nifi_health：バックプレッシャー状態の接続、無効なプロセッサー、停止中のコントローラーサービス、エラーバレットン、停止中のプロセッサーを、問題の深刻度順に、フロー全体について1回の呼び出しで確認できます。
• nifi_flowfile / nifi_provenance_event：実際の FlowFile の属性とコンテンツを、シークレットをマスクして JSON として再シリアライズし、オフセット単位でページ分割して取得できます。
• clickhouse_query：バインドされた {name:Type} パラメーターを使う読み取りステートメントを1つ実行します。結果はストリーミングされ、行数またはサイズの上限で切り詰められます。読み取った行数と経過時間も返します。ClickHouse 独自のパーサーが各ステートメントを事前に検証し、ClickHouse 自身を読み取るテーブル関数（numbers、merge、clusterAllReplicas など）のみ実行できます。url、s3、file、remote などは拒否されます。
• clickhouse_health：問題のあるレプリカ、ディスク使用量、パーツ数が最も多いパーティション、長時間実行中のクエリ、マージ、停滞しているミューテーション、最近のエラーを確認できます。
• kafka_search：時間範囲内のすべてのパーティションを走査し、キー、部分文字列、または値と一致する JSON フィールドを検索します。走査と時間の予算が設定されており、続行には resume トークンを使います。
• kafka_sample：オフセット、タイムスタンプ、または末尾からメッセージを取得し、キー、値、ヘッダーをデコード（JSON、テキスト、Schema Registry Avro、JSON Schema）してマスクします。

セキュリティモデル

ゲートウェイが最初の防護線です。読み取り専用ツールのみを登録し、各ツールには readOnlyHint の注釈が付与されます。NiFi クライアントが送信できるのは GET と、NiFi が必要とする一時的なキュー一覧取得およびプロベナンスリクエストのみです。また、別のエンドポイントに到達する ID は拒否します。ClickHouse の各ステートメントは実行前に ClickHouse（EXPLAIN AST）で解析されます。許可されるのは SELECT、WITH、SHOW、DESCRIBE、EXISTS、EXPLAIN のみです。テーブル関数は、どこに現れても許可リストに含まれている必要があります。ネットワークやディスクにアクセスする一部のスカラー関数（hasColumnInTable、file など）も拒否されます。Kafka の読み取りには常に assign() のみを使います。書き込み経路が見つかった場合にビルドが失敗するよう、テストで検証しています。

バックエンドが2番目の防護線です。ゲートウェイには、表示のみのポリシーを設定した NiFi ID、readonly と SELECT 権限を持つ ClickHouse ユーザー、Read/Describe ACL を持つ Kafka ユーザーを割り当ててください。それぞれの正確な設定方法は docs/deploying.md を参照してください。

知っておくべき2つの点：

• ゲートウェイはユーザーの ID ではなく、独自の ID を使用します。 NiFi 2 は、プロキシされたリクエストをエンドユーザー自身の権限のみで認可するため、ユーザーのリクエストをプロキシするゲートウェイは、そのユーザーが書き込める場所ならどこにでも書き込めてしまいます。その代わり、監査用にはユーザーごとの ID を保持します。
• マスキングは多層防御であり、セキュリティ境界ではありません。 結果内のシークレットや設定済みフィールドを隠し、検索（NiFi と Kafka）ではマスクされた値を照合するため、隠された値の有無を確認する用途には使えません。ただし、マスキングの対象は認識可能な形式に限られ、ClickHouse クエリではカラムに別名を付けることもできます。そのため、バックエンドの権限で機密データを確実に保護してください。NiFi のポリシー、ClickHouse のカラム権限と行ポリシー、Kafka のトピック ACL を設定してください。

脆弱性は SECURITY.md の説明に従って報告してください。

開発

uv sync
uv run pytest                                             # unit tests
docker compose -f examples/demo/compose.yaml up -d        # then seed, see examples/demo/README.md
NIFI_CLICKHOUSE_KAFKA_MCP_DEMO=1 uv run pytest tests/integration      # end to end against real NiFi, ClickHouse, Kafka

ロードマップ

• Kafka の Protobuf デコード
• NiFi プロベナンスのリネージグラフ
• Helm チャート
• トークンクレームからマッピングするユーザーごとの ClickHouse ロール

ライセンス

Apache 2.0