The connector contract¶

Connectors integrate devices and services through typed messages. This reference covers the contract the engine supports; start with Writing a connector for a working package.

Contract version¶

from fluksio.flow.connector import CONTRACT_VERSION, ConnectorNode


class MyConnector(ConnectorNode):
    contract = CONTRACT_VERSION
    title = "My connector"
    description = "Reads a device."

Declare contract on the class itself: inheritance does not count. The current contract version is 1. A mismatch is logged and the connector is skipped. Breaking changes bump the version; compatible additions do not.

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

Packaging and discovery¶

Register the class in your package's pyproject.toml:

[project.entry-points."fluksio.node_types"]
my_device = "fluksio_connector_mydevice:MyConnector"

The entry point name is the stored node type. Names already registered, including built-in types, are skipped. The editor shows the package name and version from its installed metadata.

Discovery runs at engine startup. Install connectors in the engine's environment and restart after installation or code changes.

Parameters¶

Subclass ConnectorNode.Params to keep poll_interval and add your settings:

from pydantic import Field


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

The node panel builds a form from this model's JSON schema: switches for booleans, inputs for numbers and strings, and JSON fields for objects and arrays. Parsed settings are available as self.config.

Mark credentials with x-secret. The editor stores a reference such as {"$secret": "device-key"}, and the engine resolves it while building the node.

Ports and messages¶

The flow author declares ports as MessageSpec values. Each has a message name, a DType, and optional delivery settings.

• poll() returns values keyed by output port.
• write() receives values keyed by input port.
• The base class maps local port names to message names and validates types.
• Unknown output keys fail the reading. Return only declared ports.
• Port interval limits are enforced by the engine; the connector controls how often it reads the device.

For devices with variable readings, filter to the configured outputs:

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}

Log missing device fields once so a typo does not leave a port silently empty.

Polling¶

async def poll(self):
    return {"reading": await self._read_device()}

Set poll_interval in seconds; 0 disables polling. Return None when there is nothing new. The base loop publishes only changed values and retries a failed reading or publication on the next tick. It reports connection health around each poll.

poll() runs on the event loop. Use asyncio.to_thread for blocking reads. The base loop already calls inject on a worker thread.

For pushed readings, leave poll_interval=0, manage the subscription in start() and stop(), and call self.inject({...}) when values arrive. Offload inject too when calling it from the event loop.

Writing¶

def write(self, **ports):
    self._device.set(ports["level"])
    return None

Settings stay on self.config; arguments contain input ports only. Return None, or a dictionary keyed by output port if the device responds with useful values.

Writes run on a scheduler thread. Return promptly or queue slow work. Polling and writing are independent; a connector may support both.

Set idempotent = False for commands that must not be redelivered after a crash. Leave write() unchanged on a read-only connector.

Devices whose readings are bytes¶

import asyncio


async def poll(self):
    jpeg = await asyncio.to_thread(self._grab)
    return {
        "frame": self.save_artifact(
            jpeg, "frame.jpg", media_type="image/jpeg", volatile=True
        )
    }

save_artifact(data, name="", media_type="application/octet-stream", volatile=False) returns a reference for an artifact or media port. Match the media type to an image, audio, or video port.

The artifact store is available after start(). Use volatile=True for disposable live frames: they go to a bounded memory ring and expire as new frames arrive. Use durable storage for recordings.

Local screens receive pushed live frames; portal screens fetch the bytes. Choose a modest poll rate for remote viewing.

Keeping state across restarts¶

Keep connections and device handles in node attributes. Store short strings such as pairing tokens or pinned certificates in self.store:

value = await asyncio.to_thread(self.store.get, "token:device")
await asyncio.to_thread(self.store.set, "token:device", "value")
await asyncio.to_thread(self.store.delete, "token:device")

• Keys and values are strings. get returns None for a missing key; delete returns whether it existed.
• All nodes from the connector's top-level package share the namespace. Include a device identifier in each key.
• Storage is in the engine's database and is available before start().
• Keys allow 1–200 characters, without / or whitespace. Values allow 64 KiB and each package allows 256 keys; exceeding a limit raises ValueError.
• Calls block on the database: use asyncio.to_thread on the event loop.

Superusers can list keys, delete one, or clear them all under Stored state in the node panel. Values are never returned. The equivalent API paths are:

Handle missing keys: clearing one asks the connector to pair or learn again.

Pinning a device's certificate¶

fluksio.flow.connector_tls supports trusting a local device's certificate on first contact:

import httpx

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)
try:
    with httpx.Client(verify=context_for(pem)) as client:
        response = client.get(f"https://{address}/status")
        response.raise_for_status()
except httpx.ConnectError as exc:
    raise_if_changed(exc, pem, address, self.config.timeout, "my_device")
    raise

Run this blocking work off the event loop. First contact pins the certificate under cert:<address>; a changed certificate is refused before sending a credential or command. raise_if_changed reports CertificateChanged with the fingerprint and key to clear.

Pins trust the exact certificate, including after its expiry. Clearing the key accepts the certificate presented on the next connection.

Lifecycle and health¶

async def start(self, app=None):
    await super().start(app)


async def stop(self, app=None):
    await super().stop(app)

start runs when a flow starts or rebuilds. stop runs before rebuilding or stopping, even after a failed start; make it idempotent. Call super() when overriding these hooks if you want the base polling loop.

The optional app is the FastAPI application, for connectors that mount routes. Most connectors can ignore it.

Report connection changes if you manage your own subscription:

self.report_health("ok")
self.report_health("degraded", "Some readings timed out")
self.report_health("down", "Connection lost")

Repeated statuses do not emit another event. down appears among flow issues and on Home; degraded means the connection still works. Loading failures are reported separately.

Keep shared flow values in messages, credentials in secret parameters, and blocking I/O off the event loop.