Kafka

Setup Instructions

1. 1

   Prepare the Kafka cluster

   Ensure the cluster is running and the source topic exists.

2. 2

   Create the connector

   Open the agent, click Add inbound connector in the Connector Flow, then Create New Connector and select Apache Kafka.

3. 3

   Configure and create

   Choose Inbound (Consumer) mode, enter Bootstrap Servers, Topic, and optionally Group ID, Auto Offset Reset, and security settings (SASL/SSL). Click Create and Connic starts consuming from the topic.

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

JSON objects can include a files array with each file's name, MIME type, and base64 data. Connic accepts up to 10 complete files, with 10 MiB per file and 10 MiB combined. Invalid optional files are omitted individually. At the first count or size overflow, accepted files remain; that file and all later files are omitted. file_issues records omissions with stable codes, including files omitted to fit the input-review budget. The run can continue even if no optional files remain. Enable Files required under Linked Agents when the task must stop if any file is omitted or no files are present.

End-to-End Example

Create an inbound connector for the 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

1. 1

   Prepare the Kafka cluster

   Ensure the cluster is running and the target topic exists (or enable auto-creation on the broker).

2. 2

   Create the connector

   Open the agent, click Add outbound connector, then Create New Connector and select Apache Kafka.

3. 3

   Configure and create

   Choose Outbound (Producer) mode, enter Bootstrap Servers and Topic. Click Create and results from linked agents will be published to the topic.

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