Кафка MCP-сервер

Сервер протокола контекста сообщений (MCP), который интегрируется с Apache Kafka для предоставления функций публикации и потребления для приложений LLM и Agentic.

Обзор

Этот проект реализует сервер, который позволяет моделям ИИ взаимодействовать с темами Kafka через стандартизированный интерфейс. Он поддерживает:

Публикация сообщений в темах Kafka
Потребление сообщений из тем Kafka

Related MCP server: Slack MCP Server

Предпосылки

Питон 3.8+
Экземпляр Apache Kafka
Зависимости Python (см. раздел Установка)

Установка

Клонируйте репозиторий:
git clone <repository-url> cd <repository-directory>
Создайте виртуальную среду и активируйте ее:
python -m venv venv source venv/bin/activate # On Windows, use: venv\Scripts\activate
Установите необходимые зависимости:
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 : реализация сервера MCP с инструментами для взаимодействия с Kafka
settings.py : Управление конфигурацией с помощью Pydantic

Доступные инструменты

kafka-опубликовать

Публикует информацию в настроенной теме Kafka.

кафка-потреблять

использовать информацию из настроенной темы Kafka.

Примечание: после прочтения сообщения из темы его нельзя будет прочитать снова, используя тот же идентификатор группы.

Создать тему

Создает новую тему 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 (необязательно) Включить брокеров или разделы, которые находятся в автономном режиме

Kafka MCP Server