MQTT Stream
Create or update a recurring MQTT stream connector
Custom Streaming
Create or update a recurring MQTT stream connector. Subscribes to MQTT topics and streams messages to Kafka. Use this for IoT sensors, telemetry feeds, and any MQTT broker.
Technical details
Authentication ▾
Contact us for setup details.
Ingestion ▾
Mode: Streaming
Tools exposed (1) ▾
upsert_bento_stream_mqtt
Create or update a recurring MQTT stream connector. Subscribes to MQTT topics and streams messages to Kafka. Use this for IoT sensors, telemetry feeds, and any MQTT broker.
Inputs ▾
upsert_bento_stream_mqtt
| Name | Type | Required | Description |
|---|---|---|---|
| dashboardId | string | no | Target dashboard/initiative ID. Omit to use the current dashboard. |
| brokerUrls | string[] | yes | MQTT broker URLs, e.g. ["tcp://broker.example.com:1883", "ssl://secure-broker.example.com:8883"]. |
| topics | string[] | yes | MQTT topics to subscribe to, e.g. ["sensors/#", "telemetry/+/data"]. Supports MQTT wildcards. |
| qos | number | no | Quality of Service level: 0 (at most once), 1 (at least once, default), 2 (exactly once). |
| clientId | string | no | MQTT client ID. If omitted, auto-generated as 'delphi-<connectorId>'. |
| username | string | no | MQTT broker username for authentication. |
| passwordEnvVar | string | no | Name of the environment variable containing the MQTT password. Never pass the password directly. |
| recordPath | string | no | Dot path to extract from each message JSON, e.g. 'payload.readings'. |
| connectorId | string | no | If provided, updates the existing connector instead of creating a new one. |
| description | string | no | Human-readable description of what this MQTT connector monitors. |
| idExpression | string | no | Optional Bloblang expression for the Firestore document ID. Provide this when messages have a stable unique field. |
| tlsEnabled | boolean | no | Enable TLS for the MQTT connection. |
| tlsSkipVerify | boolean | no | Skip TLS certificate verification. Only use for development/testing. |
| keepAlive | number | no | Keepalive interval in seconds. Default 30. |
| cleanSession | boolean | no | Start a clean MQTT session. Default true. |