Kafka
Connects to an Apache Kafka cluster for producing and consuming messages, and managing topics. An agent can publish and poll messages, list/create/delete topics, and check consumer group lag.
Add to an agent node's tools array with _node_type: "tool-kafka". See Connector Reference — How connectors are used.
Authentication
| Field | Required | Description |
|---|---|---|
brokers | Yes (or broker) | Comma-separated list of host:port Kafka broker addresses. |
broker | Fallback | A single host:port broker address, used if brokers is empty. |
username | No | SASL username (currently accepted but not yet wired into the client connection). |
password | No | SASL password (currently accepted but not yet wired into the client connection). |
On setup, the connector performs a fast TCP probe against the first broker and fails if it cannot be reached.
{
"name": "kafka",
"_node_type": "tool-kafka",
"brokers": "broker1.example.com:9092,broker2.example.com:9092"
}
Tools
kafka_produce
Produce (publish) a message to a Kafka topic.
Parameters
| Parameter | Type | Required | Description |
|---|---|---|---|
topic | string | Yes | Kafka topic name |
key | string | No | Message key (optional) |
value | string | Yes | Message value (string or JSON) |
headers | object | No | Optional message headers (string key-value map) |
Returns: A confirmation message, e.g. "published <N> bytes to <topic>".
kafka_consume
Consume up to N messages from a Kafka topic (poll mode).
Parameters
| Parameter | Type | Required | Description |
|---|---|---|---|
topic | string | Yes | Kafka topic to consume from |
group_id | string | No | Consumer group ID |
limit | number | No | Maximum messages to read (default 10) |
offset | string | No | Where to start: earliest or latest (default latest) |
Returns: A JSON object string with messages (array of {key, value, offset, partition, time}) and count. Polling stops after limit messages or a 10-second deadline, whichever comes first.
kafka_list_topics
List all topics on the Kafka cluster.
No parameters.
Returns: A JSON object string with topics (array of topic names) and count.
kafka_create_topic
Create a new Kafka topic.
Parameters
| Parameter | Type | Required | Description |
|---|---|---|---|
topic | string | Yes | Topic name to create |
partitions | number | No | Number of partitions (default 1) |
replicas | number | No | Replication factor (default 1) |
Returns: A confirmation message, e.g. "created topic <topic> with <N> partitions".
kafka_delete_topic
Delete a Kafka topic.
Parameters
| Parameter | Type | Required | Description |
|---|---|---|---|
topic | string | Yes | Topic name to delete |
Returns: A confirmation message, e.g. "deleted topic <topic>".
kafka_lag
Get consumer group lag (messages behind) for a topic.
Parameters
| Parameter | Type | Required | Description |
|---|---|---|---|
topic | string | Yes | Kafka topic name |
group_id | string | Yes | Consumer group ID to check lag for |
Returns: A JSON object string with topic, group_id, lag, messages, and bytes reader statistics.