Skip to main content

kafka-connect-mcp

Tests

An MCP server that exposes Kafka Connect REST API operations as tools, letting LLMs manage connectors, tasks, and plugins through natural language.

Tools

Tool Kafka Connect Endpoint Description
get_cluster_info GET / Cluster version and metadata
list_connectors GET /connectors List all connector names
get_connector GET /connectors/{name} Connector info, config, and tasks
get_connector_status GET /connectors/{name}/status Connector and task states
get_connector_config GET /connectors/{name}/config Connector configuration
create_connector POST /connectors Create a new connector
update_connector_config PUT /connectors/{name}/config Replace connector configuration
delete_connector DELETE /connectors/{name} Delete a connector
pause_connector PUT /connectors/{name}/pause Pause a connector
resume_connector PUT /connectors/{name}/resume Resume a paused connector
restart_connector POST /connectors/{name}/restart Restart a connector (optionally tasks)
get_task_status GET /connectors/{name}/tasks/{id}/status Status of a specific task
restart_task POST /connectors/{name}/tasks/{id}/restart Restart a specific task
list_connector_plugins GET /connector-plugins Available plugins on the cluster
validate_connector_config PUT /connector-plugins/{name}/config/validate Validate config against plugin schema

Setup

Prerequisites

  • Python 3.12+
  • uv
  • A running Kafka Connect cluster (or use the included Docker Compose)

Install

uv sync

Add to Claude Code

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

Configuration

Environment Variable Default Description
KAFKA_CONNECT_URL http://localhost:8083 Kafka Connect REST API base URL
KAFKA_CONNECT_ENABLE_CREATE false Allow create_connector
KAFKA_CONNECT_ENABLE_UPDATE false Allow update_connector_config
KAFKA_CONNECT_ENABLE_DELETE false Allow delete_connector
KAFKA_CONNECT_ENABLE_PAUSE_RESUME false Allow pause_connector and resume_connector
KAFKA_CONNECT_ENABLE_RESTART false Allow restart_connector and restart_task
KAFKA_CONNECT_MUTATION_ALLOWLIST (empty) Optional comma-separated connector allowlist for mutating operations

Safe mode (capability-gated)

This server is read-only by default because all mutation capabilities default to false. Mutating tools are blocked unless you explicitly enable the specific capability.

Examples:

Restart-only mode for selected connectors:

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

Enable all mutating operations (for development only):

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

Running

stdio (default, for Claude Code)

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

SSE (for Docker / remote)

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

Docker Compose (full stack)

Spins up Zookeeper, Kafka, Kafka Connect (with the Datagen plugin), and the MCP server:

docker compose up --build -d
Service Port Description
zookeeper 2181 ZooKeeper
kafka 9092 Kafka broker
kafka-connect 8083 Kafka Connect REST API
mcp-server 8000 MCP server (SSE transport)

Example: create a Datagen connector

Once the stack is up, ask Claude to create a datagen connector, or do it directly:

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"
    }
  }'

Tests

uv run pytest tests/ -v

Tests use respx to mock HTTP calls to the Kafka Connect API -- no running cluster required.

Releases

Releases are automated from main:

  1. Bump project.version in pyproject.toml (for example 0.1.0 -> 0.1.1).
  2. Merge to main.
  3. GitHub Actions creates tag/release vX.Y.Z.
  4. The published release is automatically built and published to PyPI.

One-time PyPI setup

Set up PyPI Trusted Publishing for project kafka-connect-mcp:

  • Owner: lawrencemq
  • Repository: kafka-connect-mcp
  • Workflow: .github/workflows/release.yml
  • Environment: (none required by this workflow)

Project structure

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

License

Licensed under the Apache License, Version 2.0. See LICENSE.

Metadata

Release files for kafka-connect-mcp 0.1.6

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for kafka-connect-mcp 0.1.6
File Size Uploaded Provenance Actions
kafka_connect_mcp-0.1.6.tar.gz 5.4 kB With provenance Details for kafka_connect_mcp-0.1.6.tar.gz

Built distribution (wheel)

Table of built distributions (wheels) for kafka-connect-mcp 0.1.6
File Interpreter ABI Platform Actions
kafka_connect_mcp-0.1.6-py3-none-any.whl Python 3 none any Details for kafka_connect_mcp-0.1.6-py3-none-any.whl

Total release size: 12.6 kB

Release files / kafka_connect_mcp-0.1.6.tar.gz

Download URL kafka_connect_mcp-0.1.6.tar.gz
Size 5.4 kB
Tags Source
SHA-256 checksum
How to use checksums
8fc37fa0ef1057b0595f5001f1d70edb73808072e71254d9a671bcdfa6ac2501
BLAKE2b-256 checksum
How to use checksums
f03a2daca099ed2be81082272879ba6f7d5ddd6cbaa6149ef3144b46f7a14770
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
Yes
Uploaded via twine/7.0.0 CPython/3.13.16

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Nov 9, 2026.

Transparency log

Release files / kafka_connect_mcp-0.1.6-py3-none-any.whl

Download URL kafka_connect_mcp-0.1.6-py3-none-any.whl
Size 7.2 kB
Tags Python 3
SHA-256 checksum
How to use checksums
b58a9e02cd533f2a6763cfc7bd16cb22375d85f79299709b82be97afb81d670d
BLAKE2b-256 checksum
How to use checksums
555f025a6e0b519eee02d865af26f772d90504084b076cf4bbf2b08946f4d66e
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
Yes
Uploaded via twine/7.0.0 CPython/3.13.16

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Nov 9, 2026.

Transparency log

Security for kafka-connect-mcp 0.1.6

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

Full provenance

Both files in this release have been uploaded with provenance metadata.

Select from 2 files

Attestation for kafka_connect_mcp-0.1.6.tar.gz

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Nov 9, 2026.

Transparency log

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.

To learn more about protecting yourself from malicious actors, visit our package security documentation.

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page