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 available on Orvanta Cloud.

How to use

Configure MQTT resource

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