Servidor MCP de Kafka

Un servidor de Protocolo de contexto de mensaje (MCP) que se integra con Apache Kafka para proporcionar funcionalidades de publicación y consumo para aplicaciones LLM y Agentic.

Descripción general

Este proyecto implementa un servidor que permite que los modelos de IA interactúen con temas de Kafka mediante una interfaz estandarizada. Admite:

• Publicar mensajes en temas de Kafka

• Consumir mensajes de temas de Kafka

Related MCP server: KafkaIQ

Prerrequisitos

• Python 3.8+

• Instancia de Apache Kafka

• Dependencias de Python (ver sección Instalación)

Instalación

1. Clonar el repositorio:

   git clone <repository-url>
   cd <repository-directory>

2. Crea un entorno virtual y actívalo:

   python -m venv venv
   source venv/bin/activate  # On Windows, use: venv\Scripts\activate

3. Instale las dependencias necesarias:

   pip install -r requirements.txt

   Si no existe ningún requirements.txt, instale los siguientes paquetes:

   pip install aiokafka python-dotenv pydantic-settings mcp-server

Configuración

Cree un archivo .env en la raíz del proyecto con las siguientes variables:

# 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"

Uso

Ejecución del servidor

Puede ejecutar el servidor utilizando el script main.py proporcionado:

python main.py --transport stdio

Opciones de transporte disponibles:

• stdio : Entrada/salida estándar (predeterminada)

• sse : Eventos enviados por el servidor

Integración con Claude Desktop

Para utilizar este servidor Kafka MCP con Claude Desktop, agregue la siguiente configuración a su archivo de configuración de Claude Desktop:

{
    "mcpServers": {
        "kafka": {
            "command": "python",
            "args": [
                "<PATH TO PROJECTS>/main.py"
            ]
        }
    }
}

Reemplace <PATH TO PROJECTS> con la ruta absoluta al directorio de su proyecto.

Estructura del proyecto

• main.py : Punto de entrada para la aplicación

• kafka.py : Implementación del conector de Kafka

• server.py : Implementación del servidor MCP con herramientas para la interacción con Kafka

• settings.py : Gestión de configuración mediante Pydantic

Herramientas disponibles

kafka-publicar

Publica información en el tema de Kafka configurado.

Kafka-consumir

consumir información del tema de Kafka configurado.

• Nota: una vez que se lee un mensaje del tema, no se puede volver a leer utilizando el mismo ID de grupo.

Crear tema

Crea un nuevo tema de Kafka con parámetros especificados.

• Opciones :

  • --topic Nombre del tema a crear

  • --partitions Número de particiones a asignar

  • --replication-factor Factor de replicación entre corredores

  • --config (opcional) Anulaciones de configuración a nivel de tema (por ejemplo, retention.ms=604800000 )

Eliminar tema

Elimina un tema de Kafka existente.

• Opciones :

  • --topic Nombre del tema a eliminar

  • --timeout (opcional) Tiempo de espera para que se complete la eliminación

Lista de temas

Enumera todos los temas del clúster (o filtrados por patrón).

• Opciones :

  • --bootstrap-server Dirección del agente

  • --pattern (opcional) Expresión regular para filtrar nombres de temas

  • --exclude-internal (opcional) Excluye temas internos (valor predeterminado: verdadero)

Configuración del tema

Muestra o modifica la configuración de uno o más temas.

• Opciones :

  • --describe Mostrar las configuraciones actuales de un tema

  • --alter Modificar configuraciones (por ejemplo, --add-config retention.ms=86400000,--delete-config cleanup.policy )

  • --topic Nombre del tema

Metadatos del tema

Recupera metadatos sobre un tema o el clúster.

• Opciones :

  • --topic (si se proporciona) Obtener metadatos solo para este tema

  • --bootstrap-server Dirección del agente

  • --include-offline (opcional) Incluye intermediarios o particiones que están fuera de línea