AI导哈

Kafka MCP服务器

一个基于Spring Boot和Spring AI构建的Kafka MCP服务器，通过SSE传输将Kafka集群操作暴露为AI客户端可用的工具。

  • by zakariahere
  • Java
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 zakariahere/kafka-mcp-server-sbai

使用说明

卡夫卡mcp sbai

使用Spring Boot和Spring AI构建的Kafka MCP（模型上下文协议）服务器。通过SSE传输，将Kafka集群操作作为AI客户端可通过MCP协议使用的工具公开。

技术栈

组件 版本
Java 25
弹簧靴 4.0.2
春季AI 2.0.0-M2
Apache Kafka spring-boot-starter-kafka
构建 Maven（包括包装脚本）

先决条件

  • Java 25+
  • Docker（用于本地Kafka代理）

快速开始

# Build
mvnw.cmd clean package -DskipTests          # Windows
./mvnw clean package -DskipTests            # Unix

# Run
mvnw.cmd spring-boot:run                    # Windows
./mvnw spring-boot:run                      # Unix

Spring Boot的Docker Compose集成会自动启动Kafka代理、模式注册表和中定义的Kafka UI compose.yaml 当应用程序启动时。无手册 docker compose up 需要。

MCP服务器启动于 端口9085.

本地基础设施（组成.yaml）

服务 端口 描述
Kafka（KRaft） 9092 单节点代理，自动创建主题
架构注册表 8081 冲突架构注册表
Kafka UI 8080 用于集群管理的Provectus Kafka UI

暴露的MCP工具

话题管理

工具 说明
listTopics 列出集群中的所有主题
describeTopic 返回主题的分区布局、副本、ISR和非默认配置
createTopic 创建具有指定分区计数和复制因子的主题
deleteTopic 永久删除主题及其所有消息

消息传递

工具 参数 说明
produceMessage 主题、信息、关键？头？ 发布消息；返回分区和偏移量
consumeMessages 主题，maxMessages？从一开始？，超时秒？ 通过短期消费者组读取消息
peekMessages 主题、分区、偏移、计数？ 从特定分区/偏移量读取而不提交

消费者群体

工具 说明
listConsumerGroups 列出所有消费者组ID
describeConsumerGroup 返回组状态、协调器、成员和分区分配

簇

工具 说明
describeCluster 返回集群ID、控制器和所有代理详细信息

建筑

AI Client  ──SSE──>  MCP Server (port 9085)
                          │
                    ┌─────▼──────┐
                    │ToolProvider │   @Tool-annotated methods, auto-registered
                    └─────┬──────┘      by Spring AI's MethodToolCallbackProvider
                          │
                    ┌─────▼──────┐
                    │KafkaService │   Wraps AdminClient, KafkaTemplate,
                    └─────┬──────┘   and ad-hoc KafkaConsumer instances
                          │
                    ┌─────▼──────┐
                    │   Kafka     │   Local broker via Docker Compose
                    └────────────┘
  • 没有REST层。 MCP over SSE是唯一的运输方式。
  • McpMessageStatusFilter 补丁 /mcp/message POST响应来自 200 OK 到 202 Accepted 以满足克劳德对苏格兰和南方能源公司的期望。
  • 消费者运营(consumeMessages, peekMessages)旋转短暂 KafkaConsumer 具有唯一组ID的实例，这样它们就不会干扰应用程序消费者。

配置

关键属性 application.properties:

server.port=9085
spring.ai.mcp.server.transport=sse
spring.ai.mcp.server.sse-endpoint=/sse
spring.ai.mcp.server.sse-message-endpoint=/mcp/message
spring.kafka.bootstrap-servers=localhost:9092

测试

# Full test suite
mvnw.cmd test

# Single test class
mvnw.cmd test -Dtest=com.elzakaria.kafkamcpsbai.SomeTestClass

# Single test method
mvnw.cmd test -Dtest="com.elzakaria.kafkamcpsbai.SomeTestClass#methodName"

相关 MCP