使用说明

mcp卡夫卡

Apache Kafka的开源MCP（模型上下文协议）服务器，使用Quarkus构建。使Claude、Goose和Gemini等AI助手能够与Kafka集群和Schema Registry进行交互。

设计为可直接替代 @confluentinc/mcp-confluent，针对开源Apache Kafka而不是Confluent Cloud。

特性

22工具 -Kafka主题管理、生成/消费、模式注册表CRUD、Flink SQL网关、主题搜索、集群信息
4资源 -只读集群状态数据（概述、健康检查、未复制分区、消费者延迟）
4个提示 -用于群集运行状况和故障排除的预构建诊断报告模板
模式注册表 -注册、列出、获取、删除模式；使用Avro序列化（汇流线格式）生产/消费
双重运输 -STDIO（子进程）和HTTP/SSE（远程服务器）
插入兼容 -与MCP汇流相同的MCP工具名称和响应格式
Quarkus供电 -快速启动，低内存，GraalVM原生映像就绪
可配置的 -环境变量、属性文件、工具允许/阻止列表
测试容器 -使用真实Kafka和Schema Registry进行集成测试

快速开始

先决条件

Java 21+
Docker（用于Kafka或运行测试）

从源代码构建

git clone https://github.com/gamussa/mcp-kafka.git
cd mcp-kafka
./gradlew quarkusBuild

内置应用程序位于 build/quarkus-app/.

使用Docker Compose运行

使用一个命令启动整个堆栈：

docker compose up

这将开始：

卡夫卡 上 localhost:9092
模式注册表 上 localhost:8081
Flink SQL网关 上 localhost:8083
MCP服务器 上 localhost:8080 （HTTP/SSE传输）

所有服务都经过健康检查——MCP服务器在启动之前等待Kafka、Schema Registry和Flink SQL Gateway准备就绪。所有22个工具均已启用。

将MCP客户端连接到 http://localhost:8080/mcp/sse 或者使用下面的配置进行STDIO模式。

MCP客户端配置

克劳德桌面（STDIO）

添加到您的 claude_desktop_config.json:

{
  "mcpServers": {
    "kafka": {
      "command": "java",
      "args": ["-Dquarkus.profile=stdio", "-jar", "/path/to/mcp-kafka/build/quarkus-app/quarkus-run.jar"],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "localhost:9092"
      }
    }
  }
}

克劳德代码CLI

添加一个 .mcp.json 将文件保存到项目根目录：

{
  "mcpServers": {
    "kafka": {
      "command": "java",
      "args": ["-Dquarkus.profile=stdio", "-jar", "/path/to/mcp-kafka/build/quarkus-app/quarkus-run.jar"],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "localhost:9092"
      }
    }
  }
}

或者通过Claude Code设置进行全局配置。

使用架构注册表

将架构注册表URL添加到环境中：

{
  "mcpServers": {
    "kafka": {
      "command": "java",
      "args": ["-Dquarkus.profile=stdio", "-jar", "/path/to/mcp-kafka/build/quarkus-app/quarkus-run.jar"],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "localhost:9092",
        "SCHEMA_REGISTRY_URL": "http://localhost:8081"
      }
    }
  }
}

架构注册表工具在以下情况下会自动启用 SCHEMA_REGISTRY_URL 设置，否则禁用。

使用Flink SQL网关

将Flink SQL网关URL添加到环境中：

{
  "mcpServers": {
    "kafka": {
      "command": "java",
      "args": ["-Dquarkus.profile=stdio", "-jar", "/path/to/mcp-kafka/build/quarkus-app/quarkus-run.jar"],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "localhost:9092",
        "FLINK_SQL_GATEWAY_URL": "http://localhost:8083"
      }
    }
  }
}

Flink SQL网关工具在以下情况下会自动启用 FLINK_SQL_GATEWAY_URL 设置，否则禁用。

鹅/其他MCP客户端（STDIO）

任何支持STDIO传输的MCP客户端都可以使用：

command: java -Dquarkus.profile=stdio -jar /path/to/mcp-kafka/build/quarkus-app/quarkus-run.jar
env:
  KAFKA_BOOTSTRAP_SERVERS: localhost:9092

HTTP/SSE传输（远程服务器）

对于远程部署，直接运行服务器（HTTP是默认传输）：

java -Dquarkus.http.host=0.0.0.0 \
     -jar build/quarkus-app/quarkus-run.jar

这暴露了：

流式HTTP 在 http://host:8080/mcp
传统SSE 在 http://host:8080/mcp/sse

配置MCP客户端以连接到远程URL：

{
  "mcpServers": {
    "kafka": {
      "url": "http://your-server:8080/mcp/sse"
    }
  }
}

注： HTTP/SSE模式还不包括身份验证。绑定到 127.0.0.1 （默认）或使用反向代理进行远程访问。

配置参考

Kafka连接

属性	环境变量	默认值	描述
`kafka.bootstrap-servers`	`KAFKA_BOOTSTRAP_SERVERS`	`localhost:9092`	Kafka代理地址
`kafka.security-protocol`	`KAFKA_SECURITY_PROTOCOL`	`PLAINTEXT`	`PLAINTEXT`, `SASL_PLAINTEXT`, `SSL`, `SASL_SSL`
`kafka.sasl-mechanism`	`KAFKA_SASL_MECHANISM`	（无）	`PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`
`kafka.sasl-username`	`KAFKA_SASL_USERNAME`	（无）	SASL用户名
`kafka.sasl-password`	`KAFKA_SASL_PASSWORD`	（无）	SASL密码
`kafka.ssl-truststore-location`	`KAFKA_SSL_TRUSTSTORE_LOCATION`	（无）	通往信任库的路径
`kafka.ssl-truststore-password`	`KAFKA_SSL_TRUSTSTORE_PASSWORD`	（无）	信任库密码
`kafka.ssl-keystore-location`	`KAFKA_SSL_KEYSTORE_LOCATION`	（无）	密钥库路径（mTLS）
`kafka.ssl-keystore-password`	`KAFKA_SSL_KEYSTORE_PASSWORD`	（无）	密钥库密码
`kafka.properties-file`	`KAFKA_PROPERTIES_FILE`	（无）	额外Kafka属性的路径

模式注册表

属性	环境变量	默认值	描述
`schema-registry.url`	`SCHEMA_REGISTRY_URL`	（无）	架构注册表URL（启用SR工具）
`schema-registry.auth.username`	`SCHEMA_REGISTRY_AUTH_USERNAME`	（无）	基本身份验证用户名
`schema-registry.auth.password`	`SCHEMA_REGISTRY_AUTH_PASSWORD`	（无）	基本身份验证密码

Flink SQL网关

属性	环境变量	默认值	描述
`flink-sql-gateway.url`	`FLINK_SQL_GATEWAY_URL`	（无）	Flink SQL网关URL（启用Flink工具）
`flink-sql-gateway.default-catalog`	`FLINK_SQL_GATEWAY_DEFAULT_CATALOG`	（无）	新会话的默认目录
`flink-sql-gateway.default-database`	`FLINK_SQL_GATEWAY_DEFAULT_DATABASE`	（无）	新会话的默认数据库
`flink-sql-gateway.statement-timeout`	`FLINK_SQL_GATEWAY_STATEMENT_TIMEOUT`	`30000`	语句完成的最大等待时间（ms）
`flink-sql-gateway.max-rows`	`FLINK_SQL_GATEWAY_MAX_ROWS`	`100`	每条语句的默认最大结果行数

运输

属性	环境变量	默认值	描述
`quarkus.profile`	`QUARKUS_PROFILE`	（无）	设置为 `stdio` 用于STDIO传输（MCP客户端）
`quarkus.http.host`	`QUARKUS_HTTP_HOST`	`127.0.0.1`	HTTP绑定地址（使用 `0.0.0.0` 远程）
`quarkus.http.port`	`QUARKUS_HTTP_PORT`	`8080`	HTTP服务器端口

工具筛选

属性	环境变量	默认值	描述
`mcp.tools-allow`	`MCP_TOOLS_ALLOW`	（无）	逗号分隔的工具允许列表
`mcp.tools-block`	`MCP_TOOLS_BLOCK`	（无）	逗号分隔的工具块列表
`mcp.tools-allow-file`	`MCP_TOOLS_ALLOW_FILE`	（无）	带有允许列表的文件（每行一个）
`mcp.tools-block-file`	`MCP_TOOLS_BLOCK_FILE`	（无）	带阻止列表的文件（每行一个）

可用工具

Kafka工具

工具	说明
`list-topics`	列出Kafka集群中的所有主题
`create-topics`	创建一个或多个Kafka主题
`delete-topics`	删除一个或多个Kafka主题
`produce-message`	生成消息（原始或模式注册表序列化）
`consume-messages`	消费消息（原始或架构注册表反序列化）
`get-topic-config`	获取Kafka主题的配置
`alter-topic-config`	更改Kafka主题的配置
`search-topics-by-name`	使用正则表达式或子字符串按名称搜索主题
`describe-cluster`	获取Kafka集群元数据

架构注册表工具

在以下情况下自动启用 schema-registry.url 已配置。

工具	说明
`list-schemas`	列出所有具有可选前缀过滤器的架构
`register-schema`	注册新模式（Avro、JSON模式、Protobuf）
`get-schema`	按主题和可选版本获取架构
`delete-schema`	软/硬删除主题或特定版本
`get-schema-compatibility`	获取主题或全局的兼容级别
`set-schema-compatibility`	设置兼容级别（后退、前进、完全等）

Flink SQL网关工具

在以下情况下自动启用 flink-sql-gateway.url 已配置。

工具	说明
`execute-flink-sql`	执行任意Flink SQL并返回结果
`list-flink-catalogs`	列出所有可用的Flink目录
`list-flink-databases`	在Flink目录中列出数据库
`list-flink-tables`	列出Flink数据库中的表
`describe-flink-table`	描述Flink表的模式
`get-flink-job-status`	获取Flink SQL操作的状态
`cancel-flink-job`	取消正在运行的Flink SQL操作

可用资源

MCP资源将只读Kafka集群状态作为结构化JSON数据公开。配置Kafka时，资源会自动启用。

资源URI	描述
`kafka-mcp://overview`	群集运行状况摘要：代理计数、控制器ID、主题/分区计数、运行状况
`kafka-mcp://health-check`	详细的健康评估：代理、控制器、分区和消费者组状态
`kafka-mcp://under-replicated-partitions`	分析ISR<复制因子的分区，并提出故障排除建议
`kafka-mcp://consumer-lag-report`	消费者群体滞后指标，包括群体摘要、高滞后细节和建议

可用提示

MCP提示提供预构建的诊断报告模板，可生成人类可读的标记。配置Kafka时会自动启用提示。

提示	参数	描述
`kafka-cluster-overview`	（无）	人类可读的集群摘要，包括代理计数、主题、分区和运行状况
`kafka-health-check`	（无）	包含每个子系统状态指示器的综合健康检查报告
`kafka-under-replicated-partitions`	（无）	在包含受影响分区表、原因和建议的复制报告下
`kafka-consumer-lag-report`	`threshold` （可选，默认值： `1000`)	消费者滞后分析，包括组汇总表、高滞后细节和建议

架构注册表序列化

这 produce-message 和 consume-messages 工具支持可选的架构注册表集成：

produce-message with useSchemaRegistry=true, schemaType=AVRO, schema=<avro-json>

消息使用Confluent有线格式（魔术字节+4字节模式ID+有效载荷）进行序列化/反序列化，与任何Confluent生态系统消费者/生产者兼容。

发展

构建

./gradlew quarkusBuild

在开发模式下运行

./gradlew quarkusDev

开发模式功能：

源代码更改时的热重新加载
开发用户界面 http://localhost:8080/q/dev-ui
详细的工具和MCP框架日志记录

运行测试

测试使用Testcontainers apache/kafka-native 和 cp-schema-registry 图像。Docker必须正在运行。

./gradlew test

项目结构

src/main/java/com/github/imcf/mcp/kafka/
  config/          # KafkaConfig, SchemaRegistryConfig, ToolFilter, StartupLogger
  client/          # KafkaClientManager, SchemaRegistryClient
  model/           # ClusterOverview, ConsumerLagReport, UnderReplicatedPartitionReport, HealthReport
  serde/           # SchemaRegistrySerializer, SchemaRegistryDeserializer
  tools/
    BaseToolHandler.java
    kafka/         # Topic tools, produce/consume, search, topic config
    cluster/       # Cluster tools (describe-cluster)
    schema/        # Schema Registry tools (list, register, get, delete, compatibility)
    flink/         # Flink SQL Gateway tools (execute-sql, catalogs, databases, tables)
  resources/
    BaseResourceHandler.java
    kafka/         # MCP resources (overview, health-check, under-replicated, consumer-lag)
  prompts/
    BasePromptHandler.java
    kafka/         # MCP prompts (cluster-overview, health-check, under-replicated, consumer-lag)

路线图

阶段	范围	状态
第一阶段	核心Kafka工具，STDIO+HTTP/SSE传输，Docker	完成
第2阶段	架构注册表工具、SR序列化、主题搜索	完成
第三期	Flink SQL网关工具	完成
阶段4	MCP资源和集群可观测性提示	完成
阶段5	Kafka Connect工具	计划中
第6阶段	消费者群体、API密钥授权、GraalVM本机构建	计划

许可证

看许可证了解详情。