ACTION Operations Reference - Coreflux Documentation

Overview

Complete reference for all LoT operations available within actions. These operations allow you to manipulate variables, perform calculations, transform strings, read and write MQTT topics, and implement conditional logic.

Quick Reference: Use the sidebar to jump to specific operations.

Variable Operations

SET, GET TOPIC, GET JSON, and GET PROTO are the core operations for working with data in actions. They allow you to store values, read from MQTT topics, extract fields from JSON payloads, and decode protobuf fields.

SET

Creates or assigns a value to an internal variable:

SET "variable_name" WITH <value>

Variables exist only during action execution and are referenced using {variable_name}.

Static Value
From Topic
Calculated

SET "threshold" WITH 100
SET "status" WITH "active"
SET "enabled" WITH TRUE

SET "current_temp" WITH (GET TOPIC "sensors/temperature")
// For topic "sensors/temp001/data": POSITION 2 gets the sensor ID
SET "sensor_id" WITH TOPIC POSITION 2

SET "total" WITH (GET TOPIC "count/a" + GET TOPIC "count/b")
SET "average" WITH ({sum} / {count})

Variable names must be enclosed in double quotes when using SET, but referenced with curly braces: SET "myVar" WITH 10 then use {myVar}.

GET TOPIC

Retrieves the last known value from a topic. If the topic has no value, the broker returns a default based on the requested type.

GET TOPIC "topic/path"
GET TOPIC "topic/path" AS <TYPE>

Default Values for Missing Topics

Type	Default Value
`STRING`	`""` (empty string)
`DOUBLE`	`0.0`
`INT`	`0`
`BOOL`	`FALSE`

Type Casting

GET TOPIC "sensors/temp" AS DOUBLE
GET TOPIC "config/count" AS INT
GET TOPIC "flags/enabled" AS BOOL

Wildcard Inheritance

When an action is triggered by a wildcard topic pattern, GET TOPIC inherits the matched wildcard values:

DEFINE ACTION ProcessWithContext
ON TOPIC "sensors/+/temperature" DO
    // If triggered by "sensors/temp001/temperature"
    // The + wildcard matches "temp001"
    
    // This GET TOPIC uses the same wildcard value:
    SET "humidity" WITH (GET TOPIC "sensors/+/humidity" AS DOUBLE)
    // Reads from "sensors/temp001/humidity"

Example Usage

// Read and process
SET "last_temp" WITH (GET TOPIC "sensors/temperature" AS DOUBLE)

// Check before using (EMPTY check)
IF GET TOPIC "config/threshold" == EMPTY THEN
    SET "threshold" WITH 80
ELSE
    SET "threshold" WITH (GET TOPIC "config/threshold" AS DOUBLE)

// Combine multiple topic values
SET "total" WITH (GET TOPIC "counters/a" AS INT + GET TOPIC "counters/b" AS INT)

GET JSON

Extracts fields from JSON payloads with support for nested paths and arrays.

For narrative patterns, missing-field workflows, and end-to-end PUBLISH MODEL examples, see Working with JSON.

Basic Syntax

GET JSON "field_name" IN PAYLOAD AS <TYPE>

Supported Types

Type	Use For	Example
`STRING`	Text values	`GET JSON "name" IN PAYLOAD AS STRING`
`DOUBLE`	Decimal numbers	`GET JSON "temperature" IN PAYLOAD AS DOUBLE`
`INT`	Whole numbers	`GET JSON "count" IN PAYLOAD AS INT`
`BOOL`	True/false values	`GET JSON "enabled" IN PAYLOAD AS BOOL`

Nested Path Extraction

Use dot notation to access nested objects and bracket notation for arrays:

// Nested object access
SET "nested" WITH (GET JSON "data.sensors.temperature" IN PAYLOAD AS DOUBLE)
SET "deep" WITH (GET JSON "metadata.config.threshold" IN PAYLOAD AS INT)

// Array access (0-indexed)
SET "first_reading" WITH (GET JSON "readings[0].value" IN PAYLOAD AS DOUBLE)
SET "second_sensor" WITH (GET JSON "data.sensors[1].id" IN PAYLOAD AS STRING)

For a payload like:

{
  "data": {
    "sensors": [
      { "id": "temp001", "value": 23.5 },
      { "id": "temp002", "value": 24.1 }
    ]
  },
  "metadata": { "config": { "threshold": 80 } }
}

Error Handling

If a field doesn’t exist, the result depends on the requested type:

Type	Missing Field Result
`STRING`	`""` (empty string)
`DOUBLE`	`0.0`
`INT`	`0`
`BOOL`	`FALSE`

Use EMPTY check for conditional handling:

SET "value" WITH (GET JSON "optional_field" IN PAYLOAD AS DOUBLE)
IF {value} == EMPTY THEN
    SET "value" WITH 100  // Default fallback

Complete Example

DEFINE ACTION ProcessComplexJSON
ON TOPIC "sensors/data" DO
    // Top-level fields
    SET "sensor_id" WITH (GET JSON "id" IN PAYLOAD AS STRING)
    SET "timestamp" WITH (GET JSON "timestamp" IN PAYLOAD AS STRING)
    
    // Nested fields
    SET "temp" WITH (GET JSON "readings.temperature" IN PAYLOAD AS DOUBLE)
    SET "humidity" WITH (GET JSON "readings.humidity" IN PAYLOAD AS DOUBLE)
    
    // Array access
    SET "latest" WITH (GET JSON "history[0].value" IN PAYLOAD AS DOUBLE)
    
    PUBLISH TOPIC "processed/" + {sensor_id} + "/temp" WITH {temp}
    PUBLISH TOPIC "processed/" + {sensor_id} + "/humidity" WITH {humidity}

GET PROTO

Extracts fields from binary protobuf MQTT payloads. GET PROTOBUF is the same verb. You need a matching DEFINE MODEL with WITH FORMAT PROTOBUF (or BOTH) and PROTO_TAG values that follow the vendor .proto. Full models, type mapping, and pitfalls: Protobuf Models.

Basic Syntax

GET PROTO "field_name" IN PAYLOAD AS <TYPE> USING MODEL "ModelName"

Source	Bytes come from
`IN PAYLOAD`	The MQTT payload that fired the Action
`IN GET TOPIC "…"`	Last payload stored for that topic
`IN` a LoT variable	A Base64 string in that variable

USING MODEL names the top-level message, even on a nested dotted path. Supported AS types: INT, DOUBLE, FLOAT (alias SINGLE), STRING, BOOL, BYTES. A missing or mismatched field returns null (not a type default). That usually means a wire-type mismatch, for example ADD DOUBLE on a proto float. Declare the parent message (unused arms stay as ADD OBJECT so wire numbers do not shift):

DEFINE MODEL DeviceCommand WITH FORMAT PROTOBUF COLLAPSED
    ADD INT "command_type" PROTO_TAG 1
    ADD OBJECT "header" PROTO_TAG 2

Then read a field from the triggering payload:

DEFINE ACTION OnPlcCommand
ON TOPIC "plant/+/+/command" DO
    SET "cmd" WITH (GET PROTO "command_type" IN PAYLOAD AS INT USING MODEL "DeviceCommand")
    KEEP TOPIC "plant/pending_command" WITH {cmd}

A dotted path reads a nested message. Declare that arm with ADD MODEL, and keep USING MODEL on the top-level message:

DEFINE MODEL ReadSnapshot WITH FORMAT PROTOBUF COLLAPSED
    ADD STRING "tag_group" PROTO_TAG 1
    ADD INT "max_age_ms" PROTO_TAG 2

DEFINE MODEL DeviceCommand WITH FORMAT PROTOBUF COLLAPSED
    ADD INT "command_type" PROTO_TAG 1
    ADD MODEL ReadSnapshot "read_snapshot" PROTO_TAG 6

DEFINE ACTION OnPlcCommand
ON TOPIC "plant/+/+/command" DO
    SET "cmd" WITH (GET PROTO "command_type" IN PAYLOAD AS INT USING MODEL "DeviceCommand")
    SET "tag_group" WITH (GET PROTO "read_snapshot.tag_group" IN PAYLOAD AS STRING USING MODEL "DeviceCommand")
    KEEP TOPIC "plant/pending_group" WITH {tag_group}

GET MODEL

GET MODEL reads one record from a database connector and returns it shaped as a model: a JSON object with that model’s fields.

Basic Syntax

GET MODEL "<model_name>" FROM "<connector_name>" [WHERE <field> <operator> <value> [AND …]]

The query reads the table, collection, or index with the same name as the model, not the STORE IN table.
WHERE (or WHEN) filters on columns. Operators are EQUALS, ==, !=, >, <, >=, and <=. Join conditions with AND in any mix of text and numeric comparisons. Text values are single-quoted, and a literal ' is written '' (the name O'Brien is 'O''Brien'). A {var} value is resolved when the action runs. OR is not supported and is a syntax error.
The first matching record is returned. A model field the record doesn’t have gets its type’s default (0, 0.0, false, or "").
With no match, an unknown model, or a query error, the result is an empty object {}, and the error goes to the broker log.

Supported connectors: PostgreSQL, MySQL, MariaDB, SQL Server, MongoDB, OpenSearch, FILE_STORAGE, and REST_API.

Example Usage

Read the press row when its id, state, and temperature all match. {machine} is filled in when the action runs.

DEFINE MODEL MachineState
    ADD STRING "machine_id"
    ADD STRING "state"
    ADD DOUBLE "temperature"

DEFINE ACTION ReportPress1
ON TOPIC "plant/press1/report/request" DO
    SET "machine" WITH "press1"
    SET "press" WITH GET MODEL "MachineState" FROM "PlantDB" WHERE machine_id EQUALS {machine} AND state EQUALS 'running' AND temperature > 80
    PUBLISH TOPIC "plant/press1/report" WITH {press}
    PUBLISH TOPIC "plant/press1/hot" WITH GET MODEL "MachineState" FROM "PlantDB" WHERE temperature > 80

GET MODEL runs inside the action and waits for the database. Keep it off high-rate triggers. In a cluster, a connector that runs only on its owner node, such as FILE_STORAGE, answers there only: on another node the call fails and names the owner.

Environment Variables & Secrets

Version Requirement: GET ENV, GET SECRET, KEEP ENV, KEEP SECRET, DELETE ENV, and DELETE SECRET are available from Coreflux Broker v1.9.3 and above.

LoT Actions can read, write, and delete environment variables and encrypted secrets at runtime. Environment variables store non-sensitive configuration (hostnames, ports, URLs). Secrets store sensitive credentials encrypted with AES-256-GCM.

Use environment variables for configuration that can safely appear in logs (hostnames, ports, feature flags). Use secrets for anything sensitive (passwords, API keys, tokens).

GET ENV

Reads an environment variable. The broker checks the managed .env file first, then falls back to process, user, and machine environment variables. Returns an empty string if not found.

GET ENV "NAME"

Examples

-- Read connection settings
SET "host" WITH (GET ENV "DB_HOST")
SET "port" WITH (GET ENV "DB_PORT")
SET "apiUrl" WITH (GET ENV "API_BASE_URL")

-- Use in conditional logic
IF GET ENV "FEATURE_FLAG" == "enabled" THEN
    PUBLISH TOPIC "features/new" WITH "active"

KEEP ENV

Persists a new environment variable to the .env file on disk:

KEEP ENV "NAME" WITH <value>

-- Save a timestamp
KEEP ENV "LAST_RUN" WITH TIMESTAMP "UTC"

-- Save a computed value
KEEP ENV "DEVICE_COUNT" WITH {count}

DELETE ENV

Removes an environment variable from the managed .env file:

DELETE ENV "NAME"

DELETE ENV "OLD_CONFIG"

GET SECRET

Reads an encrypted secret from secrets.json, decrypting it at runtime. Returns an empty string if the secret does not exist.

GET SECRET "NAME"

Examples

-- Read credentials
SET "dbPass" WITH (GET SECRET "DB_PASSWORD")
SET "apiKey" WITH (GET SECRET "API_KEY")

-- Use in conditional logic
IF GET SECRET "LICENSE_KEY" == EMPTY THEN
    PUBLISH TOPIC "system/warnings" WITH "No license key configured"

KEEP SECRET

Encrypts and persists a new secret to secrets.json:

KEEP SECRET "NAME" WITH "value"

KEEP SECRET "API_KEY" WITH "sk-abc123"

DELETE SECRET

Removes a secret from secrets.json:

DELETE SECRET "NAME"

DELETE SECRET "OLD_API_KEY"

Complete Example

DEFINE ACTION SecureDataForwarder
ON TOPIC "sensors/+/data" DO
    SET "sensor_id" WITH TOPIC POSITION 2

    -- Read configuration from environment
    SET "apiUrl" WITH (GET ENV "API_BASE_URL")
    SET "region" WITH (GET ENV "DEPLOY_REGION")

    -- Read credentials from secrets
    SET "apiKey" WITH (GET SECRET "API_KEY")

    IF {apiUrl} == EMPTY THEN
        PUBLISH TOPIC "system/errors" WITH "API_BASE_URL not configured"
    ELSE
        -- Forward sensor data with region tag
        PUBLISH TOPIC "forwarded/" + {region} + "/" + {sensor_id} WITH PAYLOAD
        PUBLISH TOPIC "forwarded/" + {sensor_id} + "/timestamp" WITH TIMESTAMP "UTC"

    -- Track last forwarded sensor
    KEEP ENV "LAST_FORWARDED_SENSOR" WITH {sensor_id}

For full details on environment variable resolution order, encryption key management, and deployment configuration, see Environment Variables & Secrets.

Arithmetic Operators

LoT supports standard arithmetic operations that work with numeric types (INT, DOUBLE) and can be combined with GET TOPIC, PAYLOAD, and variables.

Supported Operators

Operator	Symbol	Description
Addition	`+`	Adds two values together
Subtraction	`-`	Subtracts right value from left
Multiplication	`*`	Multiplies two values
Division	`/`	Divides left value by right
Modulo	`%`	Returns remainder of division
Unary Negative	`-`	Negates a value

Syntax

<expression> + <expression>
<expression> - <expression>
<expression> * <expression>
<expression> / <expression>
<expression> % <expression>
-<expression>

Basic Examples

// Simple arithmetic with literals
SET "sum" WITH (10 + 5)
SET "difference" WITH (100 - 25)
SET "product" WITH (6 * 7)
SET "quotient" WITH (100 / 4)
SET "remainder" WITH (17 % 5)
SET "negative" WITH (-42)

Using with Variables

SET "a" WITH 100
SET "b" WITH 25
SET "result" WITH ({a} + {b})
SET "average" WITH (({a} + {b}) / 2)

Using with GET TOPIC

// Add values from two different topics
SET "total" WITH (GET TOPIC "sensors/count/a" AS INT + GET TOPIC "sensors/count/b" AS INT)

// Calculate percentage
SET "percentage" WITH (GET TOPIC "current/value" AS DOUBLE / GET TOPIC "max/value" AS DOUBLE * 100)

// Temperature conversion (Celsius to Fahrenheit)
SET "fahrenheit" WITH (GET TOPIC "sensors/temp/celsius" AS DOUBLE * 9 / 5 + 32)

Using with PAYLOAD

DEFINE ACTION CalculateWithPayload
ON TOPIC "input/value" DO
    SET "doubled" WITH (PAYLOAD AS DOUBLE * 2)
    SET "incremented" WITH (PAYLOAD AS INT + 1)
    PUBLISH TOPIC "output/doubled" WITH {doubled}

Operator Precedence

Operations follow standard mathematical precedence:

Parentheses () - highest priority
Unary negation -
Multiplication *, Division /, Modulo %
Addition +, Subtraction -

// Without parentheses: 2 + 3 * 4 = 14 (multiplication first)
SET "result1" WITH (2 + 3 * 4)

// With parentheses: (2 + 3) * 4 = 20
SET "result2" WITH ((2 + 3) * 4)

Complete Example

DEFINE ACTION TemperatureProcessor
ON TOPIC "sensors/+/temperature" DO
    // POSITION 1="sensors", POSITION 2=sensor_id (wildcard), POSITION 3="temperature"
    SET "sensor_id" WITH TOPIC POSITION 2
    SET "celsius" WITH PAYLOAD AS DOUBLE
    
    // Convert to Fahrenheit
    SET "fahrenheit" WITH ({celsius} * 9 / 5 + 32)
    
    // Calculate deviation from setpoint
    SET "setpoint" WITH (GET TOPIC "config/setpoint" AS DOUBLE)
    SET "deviation" WITH ({celsius} - {setpoint})
    SET "deviation_percent" WITH ({deviation} / {setpoint} * 100)
    
    PUBLISH TOPIC "processed/" + {sensor_id} + "/fahrenheit" WITH {fahrenheit}
    PUBLISH TOPIC "processed/" + {sensor_id} + "/deviation" WITH {deviation_percent}

String Concatenation

The + operator concatenates strings, allowing dynamic construction of topics, messages, and values.

Syntax

<string_expression> + <string_expression>

Basic Examples

SET "greeting" WITH ("Hello, " + "World!")
SET "fullname" WITH ({firstname} + " " + {lastname})

Dynamic Topic Construction

// Build topic paths dynamically
PUBLISH TOPIC "devices/" + {device_id} + "/status" WITH "online"
PUBLISH TOPIC "sensors/" + {sensor_type} + "/" + {location} + "/reading" WITH {value}

// Using with KEEP
KEEP TOPIC "cache/" + {sensor_id} + "/last_reading" WITH PAYLOAD

Combining Strings with Other Types

// String + Number (number is converted to string)
SET "message" WITH ("Temperature is: " + {temp})
PUBLISH TOPIC "alerts/temperature" WITH ("High temp: " + {celsius} + "°C")

Complete Example

DEFINE ACTION DynamicPublisher
ON TOPIC "input/+/+/data" DO
    // POSITION 1="input", POSITION 2=region, POSITION 3=device, POSITION 4="data"
    SET "region" WITH TOPIC POSITION 2
    SET "device" WITH TOPIC POSITION 3
    SET "timestamp" WITH TIMESTAMP "UTC"
    
    // Build dynamic output topic
    SET "output_topic" WITH ("processed/" + {region} + "/" + {device})
    
    // Build descriptive message
    SET "message" WITH ("Device " + {device} + " in " + {region} + " reported at " + {timestamp})
    
    PUBLISH TOPIC {output_topic} WITH {message}

RANDOM Function

Generates random values of various types including numeric ranges and unique identifiers.

Syntax

// Random integer in range (inclusive)
RANDOM BETWEEN <min> AND <max>
RANDOM INT BETWEEN <min> AND <max>

// Random double/float in range
RANDOM DOUBLE BETWEEN <min> AND <max>

// Unique identifiers
RANDOM UUID
RANDOM GUID
RANDOM ULID

Random Integers

// Random integer between 1 and 100
SET "dice_roll" WITH (RANDOM BETWEEN 1 AND 6)
SET "random_percent" WITH (RANDOM INT BETWEEN 0 AND 100)

// Using expressions for range
SET "random_in_range" WITH (RANDOM INT BETWEEN {min_value} AND {max_value})

Random Doubles

// Random double between 0.0 and 1.0
SET "probability" WITH (RANDOM DOUBLE BETWEEN 0.0 AND 1.0)

// Random temperature for simulation
SET "simulated_temp" WITH (RANDOM DOUBLE BETWEEN 18.5 AND 25.5)

Unique Identifiers

// Generate UUID (e.g., "550e8400-e29b-41d4-a716-446655440000")
SET "transaction_id" WITH (RANDOM UUID)

// Generate GUID (same format as UUID)
SET "session_id" WITH (RANDOM GUID)

// Generate ULID (e.g., "01ARZ3NDEKTSV4RRFFQ69G5FAV")
// ULIDs are sortable by time
SET "event_id" WITH (RANDOM ULID)

Use Cases

Simulation
Unique IDs
Load Balancing

Generate test data for sensors:

DEFINE ACTION SimulateSensor
ON EVERY 5 SECONDS DO
    SET "temperature" WITH (RANDOM DOUBLE BETWEEN 20.0 AND 30.0)
    SET "humidity" WITH (RANDOM INT BETWEEN 40 AND 80)
    SET "pressure" WITH (RANDOM DOUBLE BETWEEN 1000.0 AND 1020.0)
    
    PUBLISH TOPIC "simulated/sensor/temperature" WITH {temperature}
    PUBLISH TOPIC "simulated/sensor/humidity" WITH {humidity}
    PUBLISH TOPIC "simulated/sensor/pressure" WITH {pressure}

Generate unique record identifiers:

DEFINE ACTION CreateRecord
ON TOPIC "records/create" DO
    SET "record_id" WITH (RANDOM UUID)
    SET "timestamp" WITH TIMESTAMP "UTC"
    
    PUBLISH TOPIC "records/" + {record_id} WITH PAYLOAD
    PUBLISH TOPIC "records/index/latest" WITH {record_id}

Distribute work across workers:

DEFINE ACTION RandomRouter
ON TOPIC "incoming/requests" DO
    SET "worker" WITH (RANDOM INT BETWEEN 1 AND 4)
    PUBLISH TOPIC "workers/" + {worker} + "/queue" WITH PAYLOAD

String Transformations

LoT provides string transformation operations for filtering, replacing, and parsing text data.

FILTER

Extracts content from a string using a regular expression pattern.

Syntax

FILTER <source> USING REGEX "<pattern>"

Supported Sources

PAYLOAD - The incoming message payload
GET TOPIC "path" - Value from an MQTT topic
{variable} - A previously set variable
"string literal" - A hardcoded string

Examples

// Extract all digits from payload
SET "numbers" WITH FILTER PAYLOAD USING REGEX "[0-9]+"

// Extract from a topic value
SET "extracted" WITH FILTER (GET TOPIC "data/raw") USING REGEX "[a-zA-Z]+"

// Extract email pattern
SET "email" WITH FILTER PAYLOAD USING REGEX "[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}"

// Extract numbers from mixed content
SET "value" WITH FILTER "Temperature: 23.5°C" USING REGEX "[0-9]+\\.?[0-9]*"

Common Regex Patterns

Pattern	Description	Example Match
`[0-9]+`	One or more digits	”123”, “4567”
`[a-zA-Z]+`	One or more letters	”Hello”, “World”
`[0-9]+\\.?[0-9]*`	Decimal numbers	”23.5”, “100”
`[^\\s]+`	Non-whitespace	”word”
`\\w+`	Word characters	”test_123”

REPLACE

Replaces occurrences of a pattern with a new value.

Syntax

REPLACE "<find>" WITH "<replacement>" IN <source>

Examples

// Replace in payload
SET "cleaned" WITH REPLACE "old" WITH "new" IN PAYLOAD

// Replace in topic value
SET "modified" WITH REPLACE "error" WITH "warning" IN (GET TOPIC "logs/latest")

// Replace in variable
SET "formatted" WITH REPLACE " " WITH "_" IN {sensor_name}

// Remove characters by replacing with empty string
SET "digits_only" WITH REPLACE "[^0-9]" WITH "" IN {mixed_content}

Multiple Replacements

SET "step1" WITH REPLACE "," WITH ";" IN PAYLOAD
SET "step2" WITH REPLACE " " WITH "_" IN {step1}
SET "final" WITH REPLACE "'" WITH "" IN {step2}

SPLIT CSV

Parses CSV (Comma-Separated Values) data from the payload into a structured format.

Syntax

SPLIT PAYLOAD CSV [SEPARATOR "<char>"] [HEADERS PRESENT|NONE]

Parameters

Parameter	Description	Default
`SEPARATOR`	The delimiter character	`,` (comma)
`HEADERS PRESENT`	First row contains column names	-
`HEADERS NONE`	No header row	-

Example

DEFINE ACTION ProcessCSVData
ON TOPIC "import/csv" DO
    // Payload: "name,value,unit\nTemperature,23.5,C\nHumidity,65,percent"
    SET "parsed" WITH SPLIT PAYLOAD CSV SEPARATOR "," HEADERS PRESENT
    
    PUBLISH TOPIC "import/parsed" WITH {parsed}

SPLIT CSV currently only supports PAYLOAD as input. Support for GET TOPIC and variables is planned for future releases.

Topic Operations

PUBLISH vs KEEP — When to Use Which

Operation	MQTT retain	Typical use
`PUBLISH TOPIC`	Only when you add `RETAIN`	Current subscribers, connectors, dashboards, and external systems
`KEEP TOPIC`	Never	Broker-internal cache for `GET TOPIC` — not an MQTT publish

Use KEEP TOPIC for inter-action state (counters, mode flags, last-known values). Use PUBLISH TOPIC when connectors, UIs, or other subscribers must react to the message. Use PUBLISH TOPIC ... RETAIN when new subscribers must receive the last payload.

KEEP TOPIC "state/mode" WITH "running"
KEEP TOPIC "state/lastTemp" WITH {temp}
SET "mode" WITH (GET TOPIC "state/mode" AS STRING)

PUBLISH TOPIC "sensors/data" WITH {temp}
PUBLISH TOPIC "alerts/zone1" WITH "CRITICAL"

Do not use KEEP TOPIC to drive connector events — that path is unreliable. Use PUBLISH TOPIC when a connector or external consumer must react to the message.

PUBLISH TOPIC

Sends a message to an MQTT topic. Current subscribers receive it immediately; the message can trigger other actions or connectors listening on that topic. Optional RETAIN sets the MQTT retain flag on the published message (the same keyword as on DEFINE MODEL). Optional QOS0, QOS1, or QOS2 select MQTT QoS; the default remains QoS 0. Omitting RETAIN keeps the historical non-retained behaviour. RETAIN and a QoS keyword can appear in either order (RETAIN QOS1 or QOS1 RETAIN). KEEP TOPIC is different: it stores a value for GET TOPIC in the broker cache and does not set the MQTT retain flag. Use PUBLISH TOPIC ... RETAIN when new subscribers must receive the last payload. KEEP does not accept RETAIN. Several PUBLISH TOPIC or KEEP TOPIC lines to the same topic in one run each go out, in the order the action wrote them, with that line’s own retain, QoS, and keep setting. A later GET TOPIC in the same run still reads the value just written. In a cluster, only the last KEEP of each topic is replicated.

Syntax

PUBLISH TOPIC "<topic>" WITH <expression> [RETAIN] [QOS0|QOS1|QOS2]

Retain
Retain and QoS 1
QoS only

Store the last status so a new subscriber receives the last payload:

DEFINE ACTION PublishOnline
ON START DO
    PUBLISH TOPIC "status/last" WITH "online" RETAIN

Set the MQTT retain flag and publish at QoS 1:

DEFINE ACTION PublishOnlineQos1
ON START DO
    PUBLISH TOPIC "status/last" WITH "online" RETAIN QOS1

Choose QoS without retaining the message. Omit the QoS keyword and the publish stays QoS 0:

DEFINE ACTION PublishOnlineQos2
ON START DO
    PUBLISH TOPIC "status/last" WITH "online" QOS2

Value Types

You can publish various value types:

// String
PUBLISH TOPIC "system/status" WITH "online"

// Number
PUBLISH TOPIC "sensors/reading" WITH 23.5

// Variable
PUBLISH TOPIC "output/value" WITH {calculated_result}

// Payload passthrough (string path — may corrupt non-UTF-8 data)
PUBLISH TOPIC "forwarded/data" WITH PAYLOAD

// Timestamp
PUBLISH TOPIC "events/timestamp" WITH TIMESTAMP "UTC"

Binary passthrough — `PAYLOAD AS BYTES`

Use AS BYTES when the incoming MQTT payload must be forwarded, or stored with KEEP TOPIC, bit-for-bit — for example protobuf messages, Modbus frames, or image chunks. Untyped WITH PAYLOAD still uses the string publish path and may corrupt non-UTF-8 data. To decode or encode protobuf fields rather than forward the blob, use Protobuf Models (GET PROTO / PUBLISH MODEL). To hand those bytes to an external gRPC service, publish them on the source topic of a gRPC connector (TYPE GRPC). There is no CALL GRPC.

DEFINE ACTION ForwardBinary
ON TOPIC "binary/in" DO
    PUBLISH TOPIC "binary/out" WITH PAYLOAD AS BYTES

KEEP TOPIC and SET accept the same binary form, so you can store raw frames for later reads with GET TOPIC:

DEFINE ACTION KeepBinaryState
ON TOPIC "binary/in" DO
    KEEP TOPIC "state/raw" WITH PAYLOAD AS BYTES
    SET "frame" WITH PAYLOAD AS BYTES

Binary path limitations

Do not chain string transforms (TRIM, FILTER, REPLACE, SPLIT) after PAYLOAD AS BYTES on publish or keep — the parser rejects that combination.
Empty payloads are not published or stored on the binary path.

Cannot Use Inline JSON ObjectsPUBLISH TOPIC does not support inline JSON object literals. This will NOT work:

// ❌ WRONG - This does not work
PUBLISH TOPIC "output/data" WITH {"temperature": 23.5, "humidity": 65}

Why? PUBLISH TOPIC is designed for simple scalar values (strings, numbers, timestamps, variables). It doesn’t have the type system or schema validation needed for structured JSON data.Solution: For structured JSON output, use Models with PUBLISH MODEL:

// ✅ CORRECT - Define a model and publish it
DEFINE MODEL SensorData COLLAPSED
    ADD DOUBLE "temperature"
    ADD DOUBLE "humidity"

DEFINE ACTION PublishStructuredData
ON TOPIC "sensors/input" DO
    PUBLISH MODEL SensorData TO "output/data" WITH
        temperature = 23.5
        humidity = 65

Models provide type safety, schema validation, and proper JSON serialization. See Publishing Models for complete guidance.

Dynamic Topics

Build topic paths dynamically using string concatenation:

PUBLISH TOPIC "devices/" + {device_id} + "/status" WITH "active"
PUBLISH TOPIC "sensors/" + {sensor_type} + "/" + {location} + "/reading" WITH {value}

Multiple Publishes

Actions can publish to multiple topics:

DEFINE ACTION MultiPublish
ON TOPIC "input/data" DO
    PUBLISH TOPIC "output/raw" WITH PAYLOAD
    PUBLISH TOPIC "output/timestamp" WITH TIMESTAMP "UTC"
    PUBLISH TOPIC "output/source" WITH "sensor_001"

Triggering Other Actions: When you publish to a topic, any action with ON TOPIC matching that path will execute. This enables chaining actions together for complex workflows.

KEEP TOPIC

Stores a value for GET TOPIC in the broker cache. Unlike SET (which exists only during one execution), a KEEP value stays available to later runs on this broker. It is a broker-internal cache and never sets the MQTT retain flag. KEEP does not accept RETAIN (use PUBLISH TOPIC ... RETAIN).

KEEP TOPIC does not publish to MQTT. The value stays inside the broker: subscribers, MQTT clients, the HUB Data Viewer, dashboards and connectors never receive it. Only GET TOPIC can read it. Use PUBLISH TOPIC (with RETAIN if new subscribers need the last value) whenever the value must be visible.

When to Use KEEP

Use Case	Example
Cache last known value	`KEEP TOPIC "cache/last_temp" WITH {temperature}`
Store state between runs	`KEEP TOPIC "state/counter" WITH {count}`
Save intermediate calculations	`KEEP TOPIC "calc/running_avg" WITH {avg}`

Syntax

KEEP TOPIC "<topic>" WITH <expression>
KEEP TOPIC "<topic>" WITH PAYLOAD AS BYTES

Store string or variable state, including dynamic topic paths:

KEEP TOPIC "state/mode" WITH "running"
KEEP TOPIC "device/" + {id} + "/config" WITH {config}

Example: Persistent Counter

DEFINE ACTION PersistentCounter
ON EVERY 10 SECONDS DO
    SET "current" WITH (GET TOPIC "state/counter")
    IF {current} == EMPTY THEN
        KEEP TOPIC "state/counter" WITH 1
    ELSE
        KEEP TOPIC "state/counter" WITH ({current} + 1)
    
    PUBLISH TOPIC "stats/count" WITH (GET TOPIC "state/counter")

KEEP vs PUBLISH Comparison

Operation	MQTT retain	Delivered to subscribers	Readable with `GET TOPIC`	Typical use
`PUBLISH TOPIC`	Only with `RETAIN`	Yes	Yes (until overwritten)	Connectors, dashboards, external subscribers
`KEEP TOPIC`	No	No	Yes	Broker-internal cache for `GET TOPIC`

Control Flow

IF … THEN … ELSE

Execute different logic based on conditions.

IF <condition> THEN
    <statements>
ELSE
    <statements>

Comparison Operators

Operator	Description	Example
`EQUALS` or `==`	Equality	`IF {value} EQUALS 100`
`!=`	Not equal	`IF {status} != "error"`
`>`	Greater than	`IF {temp} > 80`
`<`	Less than	`IF {count} < 10`
`>=`	Greater or equal	`IF {level} >= 50`
`<=`	Less or equal	`IF {pressure} <= 100`

Basic Conditionals

IF {temperature} > 80 THEN
    PUBLISH TOPIC "alerts/high_temp" WITH {temperature}
ELSE
    PUBLISH TOPIC "status/normal" WITH "OK"

Nested Conditionals

For multiple conditions, nest IF statements:

IF {temperature} > 90 THEN
    PUBLISH TOPIC "alerts/critical" WITH "CRITICAL: " + {temperature}
ELSE
    IF {temperature} > 80 THEN
        PUBLISH TOPIC "alerts/warning" WITH "WARNING: " + {temperature}
    ELSE
        IF {temperature} > 70 THEN
            PUBLISH TOPIC "alerts/info" WITH "Elevated: " + {temperature}
        ELSE
            PUBLISH TOPIC "status/normal" WITH "OK"

Checking for Empty Topics

Use EMPTY to check if a topic has no value:

IF GET TOPIC "data/value" == EMPTY THEN
    PUBLISH TOPIC "data/value" WITH 0
ELSE
    PUBLISH TOPIC "data/value" WITH (GET TOPIC "data/value" + 1)

Combining with Expressions

// Check calculated values
SET "average" WITH (({temp1} + {temp2}) / 2)
IF {average} > {threshold} THEN
    PUBLISH TOPIC "alerts/average_high" WITH {average}

// String comparisons
IF {status} EQUALS "active" THEN
    PUBLISH TOPIC "devices/active_count" WITH (GET TOPIC "devices/active_count" + 1)

Timestamps

Add timestamps to your messages for tracking, logging, and debugging. LoT supports two timestamp formats.

Supported Formats

Format	Output	Use Case
`TIMESTAMP "UTC"`	`2025-10-25T14:30:15Z`	Human-readable logs, APIs, ISO 8601 compliance
`TIMESTAMP "UNIX"`	`1729864215`	Databases, calculations, cross-platform compatibility

Syntax

TIMESTAMP "UTC"   // ISO 8601 format
TIMESTAMP "UNIX"  // Unix epoch (seconds since 1970-01-01)

Using with PUBLISH and KEEP

// Add timestamp to published messages
PUBLISH TOPIC "events/created" WITH TIMESTAMP "UTC"
PUBLISH TOPIC "sensors/reading/time" WITH TIMESTAMP "UNIX"

// Store timestamp for later reference
KEEP TOPIC "cache/last_update" WITH TIMESTAMP "UNIX"

Using with SET

// Store timestamp in a variable
SET "event_time" WITH TIMESTAMP "UTC"
SET "process_start" WITH TIMESTAMP "UNIX"

// Use in string concatenation
SET "log_message" WITH ("Event occurred at " + TIMESTAMP "UTC")
PUBLISH TOPIC "logs/events" WITH {log_message}

Complete Example

DEFINE ACTION AuditLogger
ON TOPIC "actions/+/executed" DO
    SET "action_name" WITH TOPIC POSITION 2
    SET "utc_time" WITH TIMESTAMP "UTC"
    SET "unix_time" WITH TIMESTAMP "UNIX"
    
    // Human-readable log
    PUBLISH TOPIC "audit/" + {action_name} + "/log" 
        WITH ("Action " + {action_name} + " executed at " + {utc_time})
    
    // Machine-readable timestamp for database storage
    PUBLISH TOPIC "audit/" + {action_name} + "/timestamp" WITH {unix_time}

Callable Actions

LoT supports callable actions - actions that can be invoked by other actions with input parameters and return values. This enables modular, reusable logic blocks.

For reusable logic, on-demand scripts, and composition patterns, see Callable ACTIONs.

Defining a Callable Action

DEFINE ACTION ActionName
INPUT param1 AS <TYPE>
INPUT param2 AS <TYPE>
DO
    <action_logic>
RETURN
    OUTPUT variable1
    OUTPUT variable2

Supported Input Types

STRING
INT
DOUBLE
BOOL
JSON

Calling an Action

CALL ACTION ActionName
    WITH param1 = <value>, param2 = <value>
    RETURN var1, var2

Basic Example

Define
Call

Create a reusable calculation:

DEFINE ACTION CalculateAverage
INPUT value1 AS DOUBLE
INPUT value2 AS DOUBLE
DO
    SET "sum" WITH ({value1} + {value2})
    SET "avg" WITH ({sum} / 2)
RETURN
    OUTPUT avg

Use the callable action:

DEFINE ACTION ProcessSensorPair
ON TOPIC "sensors/pair/data" DO
    SET "temp1" WITH (GET JSON "sensor1" IN PAYLOAD AS DOUBLE)
    SET "temp2" WITH (GET JSON "sensor2" IN PAYLOAD AS DOUBLE)
    
    CALL ACTION CalculateAverage
        WITH value1 = {temp1}, value2 = {temp2}
        RETURN result
    
    PUBLISH TOPIC "sensors/average" WITH {result}

Multiple Outputs

DEFINE ACTION AnalyzeValue
INPUT value AS DOUBLE
INPUT threshold AS DOUBLE
DO
    SET "is_above" WITH ({value} > {threshold})
    SET "difference" WITH ({value} - {threshold})
    SET "percentage" WITH ({value} / {threshold} * 100)
RETURN
    OUTPUT is_above
    OUTPUT difference
    OUTPUT percentage

DEFINE ACTION CheckThreshold
ON TOPIC "sensors/+/value" DO
    SET "sensor_id" WITH TOPIC POSITION 2
    SET "current_value" WITH PAYLOAD AS DOUBLE
    SET "threshold" WITH (GET TOPIC "config/threshold" AS DOUBLE)
    
    CALL ACTION AnalyzeValue
        WITH value = {current_value}, threshold = {threshold}
        RETURN above, diff, pct
    
    PUBLISH TOPIC "analysis/" + {sensor_id} + "/above_threshold" WITH {above}
    PUBLISH TOPIC "analysis/" + {sensor_id} + "/difference" WITH {diff}
    PUBLISH TOPIC "analysis/" + {sensor_id} + "/percentage" WITH {pct}

Utility Action Library

Build reusable utility actions for common operations:

// Reusable temperature conversion
DEFINE ACTION ConvertCelsiusToFahrenheit
INPUT celsius AS DOUBLE
DO
    SET "fahrenheit" WITH ({celsius} * 9 / 5 + 32)
RETURN
    OUTPUT fahrenheit

// Reusable percentage calculation
DEFINE ACTION CalculatePercentage
INPUT value AS DOUBLE
INPUT total AS DOUBLE
DO
    SET "pct" WITH ({value} / {total} * 100)
RETURN
    OUTPUT pct

// Main processing action using utilities
DEFINE ACTION ProcessTemperatureReading
ON TOPIC "sensors/+/celsius" DO
    SET "sensor_id" WITH TOPIC POSITION 2
    SET "temp_c" WITH PAYLOAD AS DOUBLE
    SET "max_temp" WITH (GET TOPIC "config/max_temperature" AS DOUBLE)
    
    CALL ACTION ConvertCelsiusToFahrenheit
        WITH celsius = {temp_c}
        RETURN temp_f
    
    CALL ACTION CalculatePercentage
        WITH value = {temp_c}, total = {max_temp}
        RETURN capacity
    
    PUBLISH TOPIC "sensors/" + {sensor_id} + "/fahrenheit" WITH {temp_f}
    PUBLISH TOPIC "sensors/" + {sensor_id} + "/capacity_percent" WITH {capacity}

TRIGGER Statement

The TRIGGER statement allows actions to programmatically trigger connector events or mappings. This is useful for initiating data pipeline operations or triggering external integrations.

Syntax

TRIGGER "<connector_name>" TO <data> FOR <mapping_name>

Components

Component	Description
`connector_name`	The name of the connector to trigger
`data`	The data to send (PAYLOAD, variable, or expression)
`mapping_name`	The specific mapping or event within the connector

Basic Examples

// Trigger a connector with payload data
TRIGGER "EmailRoute" TO PAYLOAD FOR "SendAlert"

// Trigger with a variable
SET "message" WITH ("Alert: " + {sensor_id} + " exceeded threshold")
TRIGGER "NotificationRoute" TO {message} FOR "EmailMapping"

// Trigger with a literal value
TRIGGER "DatabaseRoute" TO "backup_requested" FOR "BackupEvent"

Complete Example

DEFINE ACTION AlertHandler
ON TOPIC "alerts/+/critical" DO
    SET "alert_type" WITH TOPIC POSITION 2
    SET "alert_data" WITH PAYLOAD
    SET "timestamp" WITH TIMESTAMP "UTC"
    
    // Build alert message
    SET "message" WITH ("Critical alert [" + {alert_type} + "] at " + {timestamp})
    
    // Trigger email notification connector
    TRIGGER "EmailNotification" TO {message} FOR "CriticalAlertEmail"
    
    // Also trigger database storage
    TRIGGER "DatabaseStorage" TO {alert_data} FOR "StoreAlert"
    
    PUBLISH TOPIC "alerts/processed" WITH {alert_type}

Use Cases

Email Notifications
Database Operations
External APIs

DEFINE ACTION SendDailyReport
ON EVERY 24 HOURS DO
    SET "report_data" WITH (GET TOPIC "reports/daily/summary")
    TRIGGER "EmailRoute" TO {report_data} FOR "DailyReportEmail"

DEFINE ACTION ArchiveOldData
ON EVERY 1 HOUR DO
    SET "cutoff_time" WITH TIMESTAMP "UNIX"
    TRIGGER "DatabaseRoute" TO {cutoff_time} FOR "ArchiveEvent"

DEFINE ACTION NotifyExternalSystem
ON TOPIC "events/important" DO
    TRIGGER "RestAPIRoute" TO PAYLOAD FOR "WebhookEvent"

CALL MCP

CALL MCP allows actions to invoke tools provided by MCP (Model Context Protocol) connectors. This enables integration with AI assistants and external tools.

CALL MCP talks to an MCP connector (TYPE MCP) — the broker as a client of an external server. To let an AI client operate this broker, enable the MCP Server instead. TYPE MCP_SERVER was removed.A gRPC connector (TYPE GRPC) is also a client connector, and it is not invoked with CALL. It forwards MQTT payloads from SOURCE_TOPIC.

Prerequisites

An MCP connector must be defined and active
The connector must expose the tool you want to call

Syntax

CALL MCP "RouteName.tool_name"
    WITH (arg1 = <value1>, arg2 = <value2>, ...)
    RETURN AS {variable_name}

Example MCP Connector

DEFINE CONNECTOR FileSystem WITH TYPE MCP
    ADD MCP_CONFIG
        WITH SERVER_COMMAND "npx"
        WITH SERVER_ARGUMENTS "-y, @modelcontextprotocol/server-filesystem, /data"
        WITH CLIENT_NAME "CorefluxBroker"

Basic Examples

// Call an MCP tool with arguments
CALL MCP "FileSystem.read_file"
    WITH (path = "/data/config.json")
    RETURN AS {file_content}

// Call without arguments
CALL MCP "MyMCPRoute.list_resources"
    RETURN AS {resources}

// Call with multiple arguments
CALL MCP "MyMCPRoute.echo"
    WITH (message = "Hello", count = 3)
    RETURN AS {response}

Return Value

RETURN AS receives a JSON string with success, tool_name, route_name, timestamp, and execution_time_ms. A successful call adds result, the server’s tool result. A failed call adds error instead. See Return Value.

Complete Example

DEFINE ACTION ReadDeviceConfig
ON TOPIC "devices/+/request_config" DO
    SET "device_id" WITH TOPIC POSITION 2
    SET "config_path" WITH "/configs/" + {device_id} + ".json"
    
    CALL MCP "FileSystem.read_file"
        WITH (path = {config_path})
        RETURN AS {config_data}
    
    SET "read_ok" WITH (GET JSON "success" IN {config_data} AS BOOL)
    IF {read_ok} EQUALS FALSE THEN
        PUBLISH TOPIC "devices/" + {device_id} + "/config/error" WITH (GET JSON "error" IN {config_data} AS STRING)
    ELSE
        PUBLISH TOPIC "devices/" + {device_id} + "/config" WITH {config_data}

CALL LLM

CALL LLM sends a prompt to a TYPE LLM connector and returns JSON. The connector does not call tools. For an assistant that can read topics or publish, use CALL AGENT.

DEFINE CONNECTOR PlantModel WITH TYPE LLM
    ADD LLM_CONFIG
        WITH PROVIDER "openai"
        WITH MODEL "gpt-6-luna"
        WITH API_KEY GET SECRET "OPENAI_API_KEY"
        WITH API_SURFACE "auto"

DEFINE ACTION ExplainAlarm
ON TOPIC "alarms/raw/+" DO
    CALL LLM "PlantModel.generate"
        WITH (prompt = PAYLOAD)
        RETURN AS {result}
    PUBLISH TOPIC "alarms/explained" WITH (GET JSON "response" IN {result} AS STRING)

generate is one turn. chat with use_history = "true" keeps an in-memory thread per session_id. clear_history drops that thread. INVOKE LLM is the same call. Check success in the JSON before you use response.

CALL AGENT

CALL AGENT runs a DEFINE AGENT and waits for the answer. The call runs as the agent’s DEFAULT_USER. execute returns the answer text, or JSON { "success": false, "status", "code", "message" } when the task is refused or fails.

DEFINE CONNECTOR LocalLlama WITH TYPE LLM
    ADD LLM_CONFIG
        WITH MODEL "llama3.2"

DEFINE AGENT Classifier
    USES LLM CONNECTOR LocalLlama
    WITH DEFAULT_USER "agent_svc"
    WITH SYSTEM_PROMPT "Answer with one word: safety, quality or other."
    WITH MEMORY "none"

DEFINE ACTION ClassifyNote
ON TOPIC "operators/notes" DO
    CALL AGENT "Classifier.execute"
        WITH (task = PAYLOAD)
        RETURN AS {label}
    PUBLISH TOPIC "operators/notes/classified" WITH {label}

clear_history takes session_id and forgets that session for DEFAULT_USER. A name that is not a running agent returns unknown_agent.

Complete Example

This example combines the key operations covered on this page—SET, GET TOPIC, GET JSON, arithmetic, IF/THEN, PUBLISH, and KEEP—in a realistic sensor processing action.

DEFINE ACTION ComprehensiveProcessor
ON TOPIC "sensors/+/data" DO
    // Extract sensor ID from topic path (POSITION 2 = the wildcard)
    SET "sensor_id" WITH TOPIC POSITION 2
    
    // Parse JSON payload
    SET "temperature" WITH (GET JSON "temp" IN PAYLOAD AS DOUBLE)
    
    // Convert to Fahrenheit
    SET "fahrenheit" WITH ({temperature} * 9 / 5 + 32)
    
    // Conditional alerting
    IF {temperature} > 80 THEN
        PUBLISH TOPIC "alerts/" + {sensor_id} WITH "High temp: " + {temperature}
    
    // Publish processed data
    PUBLISH TOPIC "processed/" + {sensor_id} + "/temperature" WITH {temperature}
    PUBLISH TOPIC "processed/" + {sensor_id} + "/fahrenheit" WITH {fahrenheit}
    PUBLISH TOPIC "processed/" + {sensor_id} + "/timestamp" WITH TIMESTAMP "UTC"
    
    // Cache for later reference
    KEEP TOPIC "cache/" + {sensor_id} + "/last_reading" WITH PAYLOAD

What This Action Does

Triggers on any message to sensors/+/data topics
Extracts the sensor ID from the topic path
Parses the JSON payload to get temperature value
Converts Celsius to Fahrenheit using arithmetic
Alerts if temperature exceeds threshold
Publishes formatted data to processed topics
Caches the raw payload for future reference

Quick Reference

Operation	Category	Description
`+, -, *, /, %`	Arithmetic	Mathematical operations
`+ (strings)`	String	Concatenation
`RANDOM BETWEEN`	Generation	Random numbers
`RANDOM UUID/GUID/ULID`	Generation	Unique identifiers
`FILTER ... USING REGEX`	Transformation	Extract with regex
`REPLACE ... WITH ... IN`	Transformation	String replacement
`SPLIT ... CSV`	Transformation	Parse CSV data
`GET ENV`	Configuration	Read environment variable
`GET SECRET`	Configuration	Read encrypted secret
`KEEP ENV`	Configuration	Persist environment variable
`KEEP SECRET`	Configuration	Persist encrypted secret
`DELETE ENV`	Configuration	Remove environment variable
`DELETE SECRET`	Configuration	Remove secret
`CALL PYTHON`	Integration	Execute Python functions
`CALL MCP`	Integration	Invoke MCP tools
`CALL LLM`	Integration	Prompt a TYPE LLM connector
`CALL AGENT`	Integration	Run a DEFINE AGENT and wait for the answer
`CALL ACTION`	Control	Call other actions
`TRIGGER`	Control	Trigger connector events
`PUBLISH TOPIC`	Topic	Publish to connectors, dashboards, and subscribers. Optional `RETAIN` and `QOS0` / `QOS1` / `QOS2`. Omitting `RETAIN` keeps the historical non-retained behaviour.
`PUBLISH TOPIC ... WITH PAYLOAD AS BYTES`	Topic	Forward binary payloads bit-for-bit
`KEEP TOPIC`	Topic	Broker-internal cache for `GET TOPIC`. Never sets the MQTT retain flag. Does not accept `RETAIN`.
`KEEP TOPIC ... WITH PAYLOAD AS BYTES`	Topic	Store binary payloads in the broker cache without string conversion

Next Steps

Python Integration

Extend actions with Python functions for complex logic.

LoT Notebooks

Run and deploy these operations from VS Code with LoT Notebooks.

Last modified on October 25, 2026

⌘I

​Overview

​Variable Operations

​SET

​GET TOPIC

​Default Values for Missing Topics

​Type Casting

​Wildcard Inheritance

​Example Usage

​GET JSON

​Basic Syntax

​Supported Types

​Nested Path Extraction

​Error Handling

​Complete Example

​GET PROTO

​Basic Syntax

​GET MODEL

​Basic Syntax

​Example Usage

​Environment Variables & Secrets

​GET ENV

​Examples

​KEEP ENV

​DELETE ENV

​GET SECRET

​Examples

​KEEP SECRET

​DELETE SECRET

​Complete Example

​Arithmetic Operators

​Supported Operators

​Syntax

​Basic Examples

​Using with Variables

​Using with GET TOPIC

​Using with PAYLOAD

​Operator Precedence

​Complete Example

​String Concatenation

​Syntax

​Basic Examples

​Dynamic Topic Construction

​Combining Strings with Other Types

​Complete Example

​RANDOM Function

​Syntax

​Random Integers

​Random Doubles

​Unique Identifiers

​Use Cases

​String Transformations

​FILTER

​Syntax

​Supported Sources

​Examples

​Common Regex Patterns

​REPLACE

​Syntax

​Examples

​Multiple Replacements

​SPLIT CSV

​Syntax

​Parameters

​Example

​Topic Operations

​PUBLISH vs KEEP — When to Use Which

​PUBLISH TOPIC

​Syntax

​Value Types

​Binary passthrough — PAYLOAD AS BYTES

​Dynamic Topics

​Multiple Publishes

​KEEP TOPIC

​When to Use KEEP

​Syntax

​Example: Persistent Counter

​KEEP vs PUBLISH Comparison

​Control Flow

​IF … THEN … ELSE

​Comparison Operators

Overview

Variable Operations

SET

GET TOPIC

Default Values for Missing Topics

Type Casting

Wildcard Inheritance

Example Usage

GET JSON

Basic Syntax

Supported Types

Nested Path Extraction

Error Handling

Complete Example

GET PROTO

Basic Syntax

GET MODEL

Basic Syntax

Example Usage

Environment Variables & Secrets

GET ENV

Examples

KEEP ENV

DELETE ENV

GET SECRET

Examples

KEEP SECRET

DELETE SECRET

Complete Example

Arithmetic Operators

Supported Operators

Syntax

Basic Examples

Using with Variables

Using with GET TOPIC

Using with PAYLOAD

Operator Precedence

Complete Example

String Concatenation

Syntax

Basic Examples

Dynamic Topic Construction

Combining Strings with Other Types

Complete Example

RANDOM Function

Syntax

Random Integers

Random Doubles

Unique Identifiers

Use Cases

String Transformations

FILTER

Syntax

Supported Sources

Examples

Common Regex Patterns

REPLACE

Syntax

Examples

Multiple Replacements

SPLIT CSV

Syntax

Parameters

Example

Topic Operations

PUBLISH vs KEEP — When to Use Which

PUBLISH TOPIC

Syntax

Value Types

Binary passthrough — `PAYLOAD AS BYTES`

Dynamic Topics

Multiple Publishes

KEEP TOPIC

When to Use KEEP

Syntax

Example: Persistent Counter

KEEP vs PUBLISH Comparison

Control Flow

IF … THEN … ELSE

Comparison Operators