Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Publish the connector contract and authoring guide

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
stroblme
2026-08-24 12:13:40 +02:00
co-authored by Claude Opus 5
parent 607f3a814e
commit bd2c4b66e9
5 changed files with 326 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ROADMAP.md
+2 -2
View File
@@ -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
docs/code/connectors.md
+130
View File
@@ -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.
docs/reference/connector-contract.md
+189
View File
@@ -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.
docs/reference/node-types.md
+3
View File
@@ -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
zensical.toml
+2
View File
@@ -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" },
] },