AI导哈

Kafka连接管理工具

一个将Kafka Connect REST API操作暴露为工具的MCP服务器，允许LLM通过自然语言管理连接器、任务和插件。

  • by lawrencemq
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 lawrencemq/kafka-connect-mcp

使用说明

kafka连接mcp

一个MCP服务器，它公开 Kafka Connect REST API 操作作为工具，让LLM通过自然语言管理连接器、任务和插件。

工具

工具 Kafka连接端点 说明
get_cluster_info GET / 集群版本和元数据
list_connectors GET /connectors 列出所有连接器名称
get_connector GET /connectors/{name} 连接器信息、配置和任务
get_connector_status GET /connectors/{name}/status 连接器和任务状态
get_connector_config GET /connectors/{name}/config 连接器配置
create_connector POST /connectors 创建新连接器
update_connector_config PUT /connectors/{name}/config 更换连接器配置
delete_connector DELETE /connectors/{name} 删除连接器
pause_connector PUT /connectors/{name}/pause 暂停连接器
resume_connector PUT /connectors/{name}/resume 恢复暂停的连接器
restart_connector POST /connectors/{name}/restart 重新启动连接器（可选任务）
get_task_status GET /connectors/{name}/tasks/{id}/status 特定任务的状态
restart_task POST /connectors/{name}/tasks/{id}/restart 重新启动特定任务
list_connector_plugins GET /connector-plugins 集群上可用的插件
validate_connector_config PUT /connector-plugins/{name}/config/validate 根据插件架构验证配置

设置

先决条件

  • Python 3.12+
  • 紫外线
  • 正在运行的Kafka Connect集群（或使用附带的Docker Compose）

安装

uv sync

添加到克劳德代码

claude mcp add kafka-connect \
  -e KAFKA_CONNECT_URL=http://localhost:8083 \
  -- uv --directory /path/to/kafka-connect-mcp run kafka-connect-mcp

配置

环境变量 默认值 描述
KAFKA_CONNECT_URL http://localhost:8083 Kafka Connect REST API基础URL
KAFKA_CONNECT_ENABLE_CREATE false 允许 create_connector
KAFKA_CONNECT_ENABLE_UPDATE false 允许 update_connector_config
KAFKA_CONNECT_ENABLE_DELETE false 允许 delete_connector
KAFKA_CONNECT_ENABLE_PAUSE_RESUME false 允许 pause_connector 和 resume_connector
KAFKA_CONNECT_ENABLE_RESTART false 允许 restart_connector 和 restart_task
KAFKA_CONNECT_MUTATION_ALLOWLIST （空） 用于变异操作的可选逗号分隔连接器列表

安全模式（能力门控）

此服务器是 只读 默认情况下，因为所有变异功能默认为 false. 除非您明确启用特定功能，否则将阻止修改工具。

示例：

所选连接器的仅重新启动模式：

KAFKA_CONNECT_ENABLE_RESTART=true
KAFKA_CONNECT_MUTATION_ALLOWLIST=payments-sink,inventory-source

启用所有变异操作（仅用于开发）：

KAFKA_CONNECT_ENABLE_CREATE=true
KAFKA_CONNECT_ENABLE_UPDATE=true
KAFKA_CONNECT_ENABLE_DELETE=true
KAFKA_CONNECT_ENABLE_PAUSE_RESUME=true
KAFKA_CONNECT_ENABLE_RESTART=true

跑步

stdio（默认值，用于克劳德代码）

KAFKA_CONNECT_URL=http://localhost:8083 uv run kafka-connect-mcp

SSE（用于Docker/远程）

uv run kafka-connect-mcp --transport sse --host 0.0.0.0 --port 8000

Docker Compose（全栈）

启动Zookeeper、Kafka、Kafka Connect（使用Datagen插件）和MCP服务器：

docker compose up --build -d
服务 端口 描述
动物园管理员 2181 zookeeper
卡夫卡 9092 卡夫卡经纪人
kafka-connect 8083 kafka-connect REST API
mcp服务器 8000 mcp服务器（SSE传输）

示例：创建Datagen连接器

堆栈完成后，让Claude创建一个数据生成器连接器，或直接创建：

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "datagen-users",
    "config": {
      "connector.class": "io.confluent.kafka.connect.datagen.DatagenConnector",
      "kafka.topic": "users",
      "quickstart": "users",
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "value.converter.schemas.enable": "false",
      "max.interval": "1000",
      "tasks.max": "1"
    }
  }'

测试

uv run pytest tests/ -v

测试使用 respx 模拟对Kafka Connect API的HTTP调用——不需要运行集群。

发布

从以下位置自动发布 main:

  1. 碰撞 project.version 在 pyproject.toml （例如 0.1.0 -> 0.1.1).
  2. 合并到 main.
  3. GitHub Actions创建标签/发布 vX.Y.Z.
  4. 已发布的版本会自动构建并发布到PyPI。

一次性PyPI设置

为项目设置PyPI可信发布 kafka-connect-mcp:

  • 业主： lawrencemq
  • 存储库： kafka-connect-mcp
  • 工作流程： .github/workflows/release.yml
  • 环境： （此工作流不需要）

项目结构

kafka-connect-mcp/
├── pyproject.toml
├── Dockerfile
├── docker-compose.yml
├── src/kafka_connect_mcp/
│   ├── __init__.py
│   ├── safety.py            # Read-only and mutation policy gates
│   └── server.py            # MCP tools and entry point
└── tests/
    ├── conftest.py           # Shared fixtures (mock URL + respx router)
    ├── test_cluster.py
    ├── test_connectors.py
    ├── test_safety.py
    ├── test_tasks.py
    └── test_plugins.py

许可证

根据Apache许可证2.0版授权。看 许可证.

相关 MCP