Kafka MCP服务器

一个与Apache Kafka集成的消息上下文协议（MCP）服务器，提供LLM和代理应用程序的发布和消费功能。

概述

该项目实现了一个允许AI模型通过标准化接口与Kafka主题进行交互的服务器。它支持：

• 向Kafka主题发布消息
• 从Kafka主题消费消息

预备条件

• Python 3.8+
• Apache Kafka实例
• Python依赖项（见安装部分）

安装

1. 克隆仓库：

   git clone <repository-url>
   cd <repository-directory>
   

2. 创建虚拟环境并激活它：

   python -m venv venv
   source venv/bin/activate  # 在Windows上使用：venv\Scripts\activate
   

3. 安装所需的依赖项：

   pip install -r requirements.txt
   

   如果没有requirements.txt文件，请安装以下包：

   pip install aiokafka python-dotenv pydantic-settings mcp-server
   

配置

在项目根目录创建一个.env文件，并包含以下变量：

# Kafka配置
KAFKA_BOOTSTRAP_SERVERS=localhost:9092
TOPIC_NAME=your-topic-name
IS_TOPIC_READ_FROM_BEGINNING=False
DEFAULT_GROUP_ID_FOR_CONSUMER=kafka-mcp-group

# 可选：自定义工具描述
# TOOL_PUBLISH_DESCRIPTION="发布工具的自定义描述"
# TOOL_CONSUME_DESCRIPTION="消费工具的自定义描述"

使用方法

运行服务器

你可以使用提供的main.py脚本运行服务器：

python main.py --transport stdio

可用传输选项：

• stdio：标准输入/输出（默认）
• sse：服务器发送事件

与Claude Desktop集成

要将此Kafka MCP服务器与Claude Desktop一起使用，请在你的Claude Desktop配置文件中添加以下配置：

{
    "mcpServers": {
        "kafka": {
            "command": "python",
            "args": [
                "<PROJECTS绝对路径>/main.py"
            ]
        }
    }
}

将<PROJECTS绝对路径>替换为你的项目目录的绝对路径。

项目结构

• main.py：应用程序的入口点
• kafka.py：Kafka连接器实现
• server.py：具有Kafka交互工具的MCP服务器实现
• settings.py：使用Pydantic进行配置管理

可用工具

kafka-publish

向配置的Kafka主题发布信息。

kafka-consume

从配置的Kafka主题消费信息。

• 注意：一旦消息被读取，使用相同的groupid无法再次读取该消息。

Create-Topic

创建一个新的Kafka主题，指定参数。

• 选项：
  • --topic：要创建的主题名称
  • --partitions：分配的分区数
  • --replication-factor：跨代理的复制因子
  • --config（可选）：主题级别的配置覆盖（例如，retention.ms=604800000）

Delete-Topic

删除现有的Kafka主题。

• 选项：
  • --topic：要删除的主题名称
  • --timeout（可选）：等待删除完成的时间

List-Topics

列出集群中的所有主题（或按模式过滤）。

• 选项：
  • --bootstrap-server：代理地址
  • --pattern（可选）：用于过滤主题名称的正则表达式
  • --exclude-internal（可选）：排除内部主题（默认值：true）

Topic-Configuration

显示或更改一个或多个主题的配置。

• 选项：
  • --describe：显示主题的当前配置
  • --alter：修改配置（例如，--add-config retention.ms=86400000,--delete-config cleanup.policy）
  • --topic：主题名称

Topic-Metadata

检索关于主题或集群的元数据。

• 选项：
  • --topic（如果提供）：仅获取此主题的元数据
  • --bootstrap-server：代理地址
  • --include-offline（可选）：包括离线的代理或分区