Writing a connector¶

A connector adds a device or service integration to the node palette. It is a Python package with a ConnectorNode subclass and an entry point. This example works without hardware or another repository.

Create the package¶

Make this directory beside your project:

fluksio-connector-random/
  pyproject.toml
  src/
    fluksio_connector_random/
      __init__.py

Put this in pyproject.toml:

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "fluksio-connector-random"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["fluksio>=0.1,<0.2"]

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

[tool.hatch.build.targets.wheel]
packages = ["src/fluksio_connector_random"]

Put this in src/fluksio_connector_random/__init__.py:

import random

from pydantic import Field

from fluksio.flow.connector import CONTRACT_VERSION, ConnectorNode


class RandomSensor(ConnectorNode):
    contract = CONTRACT_VERSION
    title = "Random sensor"
    description = "Emits a random reading. No sensor required."

    class Params(ConnectorNode.Params):
        ceiling: float = Field(default=1.0, gt=0)

    async def poll(self):
        return {"reading": random.random() * self.config.ceiling}

The entry point name, random_sensor, becomes the stored node type. Choose a unique name; the engine skips names that are already registered.

Install and try it¶

Use the engine's Python environment, the one where you installed Fluksio:

pip install -e /path/to/fluksio-connector-random

Restart the engine from its original project directory so it keeps using the same instance data:

fluksio serve

Restart an already running engine after installing. In the browser:

1. Add a Random sensor node to a live flow.
2. Set poll_interval to 5 seconds.
3. Add an output port named reading, type float.
4. Publish the flow and watch the value change.

The palette shows the package name and version. Code edits need an engine restart; metadata changes also need another editable install.

For Docker, install the package in the backend image and rebuild it. The Modules screen installs packages for node code, so it does not install connectors into the engine.

Connect a real device¶

Replace poll() with your device read. Return a dictionary keyed by declared output port, or None when there is nothing new. The base class polls at poll_interval, publishes changed values, and reports failures.

Keep blocking I/O off the event loop with asyncio.to_thread. For pushed updates, manage the connection in start() and stop(), leave poll_interval=0, and call self.inject({...}) when data arrives.

Mark credential parameters with Field(default=None, json_schema_extra={"x-secret": True}). The editor then stores a secret reference rather than the credential in the flow.

To send commands, implement write(self, **ports): it receives values keyed by input port. Set idempotent = False for commands that must not be replayed after a crash.

Media and persistent state¶

Use self.save_artifact(data, name, media_type=...) for bytes such as camera frames. Return the reference through an image, audio, or video port. For disposable live frames, pass volatile=True.

Use self.store for short strings that must survive a restart, such as a pairing token. Include the device address in the key: all nodes from the same top-level package share this store.

Before sharing¶

• Declare contract = CONTRACT_VERSION on the class itself.
• Return only declared output ports, with values matching their types.
• Mark credentials as secrets.
• Keep stop() safe after a failed start and when called twice.
• Report connection health if you manage your own subscription.

See the connector contract for lifecycle, storage limits, and certificate pinning.