# MQTT triggers

> How do I trigger scripts and flows from MQTT broker messages in Windmill?

Windmill can connect to an [**MQTT**](https://mqtt.org/mqtt-specification/) broker and trigger runnables (scripts, flows) in response to messages published to specified topics. MQTT triggers are not available on the [Cloud](/pricing).

## How to use

### Configure MQTT resource
- Select an existing [MQTT resource](https://hub.windmill.dev/resource_types/225/mqtt) or create a new one
- Provide broker hostname and port
- Add authentication credentials and certificates as required by your broker

### Select runnable
- Choose the script or flow to execute when messages are published to your subscribed topics

### Configure topic subscriptions
- Specify one or more topics to subscribe to
- Set appropriate QoS level for each topic

#### Quality of Service (QoS) levels

| Level | Description | When to use |
|-------|-------------|-------------|
| **0** | **At most once** – Message delivered once or not at all without confirmation | Choose when it is okay for your script/flow to not be triggered (if the message is lost) or triggered only once. |
| **1** | **At least once** – Guaranteed delivery but may arrive multiple times | Choose when it is okay for your script/flow to be triggered again by an already received message from the broker. |
| **2** | **Exactly once** – Guaranteed delivery exactly once | Choose when you need your script/flow to be triggered only once and avoid any duplicates. |

For more information about MQTT QoS, see the [MQTT QoS Documentation](https://www.hivemq.com/blog/mqtt-essentials-part-6-mqtt-quality-of-service-levels/).

#### MQTT topic structure

MQTT topics are case-sensitive and follow a hierarchical structure (e.g., `home/sensor/temperature`).
For best practices on MQTT topics, see the [MQTT Topics Documentation](https://www.hivemq.com/blog/mqtt-essentials-part-5-mqtt-topics-best-practices/).

### Advanced MQTT options

By default, Windmill uses **MQTT version 5**. However, you can choose to use **MQTT version 3** or **MQTT version 5** with specific associated options.

- **MQTT v3 options**:
  - **Clean Session** (default: true): [Learn more](https://www.emqx.com/en/blog/mqtt5-new-feature-clean-start-and-session-expiry-interval#clean-session-in-mqtt-3-1-1)
  - **Client ID**: [Learn more](https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html)

- **MQTT v5 options**:
  - **Clean Start** (default: true): [Learn more](https://www.emqx.com/en/blog/mqtt5-new-feature-clean-start-and-session-expiry-interval#introduction-to-clean-start)
  - **Session Expiry Interval**: [Learn more](https://www.emqx.com/en/blog/mqtt5-new-feature-clean-start-and-session-expiry-interval#introduction-to-session-expiry-interval)
  - **Topic Alias Maximum**: [Learn more](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901051)
  - **Client ID**: [Learn more](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901059)

## Implementation examples

Below are code examples demonstrating how to handle MQTT messages in your Windmill scripts. You can either process messages directly in a basic script or use a preprocessor for more advanced message handling and transformation before execution.

### Basic script

```typescript

  // Parse JSON if applicable
  try {
    const jsonData = JSON.parse(textPayload);
    console.log("Received JSON data:", jsonData);
    // Process JSON data
  } catch (e) {
    // Handle as plain text
    console.log("Received text data:", textPayload);
    // Process text data
  }

  return { processed: true, message: "MQTT message processed successfully" };
}
```

### Script with preprocessor

If you use a [preprocessor](../../core_concepts/43_preprocessors/index.mdx), the preprocessor function receives the message payload as base64 encoded string and an MQTT object with the following fields:

#### MQTT object

- **`topic`**: The MQTT topic on which the message was received.
- **`retain`**: Boolean indicating if the message is retained.
- **`pkid`**: Packet identifier (if QoS > 0).
- **`qos`**: Quality of Service level.
- **`v5`**: MQTT v5 properties (optional).

#### MQTT v5 properties

- **`payload_format_indicator`**: Indicates if the payload is UTF-8 encoded or binary.
- **`topic_alias`**: An alias for the topic name.
- **`response_topic`**: A topic for the recipient to send a response to.
- **`correlation_data`**: Correlation data for request/response.
- **`user_properties`**: A list of user-defined properties.
- **`subscription_identifiers`**: Subscription identifiers.
- **`content_type`**: The content type of the payload.

For more information about MQTT v5 properties, see the [MQTT v5 Properties Documentation](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901109).

```typescript
/**
 * General Trigger Preprocessor
 *
 * ⚠️ This function runs BEFORE the main function.
 *
 * It processes raw trigger data (e.g., MQTT, HTTP, SQS) before passing it to `main()`.
 * Common tasks:
 * - Convert binary payloads to string/JSON
 * - Extract metadata
 * - Filter messages
 * - Add timestamps/context
 *
 * The returned object determines `main()` parameters:
 * - `{a: 1, b: 2}` → `main(a, b)`
 * - `{payload}` → `main(payload)`
 *
 * @param event - Trigger data and metadata (e.g., MQTT, HTTP)
 * @returns Processed data for `main()`
 */

    const uint8Payload = Uint8Array.from(payloadAsString, (c) => c.charCodeAt(0));

    return {
      contentType: event.v5?.content_type,
      payload: uint8Payload,
      payloadAsString
    };
  }
  // We assume the script is triggered by an MQTT message, which is why an error is thrown for other trigger kinds.
  // If the script is intended to support other triggers, update this logic to handle the respective trigger kind.
  throw new Error(`Expected mqtt trigger kind got: ${event.kind}`)
}

/**
 * Main Function - Handles processed trigger events
 *
 * ⚠️ Called AFTER `preprocessor()`, with its return values.
 *
 * @param payload - Raw binary payload
 * @param payloadAsString - Decoded string payload
 * @param contentType - MQTT v5 content type (if available)
 */
export async function main(payload: Uint8Array, payloadAsString: string, contentType?: string) {

}
```

## Error handling

MQTT triggers support local error handlers that override workspace error handlers for specific triggers. See the [error handling documentation](../../core_concepts/10_error_handling/index.mdx#trigger-error-handlers) for configuration details and examples.