Version: 1.0.8

Pipeline Node Reference

All 26 node types, generated from the pipeline node templates. For usage guidance and worked examples see Pipeline nodes.

At a glance

Type	Label	Category	In	Out	Description
`ai`	AI	ai	✓	✓	Classify, extract or summarise with AI — and optionally look things up first
`rest-api`	REST API	api	✓	✓	Define HTTP endpoint (ingress) or make outbound HTTP calls (egress). Outbound sends the incoming message only in Dynamic configuration mode
`rest-response`	HTTP Response	api	✓	—	Send the final HTTP response for a REST Ingress pipeline
`inject`	Inject	common	—	✓	Trigger flow manually or on a schedule
`link`	Link	common	✓	—	Send or receive messages between pipelines
`rbe`	RBE (Filter)	common	✓	✓	Report-by-exception: only pass a message when the value changes
`switch`	Switch	common	✓	✓	Route a message to different branches depending on its value
`timer`	Timer	common	✓	✓	Delay, debounce or watch for silence — per topic
`database-sql`	SQL Query	datasource	✓	✓	Execute SQL queries on connected database
`mongodb`	MongoDB	datasource	✓	✓	Find, Insert, Update, or Delete documents in MongoDB
`questdb`	QuestDB	datasource	✓	✓	High-speed Ingress or SQL query on QuestDB
`message-in`	Message In	messagebus	—	✓	Start this pipeline when a message arrives on a subject
`message-out`	Message Out	messagebus	✓	✓	Send a message on a subject, or request and wait for a reply
`message-reply`	Message Reply	messagebus	✓	✓	Answer the request that started this run — with a result, or with the error a Catch node caught
`notify-email`	Send Email	notification	✓	—	Send an email via a saved SMTP connection
`telegram`	Telegram	notification	✓	—	Send messages, or receive messages/commands (e.g. /data), via a saved Telegram bot
`user-roster`	On-Call Roster	notification	✓	✓	Resolve the on-shift roster members into recipients for a Send node
`whatsapp`	WhatsApp	notification	✓	—	Send WhatsApp messages (Twilio), or receive inbound messages via a webhook
`catch`	Catch	output	—	✓	Report failures from other nodes in this pipeline
`debug`	Debug	output	✓	—	Send a node's output to the Debug panel
`alarm`	Alarm	realtime	✓	✓	Subscribe to alarm events (raised, cleared, acknowledged) for alarm-enabled Tags
`industrial-io`	Industrial I/O	realtime	✓	✓	Read, write, or subscribe to Tags (OPC-UA, MQTT, etc.)
`batch-transform`	Batch Transform	transform	✓	✓	Apply transformation to each item in parallel
`join`	Join	transform	✓	✓	Collect array items or independent messages into a single array
`python`	Python Script	transform	✓	✓	Execute custom Python code for transformations
`split`	Array Iterator	transform	✓	✓	Iterate through an array and process items one by one

AI

Classify, extract or summarise with AI — and optionally look things up first

Type: ai · Category: ai · Input: yes · Output: yes

Default configuration

Key	Default
`promptTemplate`	`""`
`outputMode`	fields
`outputFields`	[]
`choices`	[]
`scope`	[]
`skills`	[]
`outputPath`	payload.classification
`temperature`	0
`maxTokens`	1024
`maxToolIterations`	8
`timeoutMs`	60000
`cache`	`{"enabled":true,"ttlSeconds":3600}`
`onError`	fail

Alarm

Subscribe to alarm events (raised, cleared, acknowledged) for alarm-enabled Tags

Type: alarm · Category: realtime · Input: yes · Output: yes

Default configuration

Key	Default
`operation`	subscribe
`payloadMode`	event
`events`	["raised"]
`priorities`	[]
`tagIds`	[]
`includeTagMeta`	false

Array Iterator

Iterate through an array and process items one by one

Type: split · Category: transform · Input: yes · Output: yes

Default configuration

Key	Default
`payloadType`	msg
`payloadValue`	`""`

Batch Transform

Apply transformation to each item in parallel

Type: batch-transform · Category: transform · Input: yes · Output: yes

Default configuration

Key	Default
`code`	`""`
`transformType`	map
`payloadType`	msg
`payloadValue`	`""`

Default code

# Batch Transform Node
# Process individual items in a collection

def transform(item, message):
    """
    Called for each item in the incoming array.
    :param item: Current item data
    :param message: Parent message context
    :return: Transformed item
    """
    return item

Catch

Report failures from other nodes in this pipeline

Type: catch · Category: output · Input: no · Output: yes

Default configuration

Key	Default
`scope`	all
`sourceNodeIds`	[]

Debug

Send a node's output to the Debug panel

Type: debug · Category: output · Input: yes · Output: no

Default configuration

Key	Default
`displayMode`	full

HTTP Response

Send the final HTTP response for a REST Ingress pipeline

Type: rest-response · Category: api · Input: yes · Output: no

Default configuration

Key	Default
`http_status`	200
`content_type`	application/json
`http_headers`	`{}`
`reply_to`	`""`

Industrial I/O

Read, write, or subscribe to Tags (OPC-UA, MQTT, etc.)

Type: industrial-io · Category: realtime · Input: yes · Output: yes

Default configuration

Key	Default
`operation`	read
`tagSource`	static
`tagId`	`""`
`path`	`""`
`valuePath`	`""`
`value`	`""`
`dataTypeHint`	`""`

Inject

Trigger flow manually or on a schedule

Type: inject · Category: common · Input: no · Output: yes

Default configuration

Key	Default
`repeat`	none
`intervalValue`	60
`intervalUnit`	seconds
`cronExpression`	`""`
`initialDelay`	0
`payloadType`	timestamp
`payloadValue`	`""`

Join

Collect array items or independent messages into a single array

Type: join · Category: transform · Input: yes · Output: yes

Default configuration

Key	Default
`mode`	auto

Link

Send or receive messages between pipelines

Type: link · Category: common · Input: yes · Output: no

Default configuration

Key	Default
`mode`	egress
`sourceNodeIds`	[]

Message In

Start this pipeline when a message arrives on a subject

Type: message-in · Category: messagebus · Input: no · Output: yes

Default configuration

Key	Default
`subject`	`""`
`includeAll`	true
`queueGroup`	`""`

Message Out

Send a message on a subject, or request and wait for a reply

Type: message-out · Category: messagebus · Input: yes · Output: yes

Default configuration

Key	Default
`subject`	`""`
`scope`	all
`mode`	send
`timeoutMs`	5000

Message Reply

Answer the request that started this run — with a result, or with the error a Catch node caught

Type: message-reply · Category: messagebus · Input: yes · Output: yes

Default configuration

Key	Default
`status`	ok
`errorMessage`	`""`

MongoDB

Find, Insert, Update, or Delete documents in MongoDB

Type: mongodb · Category: datasource · Input: yes · Output: yes

Default configuration

Key	Default
`connectionId`	`""`
`database`	`""`
`collection`	`""`
`operation`	find
`filter`	`""`
`update`	`""`
`limit`	0
`timeout`	0

On-Call Roster

Resolve the on-shift roster members into recipients for a Send node

Type: user-roster · Category: notification · Input: yes · Output: yes

Default configuration

Key	Default
`dbId`	`""`
`rosterId`	`""`

Python Script

Execute custom Python code for transformations

Type: python · Category: transform · Input: yes · Output: yes

Default configuration

Key	Default
`code`	`""`
`timeout`	60

Default code

# Python Script Node
# Supported Modules: system, math, json, datetime
# Globals: payload (data), message (metadata)

def transform(payload, message):
    """
    Transform incoming data.
    :param payload: The primary data object (e.g., from the previous node)
    :param message: The full StandardMessage including metadata and context
    :return: The transformed result to pass to the next node
    """
    
    # Example 1: Accessing real-time tags
    # value = system.tag.read("Compressor/Pressure")
    # log(f"Current Pressure: {value}")
    
    # Example 2: Running a project-scoped Named Query
    # results = system.db.runNamedQuery("GetActiveAlerts", {"area": "Floor-1"})
    
    # Example 3: Messaging views and pipelines ("On message" handlers, Message In nodes)
    # system.message.send("line1.pressure.high", {"value": value}, {"scope": "project"})
    
    return payload

QuestDB

High-speed Ingress or SQL query on QuestDB

Type: questdb · Category: datasource · Input: yes · Output: yes

Default configuration

Key	Default
`connectionId`	`""`
`mode`	ingress
`table`	`""`
`symbols`	`""`
`query`	`""`
`timeout`	0
`maxRetries`	0
`retryDelay`	0

RBE (Filter)

Report-by-exception: only pass a message when the value changes

Type: rbe · Category: common · Input: yes · Output: yes

Default configuration

Key	Default
`mode`	rbe
`property`	payload
`gap`	`""`
`inout`	out
`start`	`""`
`septopics`	true

REST API

Define HTTP endpoint (ingress) or make outbound HTTP calls (egress). Outbound sends the incoming message only in Dynamic configuration mode

Type: rest-api · Category: api · Input: yes · Output: yes

Default configuration

Key	Default
`mode`	ingress
`method`	GET
`urlPattern`	/api/v1/example
`timeout`	30
`authType`	none
`allowedIPs`	`""`
`rateLimit`	0
`burstLimit`	0
`maxPayloadSize`	1048576
`corsEnabled`	false
`allowedOrigins`	*
`ingressWhitelistHeaders`	`""`
`egressStaticHeaders`	`{}`

Send Email

Send an email via a saved SMTP connection

Type: notify-email · Category: notification · Input: yes · Output: no

Default configuration

Key	Default
`connectionId`	`""`
`recipients`	[]
`subject`	`""`
`body`	`""`

SQL Query

Execute SQL queries on connected database

Type: database-sql · Category: datasource · Input: yes · Output: yes

Default configuration

Key	Default
`connectionId`	`""`
`query`	`""`

Switch

Route a message to different branches depending on its value

Type: switch · Category: common · Input: yes · Output: yes

Default configuration

Key	Default
`property`	payload
`checkall`	true
`rules`	`[{"operator":"eq","value":"","valueType":"str"}]`

Send messages, or receive messages/commands (e.g. /data), via a saved Telegram bot

Type: telegram · Category: notification · Input: yes · Output: no

Default configuration

Key	Default
`mode`	send
`connectionId`	`""`
`to`	`""`
`body`	`""`
`commands`	`""`

Timer

Delay, debounce or watch for silence — per topic

Type: timer · Category: common · Input: yes · Output: yes

Default configuration

Key	Default
`mode`	oneshot
`duration`	5
`durationUnit`	seconds
`initialPayload`	1
`finalPayload`	0
`repeatWhileSilent`	false
`maxQueue`	1000
`direction`	down
`warningAt`	0
`maxIterations`	0
`loopTimeout`	0
`septopics`	true
`topicProperty`	topic
`maxTopics`	1000
`controlProperty`	metadata.timerCmd
`startAction`	START
`stopAction`	STOP
`resetAction`	RESET
`pauseAction`	PAUSE
`continueAction`	CONTINUE
`targetProperty`	`""`
`durationProperty`	`""`
`durationOverrideOn`	start
`maxDuration`	86400

Send WhatsApp messages (Twilio), or receive inbound messages via a webhook

Type: whatsapp · Category: notification · Input: yes · Output: no

Default configuration

Key	Default
`mode`	send
`connectionId`	`""`
`recipients`	[]
`body`	`""`
`templateId`	`""`
`waSendMode`	freeform

At a glance​

AI​

Alarm​

Array Iterator​

Batch Transform​

Catch​

Debug​

HTTP Response​

Industrial I/O​

Inject​

Join​

Link​

Message In​

Message Out​

Message Reply​

MongoDB​

On-Call Roster​

Python Script​

QuestDB​

RBE (Filter)​

REST API​

Send Email​

SQL Query​

Switch​

Telegram​

Timer​

WhatsApp​

At a glance

AI

Alarm

Array Iterator

Batch Transform

Catch

Debug

HTTP Response

Industrial I/O

Inject

Join

Link

Message In

Message Out

Message Reply

MongoDB

On-Call Roster

Python Script

QuestDB

RBE (Filter)

REST API

Send Email

SQL Query

Switch

Telegram

Timer

WhatsApp