Kafka Dataops Mcp

io.github.Aguantar由 GitHub 账号 Aguantar 发布v0.1.0更新于 Oct 24, 2026

DataOps Kafka MCP server with consumer lag diagnosis and broker monitoring

已验证STDIO仅桌面数据库安全与监控

概览

AI 生成的概览

一个本地运行的 Kafka DataOps MCP 服务器，让助手查看消费组延迟、主题和 Broker 健康状态，并给出基于真实事故的诊断提示。

可以做什么
检查消费组：读取延迟，并针对停滞的管道给出事故模式诊断提示，包括 Flink 相关故障模式和 checkpoint 警告。查看主题：获取主题详情并做 ISR 与副本健康检查，列出带内置描述的主题目录。检查集群：报告 Broker、控制器和副本不足的分区。
适用场景
适合在运维 Kafka 或 CDC 管道时使用，让助手在故障排查和日常巡检中协助分析消费延迟、副本问题和 Broker 健康状况。
运行要求
以 Python 包形式通过 pip 安装并在本地以 stdio 运行。需要能访问 Kafka 集群的网络，通过环境变量 KAFKA_BOOTSTRAP_SERVERS 配置（默认 localhost:9092）；KAFKA_COMMAND_TIMEOUT 设置命令超时秒数。
安装前请注意
它会连接你的 Kafka 集群并读取消费组、主题和 Broker 状态等运维元数据，请只指向你有权检查的集群。未声明任何认证方式，访问控制取决于集群和网络配置。

安装

在 SourceWeft 中

  1. 打开 控制台中的 Kafka Dataops Mcp，将其添加到工作区。
  2. 为需要使用其工具的对话启用该服务。

Desktop only，通过 STDIO。 STDIO 服务会启动本地进程，因此需要 SourceWeft 桌面宿主。

其他 MCP 客户端

参照 仓库 中的启动说明。

README

kafka-dataops-mcp

mcp-name: io.github.Aguantar/kafka-dataops-mcp

A DataOps-focused Kafka MCP server with consumer lag diagnosis and broker health monitoring. Diagnosis logic is based on actual CDC pipeline operational experience.

Features

  • kafka_consumer_lag — Consumer group lag with incident-pattern diagnosis
  • kafka_topic_info — Topic details with ISR/replication health checks
  • kafka_broker_status — Cluster health: brokers, controller, under-replicated partitions
  • kafka_list_topics — Topic catalog with built-in descriptions

Diagnosis based on real incidents

The diagnosis logic is not generic — it's based on actual operational experience:

  • Flink crash detection: "no active members" + growing lag = likely Flink Job failure (based on a 46-hour outage caused by MySQL DELETE → Debezium tombstone → Flink NPE)
  • Checkpoint vs consumer group: warns that Kafka consumer group reset alone is insufficient for Flink — checkpoints must be deleted first
  • ClusterIdMismatch: detects missing brokers and suggests Docker volume conflict as root cause
  • ISR monitoring: ISR < min.insync.replicas = write failures (critical)

Installation

bash
pip install kafka-dataops-mcp

Usage with Claude Code

Add to your .mcp.json:

json
{  "mcpServers": {    "kafka": {      "command": "kafka-dataops-mcp",      "env": {        "KAFKA_BOOTSTRAP_SERVERS": "localhost:9092"      }    }  }}

Environment Variables

VariableDefaultDescription
KAFKA_BOOTSTRAP_SERVERSlocalhost:9092Kafka bootstrap servers
KAFKA_COMMAND_TIMEOUT10Command timeout in seconds

License

MIT

来源：README.md，提交 6b30d7a

工具

0
SourceWeft 尚未在沙箱中启动过该软件包，因此其工具未知。

版本历史

1
  1. v0.1.0最新Sep 16, 2026

评分

已满足适用检查项的 50%

该评分衡量此收录条目的完整程度，依据注册中心条目、端点与公开来源计算得出。它不是对服务器本身的评价，也不表示服务器安全或当前可用：可连通性是带日期的检查结果，评分也不包含安装量或用户评价。

  • 近 30 天内可连通25 分

    请保持端点能够响应 MCP 握手；SourceWeft 每 30 天探测一次。

  • 已声明认证方式10 分

    请在注册中心条目中声明凭据请求头，或保持端点能够响应探测，以便判定其认证方式。

  • 已列出工具15 分

    请让服务器在无凭据时响应 tools/list，或发布可在沙箱中无凭据启动的软件包。

  • 描述必需15 分
  • 图标10 分
  • 文档15 分
  • 许可证10 分