MQTT Triggers

Orvanta can connect to an MQTT broker and trigger runnables (scripts, flows) in response to messages published to specified topics. MQTT triggers are not tier-gated: they are available on every tier.

How to use

Configure MQTT connection

• Select an existing MQTT connection 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 the 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.

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.

Advanced MQTT options

By default, Orvanta 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)
  • Client ID
• MQTT v5 options:
  • Clean Start (default: true)
  • Session Expiry Interval
  • Topic Alias Maximum
  • Client ID

Implementation examples

Below are code examples demonstrating how to handle MQTT messages in your Orvanta 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

export async function main(payload: string) {
  // Convert base64 encoded payload to string
  const textPayload = atob(payload);

  // 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, the preprocessor function receives the message payload as a 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.

/**
 * 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()`
 */
export async function preprocessor(
  event: {
    kind: "mqtt",
    payload: string, // base64 encoded payload
    topic: string,
    retain: boolean,
    pkid: number,
    qos: number,
    v5?: {
      payload_format_indicator?: number,
      topic_alias?: number,
      response_topic?: string,
      correlation_data?: Array<number>,
      user_properties?: Array<[string, string]>,
      subscription_identifiers?: Array<number>,
      content_type?: string
    }
  }) {
  if (event.kind === 'mqtt') {
    const payloadAsString = atob(event.payload);
    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 for configuration details and examples.