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.

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, 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.