nifi-clickhouse-kafka-mcp

nifi-clickhouse-kafka-mcp

一款只读 MCP 服务，让 AI 助手查看 Apache NiFi、 ClickHouse 和 Kafka：一个服务，覆盖所有环境，而且不会造成任何破坏。

向 Claude、Cursor 或任何 MCP 客户端询问这类问题：

• “生产环境中的订单流为什么积压了？”→ nifi_health 会找出完整队列、已停止的处理器和错误公告，并按严重程度从高到低排列。
• “显示队列中卡住的 FlowFile，以及它的来源。”→ nifi_queue、nifi_flowfile、nifi_provenance。
• “账单消费者是否丢失了数据？”→ kafka_consumer_lag 会标记已落后于保留期限的分区。
• “查找过去一小时内发送的订单 42 消息。”→ 使用 kafka_search 按键、子串或 JSON 字段搜索。
• “哪些表包含客户数据，数据量有多大？”→ clickhouse_search_columns、clickhouse_describe_table。

这个项目源自一个网关，旨在让数据平台团队能够通过 AI 访问 NiFi 2、ClickHouse 和 Strimzi Kafka 集群，而无需提供凭据或写入权限。

工作原理

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

每个环境（生产、预发布等）都有自己的一组后端；一个服务即可为所有环境提供服务。

为什么选择它

	nifi-clickhouse-kafka-mcp
双重只读保护	只提供读取工具，且只会发送加入允许列表的请求。每个后端身份也都是只读的，因此一个锁出问题时，另一个可以兜底。
远程访问与多用户支持	使用来自任意 OIDC 提供商（Keycloak、Entra ID、Okta、Auth0、Dex）的 OAuth 令牌进行 Streamable HTTP 连接，可选用群组门禁，且笔记本电脑上无需存放凭据。也可使用纯 stdio 模式供个人使用。
可审计	每次调用都会记录一行 JSON，包含用户、工具、参数、结果和大小。ClickHouse 查询会将用户信息写入 system.query_log。还可选用 Prometheus 指标。
上下文安全	所有列表结果都支持分页。ClickHouse 会流式读取，并在达到行数或大小上限时停止，同时说明原因。全局结果上限可防止任何结果淹没模型。
屏蔽机密信息	工具结果中的密码、令牌、API 密钥，以及 JSON、YAML、NDJSON、URL 和 key=value 文本中的凭据都会被屏蔽，还可屏蔽你配置的任何字段名（如电子邮件、电话号码）。无法屏蔽的二进制载荷会被隐藏，除非你允许显示。
支持多个集群	通过一个连接器连接多个命名环境（生产、预发布等），每个环境都可包含任意组合的 NiFi、ClickHouse 和 Kafka。
不留痕迹	Kafka 读取操作会直接分配分区：不会加入消费者组，也不会提交偏移量。NiFi 队列列表和数据溯源查询在使用后会被删除。

请参阅 docs/landscape.md，了解它与其他 NiFi、ClickHouse 和 Kafka MCP 服务的比较，以及它解决了哪些用户反馈的问题。

快速开始

尝试连接本地 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 会验证配置，并列出它将提供的工具。

连接方式

个人使用，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。 只需运行一次（使用 Docker 或 Kubernetes），每个人都用自己的账户登录：

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

没有身份提供商？将服务置于你已有的代理之后（oauth2-proxy、Cloudflare Access、Tailscale 等），并将 trusted_user_header 设置为该代理添加的标头；用户身份仍会显示在审计日志中。

docs/deploying.md 介绍身份提供商、代理模式、只读后端身份、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：一次调用即可检查整个流中的背压连接、无效处理器、已停止的控制器服务、错误公告和已停止的处理器，并按严重程度从高到低排列。
• nifi_flowfile / nifi_provenance_event：查看真实 FlowFile 的属性和内容，JSON 会重新序列化并屏蔽机密，结果按偏移量分页。
• clickhouse_query：执行一条带绑定 {name:Type} 参数的只读语句，结果以流式传输，并按行数或大小上限截断，同时返回读取行数和耗时。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() 读取。如果出现写入路径，测试会导致构建失败。

后端是第二道锁：为网关配置一个仅拥有查看权限策略的 NiFi 身份、一个具有 readonly 和 SELECT 授权的 ClickHouse 用户，以及一个具有 Read/Describe ACL 的 Kafka 用户。docs/deploying.md 提供了各项配置的具体步骤。

有两点值得注意：

• 网关使用自己的身份，而不是用户的身份。 NiFi 2 仅根据最终用户的权限授权代理请求，因此代理用户请求的网关可能会让用户写入其有权访问的任何位置。为审计而保留每个用户的身份信息。
• 屏蔽是纵深防御措施，而非安全边界。 它会屏蔽结果中的机密和已配置字段，并让搜索（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 chart
• 根据令牌声明映射每用户 ClickHouse 角色

许可证

Apache 2.0