From bd2c4b66e9b752b0315745c1b4ec52851566c032 Mon Sep 17 00:00:00 2001 From: stroblme Date: Mon, 24 Aug 2026 12:13:40 +0200 Subject: [PATCH] Publish the connector contract and authoring guide Adds reference/connector-contract.md and code/connectors.md to the site, with the node-type reference finally linking the contract it describes. Strict build passes. Co-Authored-By: Claude Opus 5 (1M context) --- ROADMAP.md | 4 +- docs/code/connectors.md | 130 ++++++++++++++++++ docs/reference/connector-contract.md | 189 +++++++++++++++++++++++++++ docs/reference/node-types.md | 3 + zensical.toml | 2 + 5 files changed, 326 insertions(+), 2 deletions(-) create mode 100644 docs/code/connectors.md create mode 100644 docs/reference/connector-contract.md diff --git a/ROADMAP.md b/ROADMAP.md index 8e1dc70..5915ac0 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -69,8 +69,8 @@ external interfaces. See `docs/architecture/structure.canvas` → *Backend – M a polling coordinator that deduplicates, `x-secret` parameters the editor renders as a secret picker, and health reporting. Connectors are installed packages found through the `fluksio.node_types` entry point group; the - contract is documented in `docs/connectors/` with a working skeleton at - `connectors/skeleton/`. The registry follows later + contract is documented at `docs/reference/connector-contract.md` with a + working skeleton at `connectors/skeleton/`. The registry follows later - [x] First real connectors written against that contract from outside the engine: WF-RAC aircon, calendar, UniFi presence and Art-Net, in `connectors/`. Built by `make connectors` and installed into the image. Calendar and diff --git a/docs/code/connectors.md b/docs/code/connectors.md new file mode 100644 index 0000000..aa111b1 --- /dev/null +++ b/docs/code/connectors.md @@ -0,0 +1,130 @@ +# Writing a connector + +A walkthrough from an empty directory to a node on the canvas. The normative +surface is in [the connector contract](../reference/connector-contract.md); +this is how to use it. + +## Start from the skeleton + +A working connector, kept as small as a correct one can be, lives at +`skeleton/` in the [connectors repository](https://git.stroblme.de/Fluksio/connectors). +Copy it: + +```bash +git clone https://git.stroblme.de/Fluksio/connectors.git +cp -r connectors/skeleton fluksio-connector-mydevice +cd fluksio-connector-mydevice +``` + +Rename the package (`src/fluksio_connector_skeleton/` and the `[project]` name +in `pyproject.toml`), then point the entry point at your class: + +```toml +[project.entry-points."fluksio.node_types"] +my_device = "fluksio_connector_mydevice:MyDevice" +``` + +## Install it into the engine + +Connectors are found through installed package metadata, so the engine has to +have yours installed. During development, install it as an editable path +dependency of the backend: + +```bash +cd app/backend +uv add --editable ../../fluksio-connector-mydevice +``` + +Editing your connector's **code** then takes effect on the next engine restart, +with no reinstall. Changing its **metadata** — the entry point, the version, +the package name — needs `uv sync` to regenerate the installed metadata before +a restart picks it up. + +Two things to know about this: + +- `uv add` writes the path dependency into `backend/pyproject.toml` and + `uv.lock`. That path does not exist inside the backend's Docker image, so + **revert both files before committing**; the dependency belongs on your + machine, not in the repo. Distribution is a later milestone. +- For the same reason, develop against the **standalone backend** rather than + the Docker stack: + + ```bash + cd app && make dev-backend # FastAPI on :8000 + make dev-frontend # Vite on :5173 + ``` + +Restart the backend and your node type appears in the add-node palette (⌘K), +labelled with the package it came from. + +## Write the node + +```python +from typing import Any + +from pydantic import Field +from fluksio.flow.connector import CONTRACT_VERSION, ConnectorNode + + +class MyDevice(ConnectorNode): + contract = CONTRACT_VERSION + title = "My device" + description = "Reads temperature and humidity over the device's HTTP API." + + class Params(ConnectorNode.Params): + host: str = "192.168.1.50" + api_key: str | None = Field( + default=None, json_schema_extra={"x-secret": True} + ) + + async def poll(self) -> dict[str, Any] | None: + reading = await self._read() + return {"temperature": reading.temp, "humidity": reading.humidity} +``` + +Then, on the canvas: place the node, set `poll_interval` to how often the +device should be read, pick the API key from the secrets store, and add output +ports named `temperature` and `humidity`. The port names are the wiring — any +node consuming those messages is downstream of this one. + +## If the device is told rather than asked + +A connector that commands something implements `write()` instead of, or +alongside, `poll()`. Input ports carry the values in: + +```python + def write(self, **ports: Any) -> None: + self._device.set(ports["level"]) +``` + +Place the node with *input* ports named `level`, and whatever publishes that +message drives the device. Set `idempotent = False` on the class: a write is a +command, and a redelivery after a crash should not undo a newer one. + +Keep the first version off the wire. A boolean setting the code checks before +it sends — `artnet`'s `transmit` is the example — lets a flow be built and +watched in the logs before anything physically moves. + +## Try it without a device + +Give `poll` something predictable first and confirm the values reach the canvas +before pointing it at hardware: + +```python +async def poll(self): + return {"temperature": 21.5} +``` + +Press Run, or wait for the poll interval, and the value appears on the edge. +Anything the connector prints shows up in the logs panel in the dock, and a +raised exception shows there with its traceback and marks the node down. + +## Checklist before sharing it + +- `contract = CONTRACT_VERSION` declared on the class itself. +- Credentials marked `x-secret`, never read from the environment. +- `stop()` is safe to call twice, and after a failed `start()`. +- Blocking I/O wrapped in `asyncio.to_thread`. +- `poll()` returns `None` rather than raising when there is simply nothing new. +- `write()` returns quickly; the thread it runs on is the scheduler's. +- Health reported when the connection comes and goes. diff --git a/docs/reference/connector-contract.md b/docs/reference/connector-contract.md new file mode 100644 index 0000000..9142961 --- /dev/null +++ b/docs/reference/connector-contract.md @@ -0,0 +1,189 @@ +# 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](../code/connectors.md). + +## Contract version + +```python +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: + +```toml +[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 + +```python +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: + +```python +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 — 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 `MessageSpec`s 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. + +## Polling + +Set `poll_interval` and implement `poll()`; the base class runs the loop: + +```python +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. +- 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: + +```python +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. + +## Lifecycle + +```python +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 + +```python +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. The +engine forwards changes to the editor, which shows them on the node. Reporting +the same status twice is free — only changes are published. 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. diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md index ec6e3dc..998edd1 100644 --- a/docs/reference/node-types.md +++ b/docs/reference/node-types.md @@ -296,6 +296,9 @@ engine does not speak is ignored rather than half-loaded. Installing or upgrading one takes effect on the next engine restart, because Python does not re-import a changed module and a rescan would promise more than it delivers. +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 diff --git a/zensical.toml b/zensical.toml index df3dc21..a1715de 100644 --- a/zensical.toml +++ b/zensical.toml @@ -43,12 +43,14 @@ nav = [ { "Code and the CLI" = [ { "The fluksio command" = "code/cli.md" }, { "Writing node code" = "code/nodes.md" }, + { "Writing a connector" = "code/connectors.md" }, { "The HTTP API" = "code/api.md" }, { "Remote workers" = "code/workers.md" }, { "Agents over MCP" = "code/agents.md" }, ] }, { "Reference" = [ { "Node types" = "reference/node-types.md" }, + { "The connector contract" = "reference/connector-contract.md" }, { "Payload types" = "reference/payload-types.md" }, { "Configuration" = "reference/configuration.md" }, ] },