Kafka MCP Server

Kafka MCP 服务

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

概览

本项目实现了一个服务，允许 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  # On Windows, use: venv\Scripts\activate
   

3. 安装所需依赖项：

   pip install -r requirements.txt
   

   如果不存在 requirements.txt，请安装以下软件包：

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

配置

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

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

# Optional: Custom Tool Descriptions
# TOOL_PUBLISH_DESCRIPTION="Custom description for the publish tool"
# TOOL_CONSUME_DESCRIPTION="Custom description for the consume tool"

使用方法

运行服务

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

python main.py --transport stdio

可用的传输选项：

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

与 Claude Desktop 集成

若要将此 Kafka MCP 服务与 Claude Desktop 搭配使用，请将以下配置添加到 Claude Desktop 配置文件中：

{
    "mcpServers": {
        "kafka": {
            "command": "python",
            "args": [
                "<PATH TO PROJECTS>/main.py"
            ]
        }
    }
}

将 <PATH TO 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跨 Broker 的复制因子
  • --config（可选）主题级配置覆盖项（例如，retention.ms=604800000）

Delete-Topic

删除现有的 Kafka 主题。

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

List-Topics

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

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

Topic-Configuration

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

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

Topic-Metadata

获取有关主题或集群的元数据。

• 选项：
  • --topic（如果提供）仅获取此主题的元数据
  • --bootstrap-serverBroker 地址
  • --include-offline（可选）包含处于离线状态的 Broker 或分区