How to Run an AI Kafka Message Queue Agent with OE Runtime

Publish events, check consumer lag, and monitor your Kafka topics with AI. OE Runtime lets you run a local Kafka agent that verifies topic health and delivers structured messages with confirmed offsets — no separate Kafka tooling required.

What You Need

• Node.js 18+ installed, or the OE Runtime binary from openenthrium.com/runtime
• An API key from OpenAI, Anthropic, or any supported LLM provider
• A Kafka broker address (local Kafka or a cloud broker like Confluent)
• A text editor — nothing else

Create the Project Folder

Create a folder called message-queues/ and add these two files:

The Skill Files

Create SKILL.md inside a message-queues/ directory. This is the portable skill — it follows the agentskills.io format and runs unchanged on Claude, Cursor, Windsurf, or OE Runtime:

---
name: message-queues
description: Publish messages to Kafka topics and verify delivery
license: Apache-2.0
metadata:
  author: Open Enthrium
  version: "1.0"
---

You are a message queue agent. Publish events to Kafka topics and confirm delivery.
Format all payloads as JSON. Include event_type, timestamp, and payload in every message.
Complete all steps fully before writing your report.

## Step 1: Check Topic Status
List the available Kafka topics and their partition counts.
Check the consumer group lag for the main topic to see if messages are being consumed.

## Step 2: Publish Test Events
Publish 3 test events to the main topic with the following structure:
  { "event_type": "system.health_check", "timestamp": "<current ISO timestamp>", "payload": { "status": "ok", "sequence": <1, 2, 3> } }
Publish them one at a time and confirm each delivery with the returned offset.

## Step 3: Report
Summarize queue activity:
- Topics found and their partition counts
- Consumer group lag before publishing
- Events published (topic, offset, timestamp for each)
- Delivery confirmation status

Create agent.yaml in the same directory to wire the skill to your connector:

name: Queue Publisher
description: Publish messages to Kafka topics and verify delivery
connectors:
  - connection_name: My Kafka
    connection_type: kafka
skills:
  - path: ./
    trigger_type: auto

The Config File

Create oe-config.json in the same directory:

{
  "llm": {
    "provider": "openai",
    "model": "gpt-4o",
    "apiKey": "YOUR_OPENAI_API_KEY"
  },
  "server": {
    "enabled": false,
    "port": 3333,
    "apiKey": "your-secret-api-key"
  },
  "connectors": [
    {
      "connection_name": "My Kafka",
      "connection_type": "kafka",
      "brokers": ["localhost:9092"],
      "clientId": "oe-runtime",
      "topic": "my-topic"
    }
  ]
}

Replace localhost:9092 with your Kafka broker address. For managed Kafka services like Confluent Cloud or AWS MSK, add SSL and SASL authentication fields to the connector config.

Download OE Runtime

Run the Agent

From the parent folder containing your skill directory:

1 npx recommended

npx -y @openenthrium/oe-runtime@latest ./message-queues

2 Windows

oe-runtime-win.exe ./message-queues

3 macOS

chmod +x oe-runtime-macos
./oe-runtime-macos ./message-queues

First run blocked? System Settings → Privacy & Security → Allow Anyway.

4 Linux

chmod +x oe-runtime-linux
./oe-runtime-linux ./message-queues

5 API Server integration

Add a "server" block to oe-config.json, then start with --serve:

{
  "llm": { ... },
  "server": { "enabled": true, "port": 3333, "apiKey": "your-secret-key" },
  "connectors": [ ... ]
}

npx -y @openenthrium/oe-runtime@latest --serve --config oe-config.json

Run with inline YAML:

curl -X POST http://localhost:3333/run \
  -H "Content-Type: application/json" \
  -H "X-API-Key: your-secret-key" \
  -d '{"yaml": "...", "params": {}}'

Or run from a file on the server:

curl -X POST http://localhost:3333/run-file \
  -H "Content-Type: application/json" \
  -H "X-API-Key: your-secret-key" \
  -d '{"file": "/path/to/agent.yaml", "params": {}}'

Use Cases

• Event-driven workflow triggering — Publish structured business events to Kafka topics to trigger downstream processing workflows automatically. Use the agent to translate plain English instructions into well-formed Kafka messages with the correct schema and metadata fields.
• Consumer lag monitoring — Check the consumer group lag across all topics to identify which consumers are falling behind and by how much. High lag indicates a processing bottleneck — the agent surfaces this in a readable report before it becomes a production incident.
• Data pipeline health checks — Verify that your data pipeline is functioning correctly by publishing test events and confirming they are consumed within an expected time window. This provides a simple end-to-end smoke test that can run on a schedule without any custom code.
• Microservices integration testing — Publish test events to your microservices' input topics and verify that the expected output events appear on the downstream topics. This lets you test event-driven integrations between services without writing dedicated test harnesses.
• Order processing systems — Publish order created, payment received, or shipment dispatched events to the relevant Kafka topics as part of an e-commerce workflow. The agent ensures each event is correctly structured and delivered before confirming the transaction to the caller.
• Log and telemetry aggregation — Publish application log entries, metrics, or tracing events to Kafka as a central aggregation point for your observability pipeline. The agent can format raw data into the structured JSON schema expected by your downstream consumers such as Elasticsearch or ClickHouse.
• Audit trail publishing — Emit structured audit events to a dedicated Kafka topic whenever sensitive actions occur in your system. The agent ensures every audit event includes the required fields — user ID, action, resource, and timestamp — before publishing.
• Scheduled batch event publishing — Run the agent on a schedule to publish a batch of events — daily summaries, weekly rollups, or scheduled notifications — to Kafka topics consumed by your reporting or notification services. This replaces cron jobs and custom scripts with a simple, readable YAML definition.