Zum Hauptinhalt springen
connic

Apache Kafka

Von connic
InboundOutbound

Jede Nachricht in einem Topic startet einen Run, und Ergebnisse gehen mit demselben Key zurück in ein Topic. connic betreibt die Consumer Group und committet die Offsets für dich.

Zu einem Agenten hinzufügenSprich mit Sales
Publisher
Von connic
Modi
Inbound, Outbound
Einrichtung
Dashboard oder connectors/*.yaml
Dokumentation
Apache Kafka Docs

Mach aus jeder Nachricht einen Run

Eine eingehende Verbindung konsumiert ein Topic und startet jeden verknüpften Agenten einmal pro Nachricht, mit der Nachricht als Eingabe.

JSON bleibt, wie es ist
Ein JSON-Objekt kommt mit seinen Feldern auf oberster Ebene an, ergänzt um _kafka mit Topic, Partition, Offset, Timestamp und Key.
Text und Tombstones
Jeder andere Wert landet unter message. Ein Tombstone startet trotzdem einen Run mit message auf null, damit der Agent auf den gelöschten Key reagieren kann.
Dateien in der Nachricht
Ein JSON-Objekt kann ein Array files mit Name, MIME-Typ und base64-Daten jeder Datei enthalten, bis zu 10 Dateien und insgesamt 10 MiB.
message on orders
{
  "order_id": "ord_8812",
  "customer": "jana@example.com",
  "total": 184.5
}
context["payload"]
{
  "order_id": "ord_8812",
  "customer": "jana@example.com",
  "total": 184.5,
  "_kafka": {
    "topic": "orders",
    "partition": 3,
    "offset": 20417,
    "timestamp": 1791882000000,
    "key": "ord_8812"
  }
}
context["payload"] for a tombstone
{
  "message": null,
  "_kafka": {
    "topic": "customers",
    "partition": 0,
    "offset": 9182,
    "timestamp": 1791882000000,
    "key": "cus_4410"
  }
}

Committe jeden Offset erst, wenn seine Runs gespeichert sind

connic committet eine Nachricht, sobald für jeden verknüpften Agenten ein Run in der Warteschlange gespeichert ist. So geht bei einem Neustart oder Deployment keine Arbeit verloren.

Ein Run pro Nachricht
Kommt eine Nachricht nach einem Neustart oder Rebalance erneut an, nutzt connic den gespeicherten Run. Jeder Agent läuft einmal pro Nachricht und Consumer Group.
Fehlgeschlagene Runs bleiben sichtbar
Schlägt ein gespeicherter Run fehl, wird die Nachricht nicht erneut zugestellt. Der Run behält seine vollständige Eingabe in Agent Runs, wo du ihn neu starten kannst.
Parallele Consumer
Mehrere eingehende Verbindungen mit derselben Group ID teilen die Partitionen eines stark genutzten Topics unter sich auf. Verbindungen mit unterschiedlichen Group IDs lesen jeweils jede Nachricht.

  1. Lesen

    Die Verbindung liest ihre Partitionen Nachricht für Nachricht.

  2. Speichern

    connic speichert für jeden verknüpften Agenten einen Run in der Warteschlange. Gespeicherte Runs überstehen Neustarts von Workern und Deployments.

  3. Committen

    Der Offset wird committet. Schlägt das Speichern fehl, kommt dieselbe Nachricht nach kurzer Pause erneut, und spätere Nachrichten warten.

    Offset committet

  4. Ausführen

    Runs laufen parallel bis max_concurrent_runs. Eine Concurrency-Regel auf dem Message Key startet Runs mit demselben Key nacheinander.

    concurrency.key: _kafka.key

Halte den Consumer neben deinen Agenten

Deklariere ihn als YAML-Datei in connectors/. Broker und Zugangsdaten liegen in einer Connection, deshalb nennt die Datei nur Topic und Consumer Group. Ein im Code deklarierter Consumer beginnt zu lesen, sobald sein Deployment live geht.

SASL und TLS
Nutze PLAINTEXT, SSL, SASL_PLAINTEXT oder SASL_SSL mit PLAIN, SCRAM-SHA-256 oder SCRAM-SHA-512. Für Mutual TLS fügst du Zertifikat und Schlüssel im PEM-Format in die Connection ein.
Private Broker
Direkte Verbindungen brauchen Broker, die zu öffentlichen IP-Adressen auflösen. Broker in deinem Netzwerk erreichst du über eine Bridge.
connectors/order-events.yaml
type: kafka
mode: inbound
name: Order events
connection: orders-kafka
config:
  topic: orders
  group_id: connic-orders
  auto_offset_reset: earliest
agents:
  - order-processor
create: connectors/order-events.yaml (kafka)

Veröffentliche Ergebnisse wieder in einem Topic

Eine ausgehende Verbindung schreibt jeden abgeschlossenen Run in ein Topic, mit dem Key der Nachricht, die ihn gestartet hat.

Keys bleiben erhalten
Ein Run, den eine Kafka-Nachricht gestartet hat, wird mit deren Key veröffentlicht. Andere Runs nutzen die Run-ID als Key.
Nur abgeschlossene Runs
Fehlgeschlagene und abgebrochene Runs werden übersprungen. connic produziert mit acks=all und automatischen Wiederholungen.
Der Agent veröffentlicht selbst
Als Agent-Tool verknüpft, lässt die Verbindung das Modell während des Runs eine eigene Payload mit optionalem Key veröffentlichen.
connectors/order-results.yaml
type: kafka
mode: outbound
name: Order results
connection: orders-kafka
config:
  topic: order-results
agents:
  - name: order-processor
    outbound:
      usage: automatic
      source: [order-events]
message on order-results (key ord_8812)
{
  "run_id": "550e8400-e29b-41d4-a716-446655440000",
  "agent_name": "order-processor",
  "status": "completed",
  "output": "ord_8812 approved for standard shipping.",
  "error": null,
  "started_at": "2026-10-13T09:00:01Z",
  "ended_at": "2026-10-13T09:00:05Z",
  "token_usage": {
    "input_tokens": 642,
    "output_tokens": 38,
    "thinking_tokens": 0,
    "cached_input_tokens": 0,
    "total_tokens": 680
  }
}

Wo Kafka passt

Wähle ein Szenario und sieh, welches Topic den Run startet, was der Agent bekommt und wohin das Ergebnis geht.

Jedes Bestell-Event startet einen Run, und das Ergebnis landet unter dem Key der Bestellung in order-results.

Checkout-Service
Kafka Consumer
order-processor
Ergebnis in order-results
context["payload"]
{
  "order_id": "ord_8812",
  "customer": "jana@example.com",
  "total": 184.5,
  "_kafka": {
    "topic": "orders",
    "partition": 3,
    "offset": 20417,
    "timestamp": 1791882000000,
    "key": "ord_8812"
  }
}
order-results.json
{
  "run_id": "550e8400-e29b-41d4-a716-446655440000",
  "agent_name": "order-processor",
  "status": "completed",
  "output": "ord_8812 approved for standard shipping.",
  "error": null,
  "started_at": "2026-10-13T09:00:01Z",
  "ended_at": "2026-10-13T09:00:05Z",
  "token_usage": {
    "input_tokens": 642,
    "output_tokens": 38,
    "thinking_tokens": 0,
    "cached_input_tokens": 0,
    "total_tokens": 680
  }
}

Der Agent bewertet jede Transaktion und schreibt nur bei hohem Risiko in fraud-alerts.

Zahlungsservice
Kafka Consumer
fraud-scorer
Warnung in fraud-alerts
context["payload"]
{
  "transaction_id": "txn_30571",
  "customer_id": "cus_4410",
  "amount": 950.0,
  "country": "PT",
  "_kafka": {
    "topic": "transactions",
    "partition": 1,
    "offset": 88213,
    "timestamp": 1791882000000,
    "key": "cus_4410"
  }
}
connector-payload.json
{
  "payload": {
    "transaction_id": "txn_30571",
    "risk": "high",
    "reason": "Card used in two countries within 10 minutes."
  },
  "key": "cus_4410"
}

Messwerte aus der Fertigung kommen als Stream an, und der Agent meldet sich in Slack, sobald eine Maschine ihre Grenzwerte überschreitet.

Sensoren in der Fertigung
Kafka Consumer
sensor-monitor
Warnung in Slack
context["payload"]
{
  "sensor_id": "press-07",
  "temperature_c": 94.2,
  "limit_c": 85,
  "_kafka": {
    "topic": "sensor-readings",
    "partition": 5,
    "offset": 513340,
    "timestamp": 1791882000000,
    "key": "press-07"
  }
}
connector-payload.json
{
  "text": "press-07 at 94.2 °C, 9.2 °C over its limit.",
  "channel_id": "C01234567"
}

Ein Tombstone im kompaktierten Topic customers startet einen Run, und der Agent löscht die Daten dieses Kunden über seine Tools.

Kundenservice
Kafka Consumer
privacy-cleanup
Daten über Tools gelöscht
context["payload"]
{
  "message": null,
  "_kafka": {
    "topic": "customers",
    "partition": 0,
    "offset": 9182,
    "timestamp": 1791882000000,
    "key": "cus_4410"
  }
}

FAQ zu Apache Kafka

Füge eine Apache-Kafka-Verbindung im Modus Inbound (Consumer) hinzu, wähle eine Connection mit deinen Brokern, gib das Topic ein und verknüpfe den Agenten. Im Dashboard erstellte Verbindungen starten pausiert, klicke also auf Resume, sobald alles bereit ist. Danach startet jede Nachricht einen Run für jeden verknüpften Agenten, mit Kafka-Metadaten unter _kafka. Ein vollständiges Beispiel zeigt die Kafka-Pipeline-Anleitung.

PLAINTEXT, SSL, SASL_PLAINTEXT und SASL_SSL, mit PLAIN, SCRAM-SHA-256 oder SCRAM-SHA-512 für SASL. Mutual TLS funktioniert mit CA, Client-Zertifikat und Client-Schlüssel im PEM-Format. Broker in einem privaten Netzwerk erreichst du über eine Bridge.

Nein. Jeder Agent bekommt einen Run pro Nachricht und Consumer Group, auch über Verbindungen mit derselben Group ID hinweg. Kommt eine Nachricht nach einem Neustart oder Rebalance erneut an, nutzt connic den bereits gespeicherten Run. Ein fehlgeschlagener Run wird durch erneute Zustellung nicht wiederholt, starte ihn deshalb in Agent Runs neu.

Runs laufen standardmäßig parallel. Füge dem Agenten eine Concurrency-Regel mit key: _kafka.key und on_conflict: queue hinzu, dann starten Runs mit demselben Key nacheinander in der Reihenfolge ihres Eingangs. Nachrichten ohne Key werden nicht serialisiert.

Weitere Verbindungen

Zum Marketplace