卡夫卡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"
