app/docs/reference/connector-contract.md

The connector contract

A connector is the device-facing node class. It talks to something outside the engine (a device, a service, a protocol) and publishes what it finds as ordinary typed messages, so the rest of a flow cannot tell the difference between a reading from a heat pump and one from a print statement.

This document is normative: everything here is what the engine relies on and what a connector may rely on in return. The authoring walkthrough is in Writing a connector.

Contract version

from fluksio.flow.connector import CONTRACT_VERSION, ConnectorNode

class MyConnector(ConnectorNode):
    contract = CONTRACT_VERSION

contract must be declared on the connector class itself. The loader reads it from the class __dict__ rather than through inheritance, so a connector written against an older base cannot pass itself off as current. A mismatch is logged and the connector is skipped; nothing else in the engine changes.

The version is bumped only when a change would break connectors written against the old surface. Additions that leave existing connectors working do not bump it.

Packaging and discovery

A connector is an ordinary Python package. It advertises its node class through an entry point in the fluksio.node_types group:

[project]
name = "fluksio-connector-skeleton"
version = "0.1.0"

[project.entry-points."fluksio.node_types"]
random_sensor = "fluksio_connector_skeleton:RandomSensor"

• The entry point name is the node type as it appears on the canvas and in a stored flow. It must not collide with a built-in type (python, mqtt, http, influxdb, delay, mlp); a collision is logged and skipped.
• The package name and version are the manifest. There is no second metadata file to keep in step, and the editor shows the provenance (fluksio-connector-skeleton 0.1.0) on the node type.

Discovery happens once, when the engine starts. A connector's code is imported, and Python does not re-import a changed module, so installing or upgrading one means restarting the engine.

Declaring what a connector is

class RandomSensor(ConnectorNode):
    contract = CONTRACT_VERSION
    title = "Random sensor"          # shown in the add-node palette
    description = "Emits a random reading."

title falls back to the entry point name; description may be empty.

Parameters

Settings are a pydantic model, and the editor builds a form from its JSON schema. Subclass the base so poll_interval stays available:

class Params(ConnectorNode.Params):
    host: str = "localhost"
    api_key: str | None = Field(default=None, json_schema_extra={"x-secret": True})

• Booleans render as switches, numbers and strings as inputs. Objects and arrays are not rendered, so keep settings flat.
• A field carrying x-secret renders as a picker over the stored secrets and writes a reference, {"$secret": "name"}, rather than the value. The engine resolves it when the node is built, so a credential never lands in flow.json. Always mark credentials this way.
• The parsed model is available as self.config.

Ports and messages

Ports are declared by whoever places the node, not by the connector, and they are MessageSpecs like any other node's: a message name, a DType, and an optional interval.

• poll() returns a dict keyed by output port, and the base class maps it onto message names; write() receives a dict keyed by input port the same way.
• Every value is checked against the port's declared DType on the way out. A wrong type is an error, not a hint.
• interval on a port is enforced by the engine, not by the connector: an output port publishes at most every n seconds, an input port wakes its node at most that often. A connector should poll at the rate the device is comfortable with and leave delivery rates to whoever wires it up.
• A key no port declares is an error, and it fails the whole reading rather than the one value. A mistyped metric name is how a training curve goes missing.

When the device decides what the ports are

A connector for a device whose readings vary by model (which components a relay has, which entities were flashed onto a board) cannot know its ports in advance, and returning everything the device reports would fail on the first value nobody bound. Narrow the reading to the ports that were declared:

declared = {spec.port for spec in self.output_ports if spec.name}
return {port: value for port, value in reading.items() if port in declared}

spec.port is the local, unqualified name and stays that way for the node's whole life, so the set can be taken fresh each time. Say something in the log when a declared port is not one the device has, once rather than once per poll: from the canvas a renamed entity and a typo look the same, and both leave a port silent forever.

Polling

Set poll_interval and implement poll(); the base class runs the loop:

async def poll(self) -> dict[str, Any] | None:
    return {"reading": await self._read_device()}

• Return None when there is nothing new.
• Only changed values are published. A device polled every few seconds usually says the same thing, and every publication wakes everything downstream, so the loop compares against what it last published, meaning what it actually published, so a publication that failed is retried next tick rather than counting as said.
• Raising is not fatal: it is reported as a health problem and retried on the next tick.
• The loop calls inject, which runs the graph, on a worker thread. poll() itself runs on the event loop and must not block it.

A connector that is pushed to rather than polled leaves poll_interval at 0, overrides start/stop to open and close its own subscription, and calls self.inject({...}) when something arrives.

Writing

A connector that commands something rather than only reading it implements write(). It is called when the node's inputs are satisfied, with the values keyed by input port:

def write(self, **ports: Any) -> dict[str, Any] | None:
    self._device.set(ports["level"])
    return None

• Settings are already parsed on self.config; write() is handed ports only.
• Return None unless the device answers something worth publishing, in which case return it keyed by output port, exactly as poll() does.
• It runs on a worker thread rather than the event loop, but the thread is the scheduler's, so a long block holds up the cascade. Queue the work if the device is slow.
• A connector that only reads leaves it alone.
• poll() and write() are independent: a connector may have both, and a node with inputs and no poll_interval never polls.

A write is a command, not a value: set idempotent = False on the class so a redelivery after a crash does not undo a newer command that already landed.

Devices whose readings are bytes

A camera frame or a recorded clip is far too big to be a message, so a connector publishes a reference to it instead:

async def poll(self) -> dict[str, Any] | None:
    jpeg = await asyncio.to_thread(self._grab)
    return {
        "frame": self.save_artifact(
            jpeg, "frame.jpg", media_type="image/jpeg", volatile=True
        )
    }

save_artifact stores the bytes and returns what an image, audio or video port carries: the media type has to match the port's type. It only works once the node has started, since the store is the engine's and is handed over then.

volatile=True is what a camera publishes with. The frame goes to a ring in memory rather than the data volume and is pushed down the websocket to whichever screens are drawing it, so a wall panel sees ten frames a second and the SD card under it is never written to. Frames last until newer ones need the room; leave the flag off for a reading somebody asked to keep.

Each reading is a new artifact, which the poll loop publishes because its digest differs from the last. Set poll_interval to what somebody actually wants to look at. Through a portal the bytes are fetched rather than pushed, so a remote panel wants a frame every second or two; higher rates than that are for the local network, and full-rate video still belongs on the camera's own stream.

Keeping state across restarts

A connection, a session or a device handle lives in the connector's own attributes and goes with the process. The few things that must not, such as the certificate a device presented on first contact or a token it paired with, go in self.store:

pem = await asyncio.to_thread(self.store.get, f"cert:{self.config.host}")
await asyncio.to_thread(self.store.set, f"cert:{self.config.host}", pem)
await asyncio.to_thread(self.store.delete, f"cert:{self.config.host}")

• Keys and values are strings; get returns None for a key that is not there, and delete returns whether there was one.
• The store belongs to the connector's top-level package, not to a node: every node of the package sees the same keys, so two nodes on one bridge share what was stored for it. Put the device in the key.
• It lives in the engine's database, so it survives a restart and needs nothing fluksio serve does not have. It is available before start.
• Limits: a key is 1 to 200 characters without / or whitespace, a value at most 64 kB, and a package at most 256 keys. Past any of them, set raises ValueError. Bytes worth keeping belong in save_artifact.
• Every call is a database round trip, so on the event loop it goes through asyncio.to_thread.

A superuser can see the keys (never the values) and clear them, all of them or one at a time, under Stored state in a connector node's settings, or through the API, which names the connector by any node type it installs:

curl -s $FLUKSIO/connectors/hue/state -H "Authorization: Bearer $TOKEN"
curl -X DELETE $FLUKSIO/connectors/hue/state/cert:10.0.0.2 -H "Authorization: Bearer $TOKEN"
curl -X DELETE $FLUKSIO/connectors/hue/state -H "Authorization: Bearer $TOKEN"

So a connector has to cope with a key disappearing: clearing is how somebody tells it to start over.

Pinning a device's certificate

A device on the local network presents a certificate no system trusts and that names a serial or a MAC rather than its address, so verify=True fails and verify=False trusts anyone. fluksio.flow.connector_tls trusts the certificate the device presented on first contact instead, kept in self.store under cert:<address>:

from fluksio.flow.connector_tls import context_for, pinned_certificate, raise_if_changed

address = f"{self.config.host}:{self.config.port}"
pem = pinned_certificate(self.store, address, self.config.timeout)  # blocking
try:
    with httpx.Client(verify=context_for(pem)) as client:
        ...
except httpx.ConnectError as exc:
    raise_if_changed(exc, pem, address, self.config.timeout, "my_node_type")
    raise

• First contact sends a TLS hello and nothing else; a device that does not complete one raises OSError and nothing is pinned.
• Any other certificate is refused in the handshake, before a key or a command is sent. raise_if_changed turns that refusal into CertificateChanged, which names the new SHA-256 and the key to clear, under the node type given.
• The pinned certificate's dates are not checked: a pin trusts that exact certificate, so one that expired, or an engine clock that is off, does not lock the device out.
• Clearing the key accepts whatever the device presents next.

Lifecycle

async def start(self, app=None) -> None: ...
async def stop(self, app=None) -> None: ...

• start is called when the connector's flow starts, and after every rebuild.
• stop is called before a rebuild and when the flow is stopped. It must be idempotent: it is called whether or not start succeeded.
• Overriding either means calling super() if the polling loop is also wanted.
• The app argument is the FastAPI application, for the rare connector that needs to mount a route. Most ignore it.

A stopped flow gets no start at all: that is what stopping it means.

Health

self.report_health("ok")
self.report_health("degraded", "3 of 5 registers timed out")
self.report_health("down", str(exc))

Three values, ok, degraded and down, plus an optional detail string. Reporting the same status twice is free, since only changes are published. A node reporting down is named among its flow's issues and counted on the health summary on Home; degraded means still working, and is not. The polling loop already reports around poll(); a connector managing its own connection should report when it connects and when it loses the connection.

Health is about the connection, not about loading: a connector that fails to load is reported separately and never runs.

What a connector must not do

• Hold state that belongs in the flow. Logic nodes are stateless and run in parallel; a connector may hold a connection, a session, or a device handle, and keep what has to outlive a restart in self.store, but a value another node needs belongs in a message.
• Block the event loop. poll, start and stop are async and run on the loop. Use asyncio.to_thread for blocking I/O.
• Read credentials from the environment. They come through parameters, so the person deploying a flow can change them without touching the host.