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.

The Agent YAML

Create a file called agent.yaml inside a message-queues/ directory:

name: Queue Publisher
description: Publish messages to Kafka topics and verify delivery
instructions: |
  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.
steps:
  - name: Check Topic Status
    content: |
      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.
  - name: Publish Test Events
    content: |
      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.
  - name: Report
    content: |
      Summarize queue activity:
      - Topics found and their partition counts
      - Consumer group lag before publishing
      - Events published (topic, offset, timestamp for each)
      - Delivery confirmation status
connectors:
  - connection_name: My Kafka
    connection_type: kafka

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

Windows

# Run the agent
oe-runtime-win.exe message-queues/agent.yaml --config message-queues/oe-config.json

macOS

# Make executable (first time only)
chmod +x oe-runtime-macos

# Run the agent
oe-runtime-macos message-queues/agent.yaml --config message-queues/oe-config.json

Linux

# Make executable (first time only)
chmod +x oe-runtime-linux

# Run the agent
oe-runtime-linux message-queues/agent.yaml --config message-queues/oe-config.json

Running as an API Server

Add --serve to start OE Runtime as an HTTP server on port 3333:

oe-runtime-win.exe --serve --config message-queues/oe-config.json

Then send a request:

curl -X POST http://localhost:3333/run \
  -H "Content-Type: application/json" \
  -d '{"yaml": "message-queues/agent.yaml", "params": {}}'

Or import the Postman collection (download above) to test all endpoints visually.

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.