Kafka MCP Server

Kafka MCP Server是一個基於Go語言實現的Apache Kafka模型上下文協議(MCP)服務器，通過標準化接口讓LLM模型能夠執行Kafka操作。它提供了生產/消費消息、管理主題和消費者組、監控集群健康等功能，支持多種安全認證方式，並可與Cursor、Claude等AI應用集成。

開發者工具數據庫 #Kafka工具 #AI集成 #消息隊列 #協議服務 .Go

評分 : 2.5分

下載量 : 5.4K

更新時間 : 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 官方文檔或聯繫技術支持團隊。

Baidu Map

已認證

百度地圖MCP Server是國內首個兼容MCP協議的地圖服務，提供地理編碼、路線規劃等10個標準化API接口，支持Python和Typescript快速接入，賦能智能體實現地圖相關功能。

Markdownify是一個多功能文件轉換服務，支持將PDF、圖片、音頻等多種格式及網頁內容轉換為Markdown格式。

Firecrawl MCP Server是一個集成Firecrawl網頁抓取能力的模型上下文協議服務器，提供豐富的網頁抓取、搜索和內容提取功能。

TypeScript

65.8K

5分

Sequential Thinking MCP Server

一個基於MCP協議的結構化思維服務器，通過定義思考階段幫助分解複雜問題並生成總結

一個基於Python的MCP服務器，通過Notion API提供高級待辦事項管理和內容組織功能，實現AI模型與Notion的無縫集成。

Magic Component Platform (MCP) 是一個AI驅動的UI組件生成工具，通過自然語言描述幫助開發者快速創建現代化UI組件，支持多種IDE集成。

JavaScript

15.5K

5分

Edgeone Pages MCP Server

EdgeOne Pages MCP是一個通過MCP協議快速部署HTML內容到EdgeOne Pages並獲取公開URL的服務

Context7 MCP是一個為AI編程助手提供即時、版本特定文檔和代碼示例的服務，通過Model Context Protocol直接集成到提示中，解決LLM使用過時信息的問題。

智啟未來，您的人工智慧解決方案智庫

Kafka MCP Server

概述

安裝

內容詳情

替代品

什麼是Kafka MCP Server？

如何使用Kafka MCP Server？

適用場景

主要功能

如何使用

使用案例

常見問題

相關資源

安裝

🚀 Kafka 分佈式計算平臺 (MCP) 用戶指南

🚀 快速開始

環境要求

安裝步驟

✨ 主要特性

消費組管理

主題管理

消息跟蹤

📦 安裝指南

環境要求

安裝步驟

💻 使用示例

基礎用法

高級用法

消費組與主題詳細信息查看

主題偏移量管理

生產者和消費者日誌監控

📚 詳細文檔

配置參考

系統配置

環境變量

使用案例

案例一：消費組偏移量查詢

案例二：主題描述

🔧 技術細節

安全注意事項

📄 許可證

🆘 幫助與支持

替代品