Kafka

Setup Instructions

How Inbound Works

Inbound Kafka connectors act as consumers that subscribe to a topic. When a message arrives, it's parsed and dispatched to all linked agents. Best for event-driven processing, real-time pipelines, streaming analytics, and microservices communication.

{
  "order_id": "12345",
  "customer": "john@example.com",
  "items": ["widget-a", "widget-b"],
  "_kafka": {
    "topic": "orders",
    "partition": 0,
    "offset": 1542,
    "timestamp": 1705312800000,
    "key": "order-12345"
  }
}

The _kafka metadata includes: topic, partition, offset, timestamp, and key.

JSON object messages are dispatched with their top-level fields as shown above. Anything else is wrapped under a message key: non-JSON values arrive as {"message": "<raw text>", "_kafka": ...}, and null-value messages (compaction tombstones) still trigger runs with message: null. Use _kafka.key to identify the deleted entity:

{
  "message": null,
  "_kafka": {
    "topic": "orders",
    "partition": 0,
    "offset": 1543,
    "timestamp": 1705312800000,
    "key": "order-12345"
  }
}

End-to-End Example

Create an inbound connector for your source topic, link it to an agent, and link an outbound connector to publish the agent output to a destination topic.

version: "1.0"

name: order-processor
type: llm
model: connic/gpt-5.6-luna
description: "Validate orders and compute routing"
system_prompt: |
  You receive an order event in JSON (from Kafka).
  1) Call orders.validate_order
  2) Call orders.score_risk
  3) Return JSON with order_id, status, risk_score, route

tools:
  - orders.validate_order
  - orders.score_risk
output_schema: order-result.json

from typing import Dict, Any

def validate_order(order_id: str, items: list[str]) -> Dict[str, Any]:
    """Basic order validation."""
    if not order_id or not items:
        return {"ok": False, "reason": "missing_fields"}
    return {"ok": True}

async def score_risk(customer: str, total: float) -> Dict[str, Any]:
    """Return a simple risk score and routing hint."""
    score = 0.02 if total < 100 else 0.12
    route = "standard" if score < 0.1 else "manual_review"
    return {"risk_score": score, "route": route}

The agent's JSON output appears in the output field of the outbound run envelope:

{
  "order_id": "12345",
  "status": "approved",
  "risk_score": 0.02,
  "route": "standard"
}

Each connector uses a consumer group to track processed messages. Different group IDs = same messages to all. Same group ID = load sharing.

• Reconnects with exponential backoff and commits offsets after successful handling

Setup Instructions

How Outbound Works

Automatic outbound connectors publish completed-run envelopes to the configured topic and can be limited to selected inputs. Agent-tool and middleware outbound connectors publish only when called.

Only runs with status completed are published. Failed and cancelled runs are skipped. Runs ended early via StopProcessing complete normally, so they are published too, with the StopProcessing response as the output. Raise it with publish_outbound=False to skip the automatic outbound connector for that run. It does not undo an agent-tool or middleware outbound connector call.

{
  "run_id": "550e8400-e29b-41d4-a716-446655440000",
  "agent_name": "order-processor",
  "status": "completed",
  "output": "Order processed successfully. Total: $234.56",
  "error": null,
  "started_at": "2024-01-15T10:30:00Z",
  "ended_at": "2024-01-15T10:30:05Z",
  "token_usage": {
    "input_tokens": 150,
    "output_tokens": 50,
    "thinking_tokens": 0,
    "cached_input_tokens": 0,
    "total_tokens": 200
  }
}

Includes run_id, agent_name, status, output, error, timestamps, and token_usage.

Agent-tool and Middleware Outbound Connectors

An agent-tool outbound connector exposes an editable tool name, defaulting to send_to_<connector_name>. Call a middleware outbound connector by its configured name through send_connector. Both use this connector-owned payload schema:

{
  "payload": {
    "order_id": "12345",
    "status": "approved"
  },
  "key": "order-12345"
}

payload becomes the Kafka message value. key is optional. Connic applies the configured topic, connection, serialization, security, retries, and Bridge routing without exposing credentials to the model.

For an automatic outbound connector, a run triggered by inbound Kafka reuses the source message key; other runs use the run ID. Agent-tool and middleware outbound connectors may provide key explicitly.

# Inbound message with key "order-123"
Message received → Agent triggered → Run completes

# Outbound message uses same key "order-123"
Result published with key: "order-123" (source: original)

# If no inbound key, uses run_id
Result published with key: "550e8400-e29b..." (source: run_id)

• Full replication acknowledgment and automatic retries