Kafka MCP 服务器

Kafka MCP 服务器是一个模型上下文协议（MCP）服务器，它提供了与Apache Kafka的集成，使用户能够通过自然语言（LLMs）与Kafka集群进行交互。

demo

预备条件

你需要有Docker或Golang来本地运行MCP服务器。

快速开始

你需要访问一个Kafka集群。请参考快速开始文档：

docker pull apache/kafka:4.0.0
docker run -p 9092:9092 apache/kafka:4.0.0

现在Kafka可以在localhost:9092上访问。

使用Claude Desktop

Docker:

{
  "mcpServers": {
    "kafka": {
      "command": "docker",
      "args": [
        "run",
        "-i",
        "--rm",
        "-e",
        "KAFKA_MCP_BOOTSTRAP_SERVERS",
        "ghcr.io/cefboud/kafka-mcp-server"
      ],
      "env": {
        "KAFKA_MCP_BOOTSTRAP_SERVERS": "localhost:9092"
      }
    }
  }
}

本地构建

cd <工作目录>
git clone https://github.com/CefBoud/kafka-mcp-server.git
cd kafka-mcp-server
go build -o kafka-mcp-server  cmd/kafka-mcp-server/main.go 

{
  "mcpServers": {
    "kafka": {
      "command": "<工作目录>/kafka-mcp-server/kafka-mcp-server",
      "args": [
        "stdio",
      ],
      "env": {
        "KAFKA_MCP_BOOTSTRAP_SERVERS": "localhost:9092"
      }
    }
  }
}

选项：

      --bootstrap-servers string   要连接的Kafka服务器的逗号分隔列表。
      --enable-command-logging     启用后，服务器将把所有命令请求和响应记录到日志文件中。
      --log-file string            日志文件的路径
      --read-only                  将服务器限制为只读操作

所有选项都可以作为环境变量传递，大写，并用下划线替换连字符，并以MCP_KAFKA_开头，例如，--bootstrap-servers变为MCP_KAFKA_BOOTSTRAP_SERVERS。

可用的MCP工具

• 列出主题
• 创建主题
• 消费消息。
• 生产消息。
• 描述集群（列出broker和controller）
• 列出消费者组及其滞后情况。
• 获取主题的最早和最晚偏移量（GetOffsetShell）
• 重置消费者组偏移量。
• Kafka Connect ??
• Schema Registry ??

🔀 MultiplexTool

demo MultiplexTool

运行许多顺序工具，特别是当每个工具都依赖于前一个工具的输出时，可能会非常繁琐且耗时。这需要客户端和服务器之间多次往返。

MultiplexTool通过允许客户端将一系列工具调用批处理成单个请求并按顺序执行它们来解决这个问题。它支持工具之间的动态依赖关系，让你可以通过基于提示的占位符引用早期的输出。

如果工具输入依赖于之前的某个结果，客户端会使用PROMPT_ARGUMENT:格式通过向LLM（目前仅支持Gemini）发出提示来动态生成该输入。 示例： "userId": "PROMPT_ARGUMENT: 创建用户的ID"

CLI标志：

• --enable-multiplex: 启用工具调用的多路复用。
• --multiplex-model: 指定用于推断PROMPT_ARGUMENT的模型（如gemini）。需要GEMINI_API_KEY环境变量。

{
  "mcpServers": {
    "kafka": {
      "command": "docker",
      "args": [
        "run",
        "-i",
        "--rm",
        "-e",
        "KAFKA_MCP_BOOTSTRAP_SERVERS",
        "ghcr.io/cefboud/kafka-mcp-server",
        "--enable-multiplex",
        "--multiplex-model",
        "gemini"
      ],
      "env": {
        "KAFKA_MCP_BOOTSTRAP_SERVERS": "localhost:9092",
        "GEMINI_API_KEY": "....."
      }
    }
  }
}

致谢

• github.com/github/github-mcp-server
• github.com/mark3labs/mcp-go
• github.com/IBM/sarama