KafkaIQ

KafkaIQ provides an MCP server that lets you manage and monitor an Apache Kafka cluster through natural language using the Model Control Protocol. You run a Python-based MCP server locally, connect your AI assistant to its MCP endpoint, and issue high-level Kafka operations as simple conversations. This approach turns complex Kafka management into intuitive, automated workflows.

How to use

You interact with the MCP server by starting the server locally and then connecting your AI assistant to its endpoint. Use natural language commands to perform core Kafka tasks such as connecting to the cluster, listing topics, creating topics, checking health, analyzing consumer lag, and receiving alerts. You can chain actions into automated monitoring and incident workflows, enabling proactive operations and rapid troubleshooting.

How to install

# Prerequisites
Python 3.8+
Apache Kafka cluster accessible from the host

# Setup and run
# 1) Install Python dependencies (from your project directory)
pip install fastmcp kafka-python

# 2) Start the MCP server locally
python kafka_mcp_server.py

Configuration and usage notes

Start by ensuring the MCP server runs on the default local port and is reachable at http://127.0.0.1:8080. You can configure email notifications and temporal memory settings through environment variables before starting the server. Use Gmail credentials if you want alert emails, and tune how long you keep historical events and how many events you store for trend analysis.

Security and environment configuration

Set up authentication and secure access as you deploy. Use environment variables to keep credentials out of code, enable SSL/TLS for Kafka connections in production, and apply network access controls. The MCP server supports HTTP-based interaction and can be extended with API keys for additional security.

Troubleshooting

If the server cannot start, check that Python 3.8+ is installed and that the script kafka_mcp_server.py is present in your working directory. Ensure the Kafka cluster is reachable from the host and that port 8080 is not blocked by a firewall. If Gmail notifications fail, verify Gmail App Password usage and two-factor authentication settings.

Temporal memory and advanced usage

KafkaIQ includes temporal memory to track alerts and trends over time. You can configure retention and maximum event counts to support trend analysis, recurring issue detection, and lag trend insights. Use these settings to distinguish transient spikes from persistent problems and to build historical dashboards for your Kafka operations.

Example workflows

Set up a monitoring rule to alert you when consumer lag exceeds a threshold, or create topics for new microservices as your architecture evolves. You can build automated sequences that start with a health check, then trigger email notifications or onboarding checks for new topics.

Notes on usage with MCP clients

Connect your MCP-enabled AI assistant to the local endpoint and begin issuing natural language commands. The assistant translates your requests into MCP calls that the server executes against your Kafka cluster, returning structured results you can present in dashboards or alerts.

Advanced operations

You can combine high-level commands into batch operations, such as creating multiple topics for a new microservices architecture or configuring monitoring rules that automatically email on high lag. The system supports a variety of operational patterns to speed up delivery and reduce manual intervention.

Available tools

initialize_kafka_connection

Connect to a Kafka cluster using MCP; establish a session with the broker endpoint and verify connectivity.

list_kafka_topics

Return a list of all topics in the connected Kafka cluster with a count.

describe_kafka_topic

Fetch details for a specific topic, including partitions and configs.

create_kafka_topic

Create a new topic with a specified name and partition count.

delete_kafka_topic

Delete a specified topic from the cluster.

get_kafka_topic_config

Retrieve configuration parameters for a given topic.

kafka_health_check

Perform a health check on the Kafka connection and basic cluster health.

get_cluster_details

Return an overview of the connected Kafka cluster, including broker count and topics.

get_consumer_lag

Analyze consumer lag for a given consumer group and topic.

kafka_alert_summary

Generate a summary of cluster health and notable alerts.

broker_leadership_distribution

Analyze partition leadership distribution across brokers.

get_offline_partitions

Identify partitions that are offline or lack leaders.

get_broker_resources

Provide basic broker resource information.

send_email_notification

Send alert emails with cluster status or custom messages.

get_temporal_health_summary

Analyze health trends over a specified time window.

get_recurring_issues

Detect issues that occur repeatedly within a time window.

get_issue_frequency

Count occurrences of specific issues within a window.

get_lag_trend_analysis

Analyze lag trends for a given topic/group over time.

get_partition_health_history

Track partition offline/online history and flapping events.

detect_transient_vs_persistent

Classify whether an issue is a temporary spike or persistent.

export_temporal_state

Save temporal state data for later restoration.

import_temporal_state

Load previously saved temporal state data.

get_temporal_statistics

Report memory usage and retention statistics for temporal data.

clear_temporal_memory

Reset all temporal memory data to start fresh.