AI导哈

Kafka消息协议服务器

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

  • by pavanjava
  • Python
  • 本地部署
  • 17 星标
  • 更新于
在 GitHub 查看 pavanjava/kafka_mcp_server

使用说明

Kafka MCP服务器

一个消息上下文协议（MCP）服务器，与Apache Kafka集成，为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  # 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主题。

卡夫卡消费

从已配置的Kafka主题中获取信息。

  • 注意：一旦从主题中读取消息，就无法使用相同的groupid再次读取

创建主题

使用指定参数创建新的Kafka主题。

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

删除主题

删除现有的Kafka主题。

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

列出主题

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

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

主题配置

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

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

主题元数据

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

  • 选项:
    • --topic（如果提供）仅获取此主题的元数据
    • --bootstrap-server经纪人地址
    • --include-offline（可选）包括脱机的代理或分区

相关 MCP