🔌 WKafka Model Context Protocol (MCP) Server

An advanced Model Context Protocol (MCP) server designed to enable AI coding assistants (like Gemini, Claude, Cursor, OpenCode, and Antigravity) to design, build, test, and adapt high-performance event-driven microservices based on WKafka (the standard Kafka orchestrator suite).

🎯 Why This MCP Server Exists

Building streaming pipelines and microservices with Kafka often requires complex configuration, custom serialization patterns, error-handling logic, and safety contexts (like SASL and SSL).

This MCP server acts as an expert assistant interface for AI agents. It equips them with the tools and domain-specific knowledge to:

1. Instantly Scaffold production-ready Kafka services adhering to strict architectural structures.
2. Generate clean code for JSON, raw images, and video stream consumers and producers.
3. Adapt existing scripts automatically to run under Kafka message triggers.
4. Validate Kafka connection configs to prevent hardcoded credentials or missing settings.

🛠️ Technologies and Ecosytem Libraries Used

This component is built using the following core technologies:

• Python (>=3.10): Core programming language.

• FastMCP: High-productivity framework for building Model Context Protocol servers in Python.
• MCP CLI/SDK: Protocol implementations supporting integration with AI environments.
• Pydantic (v2): Advanced data validation and settings schema definitions.
• Pytest & Pytest-Cov: Testing framework and coverage reports.
• Docker: For sandboxed unit test execution.

🚀 Installation & Setup

1. Install the Package

Install wkafka-mcp via PyPI:

pip install wkafka-mcp

2. Configure in your AI Agent Config

OpenCode Configuration

Add to your ~/.config/opencode/opencode.jsonc inside the "mcp" block:

    "wkafka-mcp": {
      "type": "local",
      "command": [
        "python",
        "-m",
        "wkafka_mcp.server",
        "run"
      ],
      "enabled": true
    }

Antigravity (agy) / Gemini CLI Configuration

Add to your global ~/.gemini/antigravity/mcp_config.json:

{
  "mcpServers": {
    "wkafka-mcp": {
      "command": "python",
      "args": [
        "-m",
        "wkafka_mcp.server",
        "run"
      ]
    }
  }
}

⚙️ Detailed MCP Tools List

The server exposes the following MCP tools to your agent:

💡 MCP Usage Examples

Here are some typical requests you can make to your AI assistant:

Example 1: Create a basic consumer

User request: "Crea un consumidor de kafka para el topico orders_stream"

• Agent action: The agent will ask clarifying questions about whether the payload is JSON or an image, and if a group_id is required. Then it will call generate_wkafka_consumer.

Example 2: Adapt an existing script to run as a trigger

User request: "Toma este script.py y adaptalo para que la funcion prueba1() se lance por trigger de kafka en el topico orders"

• Agent action: The agent will call adapt_code_to_wkafka, automatically wrapping the execution logic inside the callback so prueba1(data) receives the message payload upon event ingestion.

💬 Interactive Questions Flow

When requesting new consumer configurations, the assistant will automatically present you with options to select:

1. Payload Type: JSON (Structured data/dict) vs Image (cv2 frames) vs Video streams.
2. Security: Plain local configuration vs SASL Authenticated context.
3. Consumer Group: Optional choice to specify a custom group_id or default to dynamic group generators.

🧪 Running the Tests

To ensure code stability and API contracts are preserved, a comprehensive unit test suite is included.

Run Locally (pytest)

# Install development dependencies
make install

# Execute the test suite
make test

Coverage Reports

To run the tests and calculate code coverage, execute the provided script:

./run_coverage.sh

Sandboxed Testing with Docker

To run the test suite in an isolated Python 3.13 environment container (independent of local packages):

./run_tests_docker.sh

Generated by WKafka MCP by wisrovi