Kafka MCP Server

Kafka MCP Server是一个基于Go语言实现的Apache Kafka模型上下文协议(MCP)服务器，通过标准化接口让LLM模型能够执行Kafka操作。它提供了生产/消费消息、管理主题和消费者组、监控集群健康等功能，支持多种安全认证方式，并可与Cursor、Claude等AI应用集成。

开发者工具数据库 #Kafka工具 #AI集成 #消息队列 #协议服务 .Go

评分 : 2.5分

下载量 : 5.9K

更新时间 : 2025-04-29

打开站点

什么是Kafka MCP Server？

Kafka MCP Server是一个用于与Apache Kafka交互的工具，它通过标准化的Model Context Protocol (MCP)协议，使大型语言模型（LLMs）能够轻松地执行常见的Kafka操作，如消息生产和消费、主题管理以及集群健康检查。

如何使用Kafka MCP Server？

只需安装服务器并配置环境变量即可开始使用。通过集成到您的AI应用中，您可以轻松调用Kafka操作，例如发送消息或查看消费者组状态。

适用场景

适用于需要自动化Kafka操作的场景，比如数据流监控、实时日志分析和大规模消息处理。

主要功能

生产消息

向指定的Kafka主题发送消息。

消费消息

从一个或多个主题中拉取消息。

列出代理

显示当前连接的Kafka代理地址。

描述主题

获取特定主题的详细元数据。

列出消费者组

显示活动的消费者组。

描述消费者组

展示特定消费者组的状态、成员和分区偏移量。

描述配置

检索特定资源（主题或代理）的配置项。

集群概览

提供Kafka集群的整体健康状况摘要。

优势

通过标准MCP协议简化Kafka操作。

支持多种身份验证机制（SASL和TLS）。

易于与其他AI工具集成。

全面的错误处理和反馈。

局限性

需要适当的权限才能访问Kafka集群。

某些高级功能可能需要额外的配置。

对低延迟要求较高的系统可能需要优化性能。

如何使用

安装Kafka MCP Server

克隆项目代码并在本地构建服务器。

配置环境变量

设置Kafka服务器地址、客户端ID以及其他必要的选项。

启动服务器

运行服务器以监听MCP请求。

使用案例

发送订单更新消息

演示如何向Kafka主题发送一条带有订单ID的消息。

列出消费者组

展示如何列出所有活动的消费者组。

常见问题

如何在生产环境中启用TLS加密？

如果我的消费者组出现滞后怎么办？

安装

复制以下命令到你的Client进行配置

{
  "mcpServers": {
    "kafka": {
      "command": "kafka-mcp-server",
      "env": {
        "KAFKA_BROKERS": "localhost:9092",
        "KAFKA_CLIENT_ID": "kafka-mcp-server",
        "MCP_TRANSPORT": "stdio",
        "KAFKA_SASL_MECHANISM": "",
        "KAFKA_SASL_USER": "",
        "KAFKA_SASL_PASSWORD": "",
        "KAFKA_TLS_ENABLE": "false",
        "KAFKA_TLS_INSECURE_SKIP_VERIFY": "false"
      }
    }
  }
}

{
  "mcpServers": {
    "kafka": {
      "command": "kafka-mcp-server",
      "env": {
        "KAFKA_BROKERS": "kafka-broker-1:9092,kafka-broker-2:9092",
        "KAFKA_CLIENT_ID": "kafka-mcp-server",
        "MCP_TRANSPORT": "stdio",
        "KAFKA_SASL_MECHANISM": "scram-sha-512",
        "KAFKA_SASL_USER": "kafka-user",
        "KAFKA_SASL_PASSWORD": "kafka-password",
        "KAFKA_TLS_ENABLE": "true",
        "KAFKA_TLS_INSECURE_SKIP_VERIFY": "false"
      }
    }
  }
}

{
  "mcpServers": {
    "kafka": {
      "command": "docker",
      "args": [
        "run", 
        "--rm", 
        "-i", 
        "-e", "KAFKA_BROKERS=kafka-broker:9092", 
        "-e", "KAFKA_CLIENT_ID=kafka-mcp-server",
        "-e", "MCP_TRANSPORT=stdio",
        "-e", "KAFKA_SASL_MECHANISM=", 
        "-e", "KAFKA_SASL_USER=", 
        "-e", "KAFKA_SASL_PASSWORD=", 
        "-e", "KAFKA_TLS_ENABLE=false",
        "ghcr.io/tuannvm/kafka-mcp-server:latest"
      ],
      "env": {}
    }
  }
}

{
  "mcpServers": {
    "kafka": {
      "command": "kafka-mcp-server",
      "args": [],
      "env": {
        "KAFKA_BROKERS": "localhost:9092",
        "KAFKA_CLIENT_ID": "kafka-mcp-server",
        "MCP_TRANSPORT": "stdio",
        "KAFKA_SASL_MECHANISM": "",
        "KAFKA_SASL_USER": "",
        "KAFKA_SASL_PASSWORD": "",
        "KAFKA_TLS_ENABLE": "false"
      }
    }
  }
}

{
  "mcpServers": {
    "kafka-http": {
      "url": "http://localhost:9097/sse"
    }
  }
}

注意：您的密钥属于敏感信息，请勿与任何人分享。

🚀 Kafka 分布式计算平台 (MCP) 用户指南

Kafka 分布式计算平台（简称 Kafka MCP）是一套强大的工具集，能简化与 Apache Kafka 集群的交互。它提供了多种命令行工具和 REST API，可用于消费组描述、主题偏移量管理以及生产者/消费者日志记录等操作。

🚀 快速开始

Kafka 分布式计算平台（Kafka MCP）是一个强大的工具集，可帮助用户轻松与 Apache Kafka 集群进行交互。下面为你介绍如何快速开启使用之旅。

环境要求

Java 8 或更高版本
Kafka 版本 2.13+

安装步骤

下载并解压 Kafka MCP 发行包：

wget https://example.com/kafka-mcp.tar.gz
tar -xzvf kafka-mcp.tar.gz
cd kafka-mcp

配置环境变量（可选）：

export PATH=$PATH:/path/to/kafka-mcp/bin

启动 MCP 服务（如果使用 REST API 版本）：

./start_mcp_server.sh

✨ 主要特性

消费组管理

消费组描述：可获取指定消费组的消费者 ID 及其所属客户端 ID。
偏移量查询：能查看特定主题和分区在消费组中的当前消费位置。
重置偏移量：支持手动调整消费组的偏移量，实现精确消费控制。

主题管理

主题创建：可通过定义分区数、复制因子等参数快速创建新主题。
主题删除：能安全删除不再需要的主题，释放集群资源。
ISR 监控：可实时查看主题分区的 _isr（In-Sync Replicas）状态，确保数据一致性。

消息跟踪

消息消费：可实时消费指定主题的消息，并支持按时间范围、批次大小等方式灵活获取。
生产日志生成：能方便地向指定主题发送测试或调试用的消息。
偏移量调整：必要时可手动调整消费者偏移量，实现精准的流处理。

📦 安装指南

环境要求

属性	详情
运行环境	Java 8 或更高版本
Kafka 版本	2.13+

安装步骤

下载并解压 Kafka MCP 发行包：

wget https://example.com/kafka-mcp.tar.gz
tar -xzvf kafka-mcp.tar.gz
cd kafka-mcp

配置环境变量（可选）：

export PATH=$PATH:/path/to/kafka-mcp/bin

启动 MCP 服务（如果使用 REST API 版本）：

./start_mcp_server.sh

💻 使用示例

基础用法

Kafka MCP 提供了多种命令行工具，以下是一些基础用法示例：

# 获取所有Kafka消费组列表
list_consumer_groups --bootstrap-server <broker_address>

# 列出Kafka集群中的所有主题
list_topics --bootstrap-server <broker_address>

高级用法

消费组与主题详细信息查看

# 查看指定消费组的详细信息，包括消费者ID和偏移量
describe_consumer_group --group-id <consumer_group_id> \
                       --bootstrap-server <broker_address> \
                       --include-protocol-info \
                       --include-consumer-info \
                       --include-offsets

# 查看指定主题的分区分配、副本和ISR状态
describe_topic --topic-name <topic_name> \
               --bootstrap-server <broker_address>

主题偏移量管理

# 消费特定主题的所有分区的偏移量
consume_offsets --group-id <consumer_group_id> \
               --topic-name <topic_name> \
               --bootstrap-server <broker_address>

# 提供指定分区的当前偏移量
seek_consumer_partition --group-id <consumer_group_id> \
                       --topic-name <topic_name> \
                       --partition <partition_number> \
                       --offset <target_offset> \
                       --bootstrap-server <broker_address>

生产者和消费者日志监控

# 消费指定主题的最新消息
consume_logs --topic-name <topic_name> \
             --bootstrap-server <broker_address>

# 生成模拟生产者日志
produce_logs --topic-name <topic_name> \
             --message "your_message_here" \
             --bootstrap-server <broker_address>

📚 详细文档

配置参考

系统配置

Kafka 代理地址（--bootstrap-server）：指定 Kafka 服务的 IP 和端口，格式为 host:port。
消费者组 ID（--group-id）：用于标识一组消费者实例。
主题名称（--topic-name）：操作的具体主题。

环境变量

# 设置默认代理地址
export KAFKA_BOOTSTRAP_SERVER=localhost:9092

使用案例

案例一：消费组偏移量查询

# 查看指定消费组的偏移量
consume_offsets --group-id my_consumer_group \
               --bootstrap-server localhost:9092

案例二：主题描述

# 获取特定主题的详细信息
describe_topic --topic-name user_activity \
               --bootstrap-server localhost:9092

🔧 技术细节

安全注意事项

认证授权：在生产环境中，建议启用 SASL/PLAINTEXT 或 TLS 加密通信，确保 API 调用的安全性。
访问控制：通过 Kafka 的 ACL（Access Control List）机制限制不同用户或应用的权限范围。
日志监控：定期审查操作日志，识别异常行为并及时响应。

📄 许可证

文档中未提及相关许可证信息。

🆘 帮助与支持

如需更多信息，请参考 Kafka 官方文档或联系技术支持团队。

Duckduckgo MCP Server

已认证

DuckDuckGo搜索MCP服务器，为Claude等LLM提供网页搜索和内容抓取服务

Framelink Figma MCP Server是一个为AI编程工具（如Cursor）提供Figma设计数据访问的服务器，通过简化Figma API响应，帮助AI更准确地实现设计到代码的一键转换。

Firecrawl MCP Server是一个集成Firecrawl网页抓取能力的模型上下文协议服务器，提供丰富的网页抓取、搜索和内容提取功能。

Exa MCP Server是一个为AI助手（如Claude）提供网络搜索功能的服务器，通过Exa AI搜索API实现实时、安全的网络信息获取。

TypeScript

40.2K

5分

Edgeone Pages MCP Server

EdgeOne Pages MCP是一个通过MCP协议快速部署HTML内容到EdgeOne Pages并获取公开URL的服务

MiniMax Model Context Protocol (MCP) 是一个官方服务器，支持与强大的文本转语音、视频/图像生成API交互，适用于多种客户端工具如Claude Desktop、Cursor等。

百度地图MCP Server是国内首个兼容MCP协议的地图服务，提供地理编码、路线规划等10个标准化API接口，支持Python和Typescript快速接入，赋能智能体实现地图相关功能。

Context7 MCP是一个为AI编程助手提供实时、版本特定文档和代码示例的服务，通过Model Context Protocol直接集成到提示中，解决LLM使用过时信息的问题。

智启未来，您的人工智能解决方案智库

Kafka MCP Server

概述

安装

内容详情

替代品

什么是Kafka MCP Server？

如何使用Kafka MCP Server？

适用场景

主要功能

如何使用

使用案例

常见问题

相关资源

安装

🚀 Kafka 分布式计算平台 (MCP) 用户指南

🚀 快速开始

环境要求

安装步骤

✨ 主要特性

消费组管理

主题管理

消息跟踪

📦 安装指南

环境要求

安装步骤

💻 使用示例

基础用法

高级用法

消费组与主题详细信息查看

主题偏移量管理

生产者和消费者日志监控

📚 详细文档

配置参考

系统配置

环境变量

使用案例

案例一：消费组偏移量查询

案例二：主题描述

🔧 技术细节

安全注意事项

📄 许可证

🆘 帮助与支持

替代品