# Node types

Built-in nodes cover common integrations, timing, and logic. Configure their
settings in the node panel; use a **Function** node for your own Python.

`GET /flows/node-types` returns this list with each type's full parameter
schema, including any [connectors](#connectors) installed on your instance.

MQTT and InfluxDB need their [extras](configuration.md#image-options): `mqtt`
and `influx` in the image's `EXTRAS`, or `pip install 'fluksio[mqtt,influx]'`
on a Python installation. Run the broker and database separately.

## Function

**`python`** — run your own code when its triggering inputs arrive.

The one type with a source file. Its arguments are its input ports and its own
settings; its return value is a dict keyed by output ports. See
[Writing node code](../code/nodes.md).

Settings on a Function node are free-form: you add them by name, and they
arrive as keyword arguments.

## Integrations

### MQTT

**`mqtt`** — listen on topics, or send to them.

The ports pick the direction: outputs only make a **listener**, inputs a
**sender**. A listener subscribes to its topics, a sender publishes its inputs
to them.

| Setting | Default | Notes |
|---|---|---|
| `topic` | `*` | one topic for every port, or `{"message": "some/topic"}` per port |
| `broker_host` | `localhost` | |
| `broker_port` | `1883` | |
| `username` / `password` | — | `password` takes a secret reference |
| `client_id` | — | |
| `qos` | `0` | 0, 1 or 2 |
| `retain` | `false` | on published messages |
| `keepalive` | `60` | seconds |
| `timeout` | `10.0` | seconds before a broker operation is given up on |
| `send_queue_size` | `256` | payloads that may wait for the broker; past this the oldest is dropped and the node reports degraded |
| `json_key` | — | key to lift out of an object payload; one for every port, or `{"port": "key"}` |

A topic may be a filter: `+` matches one level, `#` the rest. Everything a
filter matches lands on the same port, so use one port per topic where the
difference matters.

Set an explicit topic: the default `*` is a literal topic, not MQTT's
match-all filter.

For `{"value": 47}`, set `json_key` to `value`. Without it, an object is
unwrapped only if it contains the port's name as a key.

### HTTP

**`http`** — listen on a webhook, or send to a URL and pass the reply on.

The ports pick the direction: outputs only make a **listener**, inputs a
**sender**. A listener is a **webhook**: the engine mounts a route at
`/hooks/{flow}/{url}/{secret}` while the flow runs. A sender requests `url`
and publishes a JSON object response on its outputs, one key per output port.
Keys no port declares are dropped, and a response that is not an object
publishes nothing.

| Setting | Default | Notes |
|---|---|---|
| `url` | — | the route path for a webhook, the full URL for a sender |
| `method` | `POST` | `GET` or `POST` |
| `timeout` | `30` | seconds, sender only |
| `headers` | `{}` | request headers, sender only; a value may be a secret reference |
| `query` | `{}` | fixed query parameters, sender only; a value may be a secret reference |
| `send_inputs` | `true` | off when the inputs only trigger the request |
| `secret` | — | shared secret appended to the webhook URL; takes a secret reference |

!!! warning "A webhook with no secret is open to anyone who can reach the host."

    The canvas flags this as an advisory issue rather than stopping the flow,
    because a webhook on a private network is a legitimate thing to want.
    Through the [portal](../interface/portal.md#hooks-and-agents-through-the-portal)
    a webhook or served socket answers only with a secret.

### WebSocket

**`websocket`** — listen on a socket, or send to it, as a client or a server.

The URL picks the role: `ws://…` or `wss://…` connects to a server and
reconnects when it drops; a path such as `/ticker` is served at
`/hooks/{flow}/{url}/{secret}` while the flow runs. The ports pick the
direction: outputs only make a **listener**, inputs a **sender**. A served
sender sends to every connected client.

| Setting | Default | Notes |
|---|---|---|
| `url` | — | `ws://` or `wss://` URL, or the path to serve |
| `headers` | `{}` | handshake headers, client only; a value may be a secret reference |
| `on_connect` | `[]` | frames sent after every (re)connect, client only — a subscribe or auth message; an item may be a secret reference |
| `secret` | — | shared secret appended to a served socket's URL; takes a secret reference |
| `timeout` | `10.0` | seconds for the handshake, and for a send to one client of a served socket |
| `send_queue_size` | `256` | messages that may wait to be sent; past this the oldest is dropped and the node reports degraded |
| `json_key` | — | as on `mqtt` |

Frames are text. An incoming frame is parsed as JSON: an object feeds the
ports it names, and a node with a single output takes any other frame whole.
A single input is sent as it stands — a string raw, anything else as JSON —
and several go as one object keyed by port.

### InfluxDB

**`influxdb`** — write measurements to a bucket, or read them back.

| Setting | Notes |
|---|---|
| `url`, `token`, `org`, `bucket` | connection; `token` takes a secret reference |
| `write_precision` | `ns`, `us`, `ms` (default) or `s` |
| `query_range` | default window for queries, e.g. `-1h` |
| `timeout` | operation timeout in seconds; default `10.0` |
| `writes` | per-input: `measurement`, `field`, `tags` |
| `queries` | per-output: `measurement`, `field`, `tags`, `range`, `aggregation` |

```json
{
  "writes": {
    "living_temperature": {
      "measurement": "environment",
      "field": "temp_c",
      "tags": {"room": "living"}
    }
  }
}
```

**Query passthrough.** An incoming object with a `flux` key supplies the query.
The first output receives an object with `rows`, each containing `ts`, `value`,
`field`, `measurement`, and `tags`. Use Function nodes to build queries and
shape results for charts.

### Notification

**`ntfy`** — push an incoming value to a phone through [ntfy](https://ntfy.sh).

| Setting | Default | Notes |
|---|---|---|
| `server` | `https://ntfy.sh` | |
| `topic` | — | required |
| `title` | — | |
| `priority` | `default` | `min`, `low`, `default`, `high`, `urgent` |
| `tags` | — | comma-separated ntfy tags |
| `token` | — | for a protected topic; takes a secret reference |
| `timeout` | `10.0` | request timeout in seconds |

For engine failures rather than flow messages, configure
[Alerts](../interface/operations.md#alerts).

## Timing

### Inject

**`inject`** — emit a value on request, on a timer, or when the flow starts.

| Setting | Default | Notes |
|---|---|---|
| `payload` | current time | what to emit |
| `payloads` | `{}` | per-output-port payloads, keyed by port |
| `interval` | `0` | emit every n seconds; 0 means never on its own |
| `cron` | — | a five-field cron expression |
| `at_start` | `false` | emit once when the flow starts |
| `start_delay` | `1.0` | how long to wait before that first emission |

Use `cron` for scheduled emissions, or trigger the node by hand for testing.

### Delay & schedule

**`delay`** — hold messages back, limit their rate, or emit on a schedule.

| Setting | Default | Notes |
|---|---|---|
| `delay` | `0` | seconds to hold each message, fractional |
| `interval` | `0` | minimum seconds between forwards, per output |
| `mapping` | `{}` | input port → output port; paired in order when empty |
| `cron` | — | five-field expression |

Order of operations: delay → rate check → forward. A message arriving inside
the `interval` window is held, and when the window ends the latest one held is
sent. With a `cron` and no inputs it emits the current time on each tick; with
inputs it emits the last value it received.

Timers wait until their deadline; busy workers can delay execution.

!!! note "Not in a batch flow"

    Batch flows reject rate-limited non-streaming ports. Runs have no live
    timer to release a held value.

### Debounce

**`debounce`** — send one value now and another once things go quiet. A
debounce and hold rather than a scheduler: everything it sends starts from a
value arriving. For a cron tick, see [`inject`](#inject). Node-RED calls this
node *trigger*.

| Setting | Default | Notes |
|---|---|---|
| `first` | `true` | sent as soon as a value arrives |
| `then` | `false` | sent when the wait expires; empty sends nothing |
| `wait` | `60` | seconds of quiet before the second value |
| `extend` | `true` | a value arriving during the wait starts it over |
| `passthrough` | `false` | send the incoming value instead of `first` |
| `wait_port` | — | an input carrying the wait, when it differs per message |

For example: turn a light on when the door opens, then off after two quiet
minutes.

Use `wait_port` when the delay comes from an input. A wait of zero or less
cancels the pending second value and schedules nothing new.

## Logic

### Switch

**`switch`** — send a value down one branch or another, by rule.

Each rule names an output port; a matching value leaves through that port.
Comparisons: `eq`, `ne`, `gt`, `gte`, `lt`, `lte`, `contains`, `between`.

| Setting | Default | Notes |
|---|---|---|
| `rules` | `[]` | checked in order; each names the output it routes to |
| `stop_at_first` | `true` | leave through the first matching rule only |
| `otherwise` | — | output for a value that matched nothing |

Comparing a string to a number is a mistake in the rule, not a reason to take
the flow down: the rule simply does not match.

### Change

**`change`** — scale, offset, round or map a value on its way past.

| Setting | Default | Notes |
|---|---|---|
| `scale` | `1.0` | multiply numbers by this |
| `offset` | `0.0` | then add this |
| `round_to` | — | decimal places |
| `mapping` | `{}` | replace a value with another, looked up as text |
| `default` | — | value when the lookup misses; empty passes it through |

### Filter unchanged

**`rbe`** — pass a value on only when it has actually changed.

| Setting | Default | Notes |
|---|---|---|
| `deadband` | `0.0` | ignore numeric changes smaller than this |
| `deadband_percent` | `false` | read the deadband as a percentage |

The cheapest fix for a sensor that reports every second and changes every hour.

### Join

**`join`** — gather several inputs into one object or list.

| Setting | Default | Notes |
|---|---|---|
| `mode` | `object` | `object` or `array` |

## System

### Command

**`exec`** — run a command where the engine runs and read its output.
Adding one or changing its settings requires a superuser.

| Setting | Default | Notes |
|---|---|---|
| `command` | — | the command to run |
| `append_payload` | `false` | add the incoming value as one argument |
| `timeout` | `30` | seconds |
| `fail_on_error` | `false` | treat a non-zero exit as a node failure rather than output |

Outputs the command's stdout, stderr and exit code.

In Docker, the command runs inside the backend container. Mount host files
explicitly if the command needs them.

### Service

**`service`** — keep a server running on a [remote worker](../code/workers.md#services)
for as long as its flow runs, and publish the url it listens on. Adding one
or changing its settings requires a superuser.

| Setting | Default | Notes |
|---|---|---|
| `command` | — | the server, such as `parrot serve --host 0.0.0.0 --port {port}` |
| `port` | `8100` | where the server listens, `{port}` in the command |
| `worker` | — | label of the worker to run it on, such as `gpu` |
| `health` | `/health` | path that answers 200 once the server is ready |
| `env` | `{}` | environment variables for the server |
| `weights` | — | an artifact digest, fetched to the worker as `{weights}` |

The flow starts it, so it takes no inputs. Every output port carries
`http://<worker>:<port>` once the health check answers 200, and an empty
string the moment the server stops. When it stops, it starts again after a
backoff. Its health reads `degraded` while it starts, restarts or waits for a
worker, `down` after three starts in a row that never got healthy, and `ok`
while it serves.

### File

**`file`** — read a file into the flow, or write one out of it.

| Setting | Default | Notes |
|---|---|---|
| `path` | — | relative to the engine's files directory |
| `mode` | `read` | `read`, `write` or `append` |
| `format` | `text` | `text` or `json` |
| `newline` | `true` | end each written record with a newline |

Paths stay inside the engine's files directory.

## Numeric

### Perceptron

**`mlp`** — a small neural layer over its numeric inputs.

`output = weights @ inputs + biases`, with weights drawn from `seed` so a node
reloads identically.

| Setting | Default | Notes |
|---|---|---|
| `seed` | `0` | |

A worked example of numeric logic, not a modelling tool. To train something,
use a [batch flow](../concepts/runs.md) and a Function node.

## Language

### Wren

**`wren`** — turn a typed command into values for the targets wired to it.

[Wren](https://github.com/stroblme/wren) is an optional local language model
that turns short commands into calls on a fixed list of targets. Install it in
the engine's environment with
`pip install git+https://github.com/stroblme/wren`. The `wren` package on PyPI
is an unrelated one. Without the model the node reports a missing dependency.
[Wren's own docs](https://stroblme.github.io/wren/) explain how to describe
targets, train a model for them, and what a model can and cannot do.

| Input | Type | Notes |
|---|---|---|
| `prompt` | `str` | the command, and the only input that runs the node |
| any other | `bool`, `int`, `float` or `str` | a **reading**: a target a command may ask about but not set |

| Output | Type | Notes |
|---|---|---|
| `actions` | `list` | one record per call |
| `reply` | `str` | what was set or read, as "Kitchen light on." |
| `unhandled` | `str` | the prompt, when the node did nothing with it |
| any other | `bool`, `int`, `float` or `str` | a **target** a command may set |

A target is named by its port and bound to the message it stands for:
`living_room_heating` bound to `home.setpoint` is the living room's heating.
Its name, type and current value are what the model is shown, so the wiring
is the allowlist and the model can reach nothing that is not wired to the
node. A target nothing has set yet is shown as off, as 0 within its range, or
as its first value. A port is named after its message unless it is given a
name of its own, in the node panel's port name field or as
`{"name": "setpoint", "port": "living_room_heating"}` through the flow API.

| Setting | Default | Notes |
|---|---|---|
| `apply` | `false` | write to the targets; off, the node only reports |
| `model` | empty | the trained model, an artifact digest; empty runs wren's demo model |
| `min_confidence` | empty | the confidence a call needs; empty takes the model's own threshold |
| `targets` | empty | by target port, what the model is told of it and how far it is trusted (below) |
| `aliases` | empty | your own words, or words the recogniser hears wrong, in the model's: `{"traverse": "living room light", "page too": "patio"}` |

Every field of a `targets` entry is optional:

| Field | Notes |
|---|---|
| `description` | more words for the target, which help pick it: `"awning over the patio"` |
| `values` | the only values a `str` target takes: `["UP", "DOWN", "STOP"]` |
| `range` | the lowest and highest value of a number: `[5, 30]` |
| `unit` | said after a number in the reply, such as `"°C"`; the model never sees it |
| `min_confidence` | this target's own bar, used over the node's, such as a higher one for a lock |
| `pass_on` | wren never sets it, and a command on it goes on to `unhandled` |

A target that is not a `bool`, `int`, `float` or `str`, a `targets` entry
naming no target, `values` on a target that is not a `str` and a `range` on
one that is not a number are refused when the flow is built.

A command is applied whole or not at all. Its calls are written only when
`apply` is on, every call reaches its target's bar, and none is on a
`pass_on` target or on one the model was not trained on. Applying part of a
command would leave half of a duration ("off for 7 minutes" is off now and on
again later), or let the assistant do again what was already done. Otherwise
nothing is written and `unhandled` carries the prompt, for something else to
answer, such as an [Assistant](#assistant). When no node reads `unhandled`,
`reply` says "Sorry, I can't do that yet." instead, so a spoken command never
goes unanswered.

`reply` says what was set, as "Kitchen light on, living room heating to 21 °C
in 5 minutes.", and answers a read from the target's current value, as
"Outdoor temperature is 6.2 °C.", with `apply` off too, since a read writes
nothing. A read of a target nothing has written yet goes on to `unhandled`. Each record in `actions` contains `message`, `value`,
`applied`, `delay_s`, `call` as wren writes it, such as
`set(kitchen_light, true)`, and that call's `confidence`. Keep `apply=false`
while checking how the model reads commands.

A call with a delay ("in 10 minutes") is held by the engine's work queue, like
the Debounce node's second value: it survives a restart where the queue is in
Redis, and not otherwise.

A model knows the targets it was trained on: their names, types, `values`,
`range`, descriptions and which are readings. A target wired since, or
described otherwise since, is one it never learnt. The node then reads
degraded, naming such targets and asking for the model to be retrained, and a
call on one is never applied. wren's demo model knows the 20 targets of its
[demo home](https://github.com/stroblme/wren/blob/main/wren/home.json), so
it runs without that warning only on ports named and described as there. A
`model` the artifact store does not have or cannot read also reads degraded,
and every command goes on to `unhandled`. A model trained with wren goes into
the store with `PUT /artifacts` (see the [API](../code/api.md#artifacts)).

The command is read in lower case without sentence marks and with number words
as digits ("in five minutes" as "in 5 minutes"), as the model was trained.
`aliases` are applied before, whole words in any case, and `unhandled` carries
the prompt with them.

The chat's [Control tab](../code/agents.md#control) sends commands by hand.

### Assistant

**`assistant`** — answer a request with the instance's AI provider, acting only
on what is wired to it.

| Input | Type | Notes |
|---|---|---|
| `prompt` | `str` | the request, such as a Wren node's `unhandled`; it runs the node |
| `listening` | `bool` | optional; turning true opens the connection and loads the model while the request is spoken |
| any other | any | read as context when a request arrives, as `outside_temperature: 12.5` |

| Output | Type | Notes |
|---|---|---|
| `reply` | `str` | what was set, as "Kitchen light off.", or the answer, one sentence at a time while the model writes it |
| `actions` | `list` | one record per value set: `message`, `value`, `applied`, `delay_s`, `call` |
| any other | `bool`, `int`, `float` or `str` | a **target**: one tool, named after its port, taking a `value` of its type |

| Setting | Default | Notes |
|---|---|---|
| `model` | empty | the model to ask; empty asks the instance's |
| `extra_body` | `{"temperature": 0, "reasoning_effort": "none", "keep_alive": "30m"}` | added to every request as it is |
| `values` | empty | for a `str` target, by port: the only values it takes and what the reply says for each, as `{"canopy": {"UP": "open", "DOWN": "closed"}}` |

It asks the provider set under **Settings → Assistant**. As for Wren, the
wiring is the allowlist: the model is offered the targets and nothing else,
never the tools the chat's assistant uses. Its instructions list the context
inputs and what each target holds now. A target is set the moment the model
calls its tool. The reply then names what was set ("Living room light on,
living room setpoint to 22."), so the model is asked once per request. A
target with `values` is offered only those, each explained to the model, and
confirmed in its word ("Canopy open."). Any other value is refused.

Every request ends in a reply:

- When no tool fits or the request makes no sense, the model says so, as
  "Sorry, I can't do that.".
- When no provider is set up, the connection is refused, it takes over 3
  seconds to connect or 15 seconds to answer, the reply is "The assistant
  isn't available right now." and the node reads degraded. A model that has to
  be loaded first can take about 10 seconds.

The default `extra_body` is for Ollama. `temperature` makes the same request
do the same thing each time. `reasoning_effort` turns thinking off through
Ollama's OpenAI-compatible API, which ignores Ollama's own `think`.
`keep_alive` keeps the model loaded for 30 minutes after the last request
instead of 5. Drop a field the provider refuses.

In a voice flow, `stt` writes the prompt of a Wren node, Wren's `unhandled` is
the assistant's prompt, both `reply` outputs go to `tts`, and the device's
`listening` warms the assistant.

### Speech to text

**`stt`** — turn a microphone's audio into text, through a speech server.

| Input | Type | Notes |
|---|---|---|
| `audio` | `audio` | 16 kHz mono PCM16 chunks (`audio/L16`), or a whole clip such as a WAV |
| `listening` | `bool` | true while the device listens; false ends the utterance |
| `url` | `str` | the speech server, from a [Service](#service) node; read, never waited for |

| Output | Type | Notes |
|---|---|---|
| `partial` | `str` | the words agreed on so far, as they come, then the final text |
| `text` | `str` | what was said, once per utterance |

The server is any that speaks OpenAI's audio API, such as `parrot serve`. Each
utterance is one realtime transcription session (`{url}/v1/realtime`), opened
when `listening` turns true or at the first chunk and committed when it turns
false. A clip is sent whole to `{url}/v1/audio/transcriptions`. The node
returns at once and publishes when the server answers, since a device's next
chunk waits for it. An utterance that comes back empty, fails, or gets no
chunk for 10 seconds publishes "Didn't catch that." on `partial` alone, so a
caption shows it and nothing speaks it. A lost one is logged.

`listening` has to hold a value before a chunk wakes the node. Declare it as a
flow input with the initial value `false` when a device may never send it,
such as one that sends whole clips.

### Text to speech

**`tts`** — speak text as audio for a speaker, through a speech server.

| Input | Type | Notes |
|---|---|---|
| `text` | `str` | what to say, a whole reply or one sentence at a time |
| `url` | `str` | the speech server, as for Speech to text |

| Output | Type | Notes |
|---|---|---|
| `speech` | `audio` | one MP3 reference per sentence |

| Setting | Default | Notes |
|---|---|---|
| `voice` | `af_heart` | one of Kokoro's voices, such as `bm_george` |

Each sentence is one call to `{url}/v1/audio/speech` and published as soon as
it is spoken. Units after a number and clock times are written out first,
since a speech model reads them letter by letter: "22 °C" is said "22
degrees", "3.4 kWh" "3.4 kilowatt hours" and "07:30" "7 30". How a word
itself is pronounced is the server's: `parrot serve --lexicon` gives words
their phonemes. Calls to one server share a connection, which Speech to text opens
when an utterance starts, so a reply's first sentence does not wait for one. A sentence spoken before in the same voice is reused rather than
spoken again, for the last 256 sentences. A reference carries `"append": true`
while the sentences before it are still playing, which queues it on a panel's
`speaker`. The first sentence of a reply carries none and cuts off whatever
was playing.

## Devices

### Device

**`device`** — a paired device's own ports: what it publishes, what it follows.

The engine never runs it. Its outputs are the device's `in` ports, published by
the device itself, and its inputs the `out` ports the device follows, which
never wake anything. Each port is named by the device, such as `gpio.4` or
`backlight`. See [the device's own ports](../interface/dashboards.md#the-devices-own-ports).

| Setting | Default | Notes |
|---|---|---|
| `device` | | the device's id |
| `config` | `{}` | per port, a JSON object handed to the device unread; a reported port's settings get a form |

The node is degraded, *Offline*, while the device is not connected: amber on
the canvas and a warning on Home, never a fault.

## Connectors

Installed connector packages add types through the `fluksio.node_types`
entry point group. The editor shows the source package. Incompatible contract
versions are skipped; installing or upgrading a connector requires an engine
restart.

The contract itself is in [The connector contract](connector-contract.md), and
[Writing a connector](../code/connectors.md) walks through building one.

## See also

- [Payload types](payload-types.md) — what a port may carry
- [Writing node code](../code/nodes.md) — the Function node
- [The flow editor](../interface/flow-editor.md) — placing and configuring them