关键词

KafkaKafka ConnectBurrowCruise Control数据集成LLM工具上下文服务

使用说明

项目简介

mcp-kafka 是一个基于 Model Context Protocol (MCP) 的服务器端实现，专注于 Apache Kafka 及其生态系统集成。它允许语言模型 (LLM) 通过标准化的 MCP 协议，安全可靠地与 Kafka 集群、Kafka Connect、Burrow 和 Cruise Control 等组件进行交互，从而为 LLM 应用提供丰富的上下文数据和操作能力。

主要功能点

  • Kafka 核心操作: 支持 Kafka Admin API、Consumer API 和 Producer API 的常用功能，例如：
    • 集群信息查询: 获取 Kafka 集群的 Broker 节点、控制器等元数据。
    • 主题管理: 列出 Kafka 主题。
    • 消费者组管理: 列出和描述 Kafka 消费者组的详细信息。
    • 消息生产与消费: 向 Kafka 主题发送消息，以及从主题消费消息。
  • Kafka Connect 集成: 支持 Kafka Connect REST API，允许 LLM 获取 Kafka Connect 集群和连接器信息。
  • Burrow 集成: 支持 Burrow REST API，用于监控 Kafka 消费者组的健康状态和积压情况。
  • Cruise Control 集成: 支持 Cruise Control REST API，用于获取 Kafka 集群的负载、状态和优化建议等信息。
  • 动态能力加载: 根据环境变量配置，动态启用或禁用不同的 API 功能模块。
  • Prompt 模板: （待完善）计划支持 Prompt 模板，以便更灵活地与 LLM 进行交互。

安装步骤

  1. 环境准备: 确保已安装 Python 3.8 或更高版本。
  2. 克隆仓库: 从 GitHub 克隆 'mcp-kafka' 仓库到本地。 
    git clone https://github.com/brandon-powers/mcp-kafka.git
cd mcp-kafka
  3. 安装依赖: 使用 pip 安装项目依赖。建议使用虚拟环境。 
    pip install -r requirements.txt
    或者 (如果 'requirements.txt' 不存在，根据 'server.py' 中的 'dependencies' 列表手动安装): 
    pip install python-dotenv uvicorn confluent-kafka requests

服务器配置

MCP 客户端需要配置以下信息以连接到 'mcp-kafka' 服务器。配置信息为 JSON 格式：

{
  "server_name": "mcp-kafka",
  "command": "python",
  "args": ["-m", "mcp_kafka.server"]
}

参数说明:

  • 'server_name': 服务器名称，设置为 "mcp-kafka"。
  • 'command': 启动服务器的命令，这里使用 'python'。
  • 'args': 命令参数，指定运行 'mcp_kafka.server' 模块以启动服务器。

环境变量配置 (重要):

'mcp-kafka' 服务器的功能模块通过环境变量进行启用和配置。您需要根据需要设置以下环境变量：

  • 'KAFKA_BOOTSTRAP_SERVERS': Kafka 集群的 Bootstrap Servers 地址。启用 Kafka Admin/Consumer/Producer API 功能模块。 例如：'localhost:9092'。
  • 'KAFKA_CONNECT_API_URL': Kafka Connect REST API 的 URL。启用 Kafka Connect 集成功能模块。 例如：'http://localhost:8083'。
  • 'KAFKA_BURROW_API_URL': Burrow REST API 的 URL。启用 Burrow 集成功能模块。 例如：'http://localhost:8000'。
  • 'KAFKA_CRUISE_CONTROL_API_URL': Cruise Control REST API 的 URL。启用 Cruise Control 集成功能模块。 例如：'http://localhost:9090'。

启动服务器:

在配置好环境变量后，即可启动 'mcp-kafka' 服务器。在项目根目录下执行：

python -m mcp_kafka.server

服务器默认监听 MCP 客户端的请求 (具体端口和协议取决于 'FastMCP' 框架的默认配置，请参考 'FastMCP' 文档或仓库信息)。

基本使用方法

  1. 配置 MCP 客户端: 在您的 LLM 应用或 MCP 客户端中，配置连接到 'mcp-kafka' 服务器，使用上述提供的服务器配置 JSON。

  2. 调用工具: 通过 MCP 客户端，您可以调用 'mcp-kafka' 服务器提供的工具，例如：

    • 'Describe cluster' (描述 Kafka 集群信息)
    • 'List topics' (列出 Kafka 主题)
    • 'List & describe consumer groups' (列出和描述消费者组)
    • 'Publish message to Kafka topic' (发布消息到 Kafka 主题)
    • 'Read messages from Kafka topic' (从 Kafka 主题读取消息)
    • 'Describe Connect connectors' (描述 Kafka Connect 连接器)
    • 'Describe Burrow group status' (描述 Burrow 消费者组状态)
    • 'Get Cruise Control (CC) status' (获取 Cruise Control 状态)

    具体的工具调用方式和参数，请参考 MCP 协议文档和 'mcp-kafka' 项目的代码及注释。

注意: 'mcp-kafka' 项目目前处于 WIP (Work In Progress) 状态，部分功能可能尚未完善或存在潜在问题。请关注仓库的更新和文档说明。

信息

分类

数据库与文件

访问项目