How to Trigger AI Agents from Kafka Topics

Transactions, orders, and telemetry already flow through Kafka. The missing piece is an AI agent that reacts to them: scores the transaction, triages the alert, updates the recommendation. Connic does not require a separate consumer service for that. A Kafka inbound connector points at the topic, and every message starts an agent run with the payload attached. This tutorial walks through the whole setup, from scaffolding a project to watching the first run.

Why Kafka Is a Strong Agent Trigger

Kafka is a durable, ordered log. Triggering agents from it provides at-least-once delivery, ordering within a partition, and the ability to replay history by resetting a consumer to an earlier offset, guarantees a plain HTTP webhook cannot make. The connector joins the cluster as a consumer, so there is no polling loop, no middleware service, and no bot process to keep alive between Kafka and the agents. Our trigger comparison covers webhook, Kafka, Postgres, and SQS. This post focuses on the implementation with Kafka.

1. 1

   Scaffold a Project from the Template

   Prerequisites

   • Python 3.10+ installed
   • A reachable Kafka cluster with a topic to consume (this tutorial uses transactions)
   • A Connic project connected to a Git repository
   • The template pins an Anthropic model, so either add an Anthropic key in Project Settings or switch the agent YAML to an exact connic/* model and use Project credit

   The fastest follow-along path is the Kafka Fraud Detector template from the Connic Marketplace: a working project that consumes transactions from a Kafka topic and scores each one for fraud risk, with an escalation agent for high-risk cases. Install the SDK and scaffold from it:

   terminal

   Copy

   pip install connic-composer-sdk
   connic init my-project --templates=kafka-fraud-detector
   cd my-project

   This creates a complete, deployable structure:

   file tree

   Copy

   my-project/
     agents/
       kafka-fraud-detector/
         fraud-scorer.yaml
         fraud-escalator.yaml
     tools/
       fraud_tools.py
     middleware/
       fraud-scorer.py
     hooks/
       fraud-scorer.py
     schemas/
       fraud-assessment.json
     tests/
     requirements.txt

   Two agents, plain Python tools, middleware that flags admin requests, a hook that enforces the admin override, an output schema, and a test suite that runs with connic test. The Kafka Fraud Detector template includes the architecture diagram. Other Marketplace agent templates cover use cases beyond fraud scoring. Everything below works the same for an agent written from scratch.

2. 2

   Look at the Agent YAML

   Each YAML file under agents/ defines one agent. Here is the scorer from the template, with the system prompt shortened:

   agents/kafka-fraud-detector/fraud-scorer.yaml

   Copy

   version: "1.0"
   
   name: fraud-scorer
   type: llm
   model: connic/gpt-5.6-terra
   description: "Scores transactions for fraud risk in real-time from a Kafka stream"
   
   system_prompt: |
     You are a fraud detection specialist analyzing financial transactions in
     real-time from a Kafka stream.
   
     Use _kafka.key for customer correlation and _kafka.timestamp to detect
     processing latency.
   
   temperature: 0.1
   output_schema: fraud-assessment
   
   concurrency:
     key: "data.customer_id"
     on_conflict: queue
   
   tools:
     - fraud_tools.calculate_velocity
     - fraud_tools.check_geo_anomaly
     - fraud_tools.create_alert
     - fraud_tools.search_fraud_patterns
     - fraud_tools.store_fraud_pattern
     - fraud_tools.admin_override: context.is_admin == True

   For Kafka workloads, the prompt tells the model to use _kafka.key and _kafka.timestamp, metadata the connector attaches to every message. The concurrency block queues runs that share a customer_id, so two transactions from the same customer never race each other while the rest of the stream processes in parallel. And the last tool entry is conditional: admin_override only exists for the model when middleware has set is_admin in the run context. The agent configuration reference covers the full set of options.

3. 3

   Deploy

   Push the project to the environment's deployment branch, and Connic builds and deploys it automatically. Or deploy straight from the CLI:

   terminal

   Copy

   connic login
   connic deploy

   While iterating, connic dev provides a hot-reloading dev server for adjusting the prompt and tools before wiring up production traffic.

4. 4

   Create the Kafka Inbound Connector

   Connectors are created in the dashboard and linked to agents there. Open the fraud-scorer agent's detail page, click Add inbound connector on the Connector Flow, choose Create New Connector, and select Apache Kafka. For the tutorial setup:

   Setting	Value	Notes
   Mode	Inbound (Consumer)	Consumes messages and triggers runs
   Bootstrap servers	kafka:9092	Comma-separated broker addresses, reachable from Connic
   Topic	transactions	The topic to consume from
   Consumer Group ID	fraud-detector	Optional; auto-generated if left empty
   Auto offset reset	latest	Use earliest to replay from the beginning
   Security protocol	PLAINTEXT	Or SSL, SASL_PLAINTEXT, SASL_SSL
   SASL mechanism	PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512	Plus username and password, SASL only

   The connector consumes from our infrastructure, so the brokers must be reachable from Connic, directly or through Bridge: a managed endpoint works out of the box, and a private cluster connects through Bridge, covered at the end of this post. For managed Kafka such as Confluent Cloud, Amazon MSK, or Aiven, use SASL_SSL with SCRAM-SHA-256 and the service credentials. Mutual TLS is supported by pasting the CA certificate, client certificate, and client key PEM contents directly into the form; the connector does not accept local file paths for certificates. Once created, the connector is linked to the agent automatically and starts consuming.

   Consumer groups retain Kafka's standard behavior: connectors with different group IDs each receive every message, connectors sharing a group ID split the load. The connector tracks offsets, reconnects with exponential backoff, and picks up configuration changes without redeploying the agent.

   What the Agent Receives

   JSON messages arrive with their top-level fields intact, plus a _kafka block carrying topic, partition, offset, timestamp, and key:

   message-payload.json

   Copy

   {
     "transaction_id": "tx-9917",
     "customer_id": "cust-1042",
     "amount": 4899.0,
     "currency": "EUR",
     "merchant": "TechWorld Berlin",
     "country": "DE",
     "_kafka": {
       "topic": "transactions",
       "partition": 0,
       "offset": 1542,
       "timestamp": 1705312800000,
       "key": "cust-1042"
     }
   }

   Non-JSON messages are wrapped under a message key, and compaction tombstones still trigger runs with message: null, so an agent can react to deletions via _kafka.key. The full payload behavior is in the Kafka connector reference.

5. 5

   Publish Results Back to a Topic

   Downstream systems usually want the agent's verdict on a stream of their own. Repeat the connector setup on the same fraud-scorer agent, this time choosing Outbound (Producer) mode with fraud-alerts as the topic. Every time the agent completes a run, its structured assessment is published with the run ID, agent name, status, and output. Failed and cancelled runs are skipped. If the run was triggered by an inbound Kafka message with a key, the outbound message reuses the same key, so per-customer ordering survives the round trip through the agent.

6. 6

   Send a Test Event and Watch the Run

   Produce a keyed test message to the topic:

   terminal

   Copy

   kafka-console-producer \
     --bootstrap-server kafka:9092 \
     --topic transactions \
     --property parse.key=true \
     --property key.separator=:
   >cust-1042:{"transaction_id":"tx-9917","customer_id":"cust-1042","amount":4899.0,"currency":"EUR","merchant":"TechWorld Berlin","country":"DE"}

   Within moments a new run appears in the project's Runs section, with the full trace: the incoming payload, every tool call the scorer made, and the structured risk assessment it returned. The trace documentation explains how to read those traces. From here the loop is tight: adjust the prompt or tools, push, and the next message on the topic exercises the new version.

Private Kafka Clusters

Brokers on a private network do not need public exposure. With Connect via Bridge enabled and Connic Bridge running inside the private network, the connector uses a secure outbound tunnel.

The connector also works for content moderation, personalization, and anomaly detection streams. The Marketplace describes the Kafka connector capabilities.