Worker Messages (Configuration RPC)

A FlowBox configuration UI often needs to talk to the running worker: validate a connection string, fetch a list of topics, or echo a value back to the user. Worker Messages is the request/response channel between the configuration microfrontend and a live worker instance. On the Python side the entry point is FlowBoxWorkerBridge (module industream.flowmaker.sdk.worker_bridge, re-exported from industream.flowmaker.sdk), the Python sibling of the TypeScript and C# FlowBoxWorkerBridge.

[NOTE!lightbulb/SCOPE] This page covers the worker side (Python). The Svelte side—how a configuration bundle sends a request and how the platform correlates it—is documented in Worker Messages from the Configuration UI. Read that first if you have not.

Request and response flow

Worker Messages request and response flow

The platform UI owns authentication, worker selection, request IDs, timeouts and cleanup.
The worker only sees a WorkerRequest on a Socket.IO namespace and answers with a matching WorkerResponse.
The bridge dispatches each request to the router registered for its boxVersionId — format "<box_id>/<version>", e.g. industream/data-logger/1.0.0, derived by the builder as f"{registration_info.id}/{registration_info.current_version}".
The requestId is echoed back unchanged, so concurrent requests stay isolated.
The wire events are the constants on WorkerBridgeEvents (REQUEST = "worker:request", RESPONSE = "worker:response", HEALTH = ":health"); you never emit or listen to them yourself.
A request whose requestId, boxVersionId, or event is missing or empty fails WorkerRequest.from_dict and is dropped without any response — the UI-side promise can only time out.

Registering the bridge

Create a bridge, register one WorkerConfigHandlers router per box version, and hand it to the builder with use_worker_bridge. The real py-data-logger worker does exactly this in main.py:

from industream.flowmaker.sdk import (
    FlowBoxWorkerBridge,
    FlowMakerWorkerBuilder,
    FlowMakerWorkerOptions,
)

bridge = FlowBoxWorkerBridge()
bridge.register_handlers(f"{BOX_ID}/{VERSION}", create_configuration_handlers())

FlowMakerWorkerBuilder() \
    .configure(configure_options) \
    .declare_flowbox(DataLoggerBox) \
    .use_worker_bridge(bridge) \
    .build() \
    .run()

The bridge constructor is FlowBoxWorkerBridge(socket_url_prefix: str | None = None, handler_timeout_seconds: float = 30.0, transport_factory: Callable[[str, str], WorkerRpcTransport] | None = None). The Socket.IO base URL defaults to FM_WORKER_LOG_SOCKET_IO_ENDPOINT (or http://localhost:4123); the handler timeout defaults to 30 seconds. transport_factory is the test seam: swap in an in-memory WorkerRpcTransport to unit-test dispatch without a Socket.IO server.

[INFO!info/AUTO CONNECT] You never call bridge.connect(worker_id, box_version_ids) yourself. After the worker registers with the runtime, FlowMakerWorker.run_async() calls it automatically, using the FM_WORKER_ID environment variable as worker_id and the IDs of the declared flow boxes as the served box versions, and calls bridge.disconnect() on shutdown. connect/disconnect are only public for advanced or test scenarios.

Handling events

A handler receives the request payload (the raw deserialized JSON object, typically a dict) and returns the response payload, synchronously or as a coroutine. The bridge itself parses the envelope into a frozen WorkerRequest dataclass (request_id, box_version_id, event, payload), but handlers only ever see the payload—there is no typed payload deserialization step in the Python SDK.

The `WorkerConfigHandlers` router (the only registration shape)

WorkerConfigHandlers is a fluent per-event router. Map each event name to a handler with .on(event, handler), then pass the router to register_handlers. An unregistered event is answered with UNKNOWN_EVENT.

from typing import Any
from industream.flowmaker.sdk import WorkerBridgeException, WorkerConfigHandlers

DOUBLE_EVENT = "data-logger:double"

def _double(payload: Any) -> str:
    raw = payload.get("value") if isinstance(payload, dict) else None

    try:
        value = float(raw)
    except (TypeError, ValueError):
        raise WorkerBridgeException(f"Invalid number: {raw}")

    doubled = value * 2
    return f"your doubled number is : {int(doubled) if doubled.is_integer() else doubled}"

def create_configuration_handlers() -> WorkerConfigHandlers:
    return WorkerConfigHandlers().on(DOUBLE_EVENT, _double)

This is the actual handler module of py-data-logger. Because the payload arrives as a plain dict, validate it yourself and raise WorkerBridgeException with a clear message—the SDK has no INVALID_PAYLOAD equivalent that fires automatically.

Async handlers work the same way; the router awaits the result if the handler returns an awaitable:

async def _fetch_topics(payload: Any) -> dict[str, Any]:
    async with httpx.AsyncClient() as client:
        response = await client.get(payload["url"], timeout=5)
        response.raise_for_status()
        return {"topics": response.json()}

handlers = WorkerConfigHandlers().on("my-box:topics", _fetch_topics)

There is no register_class or raw-delegate overload in Python (unlike C# and TypeScript): register_handlers(box_version_id, handlers) accepts exactly one WorkerConfigHandlers per box version. unregister_handlers(box_version_id) removes a router again and returns whether there was one.

Error handling

Raise a WorkerBridgeException(message, code=None) from any handler to return a typed error. The bridge catches it, puts message and code in the response's error (an omitted code becomes HANDLER_ERROR), and the platform rejects the Svelte-side request() promise with that message. Any other exception is reported as HANDLER_ERROR with str(ex) as the message.

raise WorkerBridgeException("Connection refused by device", "DEVICE_REFUSED")

The reserved error codes are defined on WorkerBridgeErrorCodes:

Code	Constant	When
`UNKNOWN_BOX`	`UNKNOWN_BOX`	`boxVersionId` is not served by this worker
`NO_HANDLER`	`NO_HANDLER`	Box is served but has no router registered
`TIMEOUT`	`TIMEOUT`	Handler exceeded the configured timeout (default 30s)
`HANDLER_ERROR`	`HANDLER_ERROR`	Handler raised anything else (default)
`INVALID_PAYLOAD`	`INVALID_PAYLOAD`	Reserved—never raised by the Python bridge itself
`UNKNOWN_EVENT`	`UNKNOWN_EVENT`	Event not registered on the `WorkerConfigHandlers` router

[WARNING!warning/NO_HANDLER DIFFERS FROM .NET] For a served box version without a registered router, the Python bridge answers NO_HANDLER. The .NET SDK inserts a default throwing handler in that situation and yields HANDLER_ERROR instead—don't rely on identical error codes across languages.

The `:health` event

The bridge answers :health with { status: "ok" } for every served box version, before dispatching to any handler (but after the UNKNOWN_BOX check).

[NOTE!info/AUTO HEALTH] Do not implement :health yourself. It is answered automatically by FlowBoxWorkerBridge (see WorkerBridgeEvents.HEALTH). A configuration bundle can always ping it to check that a matching worker is live.

Full worked example

The two pieces from py-data-logger together: the router module and the wiring.

# data_logger_configuration_handler.py
from typing import Any
from industream.flowmaker.sdk import WorkerBridgeException, WorkerConfigHandlers

DOUBLE_EVENT = "data-logger:double"

def _double(payload: Any) -> str:
    raw = payload.get("value") if isinstance(payload, dict) else None

    try:
        value = float(raw)
    except (TypeError, ValueError):
        raise WorkerBridgeException(f"Invalid number: {raw}")

    doubled = value * 2
    return f"your doubled number is : {int(doubled) if doubled.is_integer() else doubled}"

def create_configuration_handlers() -> WorkerConfigHandlers:
    return WorkerConfigHandlers().on(DOUBLE_EVENT, _double)

# main.py
from industream.flowmaker.sdk import (
    FlowBoxWorkerBridge,
    FlowMakerWorkerBuilder,
    FlowMakerWorkerOptions,
)
from data_logger_box import BOX_ID, VERSION, DataLoggerBox
from data_logger_configuration_handler import create_configuration_handlers

def configure_options(options: FlowMakerWorkerOptions) -> None:
    options.router_transport_address = "tcp://*:5560"
    options.runtime_http_address = "http://localhost:3120"

def main() -> None:
    bridge = FlowBoxWorkerBridge()
    bridge.register_handlers(f"{BOX_ID}/{VERSION}", create_configuration_handlers())

    FlowMakerWorkerBuilder() \
        .configure(configure_options) \
        .declare_flowbox(DataLoggerBox) \
        .use_worker_bridge(bridge) \
        .build() \
        .run()

if __name__ == "__main__":
    main()

DataLoggerBox instances are created per flow execution; the WorkerConfigHandlers router is created once for industream/data-logger/<VERSION> and reused for all configuration requests.

Common Pitfalls

Registering the same box version twice

# !!! WRONG !!! : Two routers for the same box version
bridge.register_handlers("myorg/my-box/1.0.0", first_router)
bridge.register_handlers("myorg/my-box/1.0.0", second_router)  # ValueError

# +++ CORRECT +++ : One router per box version, fan out events inside it
router = (
    WorkerConfigHandlers()
    .on("my-box:a", handle_a)
    .on("my-box:b", handle_b)
)
bridge.register_handlers("myorg/my-box/1.0.0", router)

[ERROR!error/DUPLICATE REGISTRATION] register_handlers raises ValueError if a router is already registered for the box version (and WorkerConfigHandlers.on raises ValueError if the same event is mapped twice). One router per box version.

Blocking the event loop inside a handler

# !!! WRONG !!! : Blocking call inside the handler
def _load(payload):
    return requests.get(url).text  # blocks the whole worker, not just this request

# +++ CORRECT +++ : Await asynchronously
async def _load(payload):
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=5)
        return response.text

[WARNING!warning/EVENT LOOP BLOCKING] Requests are dispatched on the worker's asyncio event loop (asyncio.wait_for with the configured timeout). A synchronous blocking call inside a handler stalls the event loop—the ZMQ message loop, other in-flight requests, and the Socket.IO client all freeze until it returns. Use async handlers and async I/O clients.

Forgetting `use_worker_bridge`

# !!! WRONG !!! : Bridge built but never attached — it never connects
bridge = FlowBoxWorkerBridge()
bridge.register_handlers("myorg/my-box/1.0.0", router)
FlowMakerWorkerBuilder().declare_flowbox(MyBox).build().run()

# +++ CORRECT +++ : Attach the bridge to the builder
FlowMakerWorkerBuilder() \
    .declare_flowbox(MyBox) \
    .use_worker_bridge(bridge) \
    .build() \
    .run()

[ERROR!error/BRIDGE NEVER CONNECTS] Without use_worker_bridge(bridge), the worker never calls connect, so it never joins the /worker/<workerId> namespace and every request from the UI fails.

Assuming a typed payload

# !!! WRONG !!! : Trusting the payload shape
def _double(payload):
    return payload["value"] * 2  # KeyError / TypeError -> HANDLER_ERROR with a cryptic message

# +++ CORRECT +++ : Validate and raise a typed error
def _double(payload):
    if not isinstance(payload, dict) or "value" not in payload:
        raise WorkerBridgeException("Expected { value: <number> }", WorkerBridgeErrorCodes.INVALID_PAYLOAD)
    return payload["value"] * 2

[NOTE!warning/RAW PAYLOADS] The Python bridge hands handlers the raw JSON payload and never deserializes it into a typed model, so INVALID_PAYLOAD is never raised for you. Validate the shape yourself and raise WorkerBridgeException with a message the configuration UI can show.

Manually implementing `:health`

# !!! WRONG !!! : Re-implementing the reserved health event
WorkerConfigHandlers().on(":health", lambda payload: {"status": "ok"})

# +++ CORRECT +++ : Leave it alone — the bridge answers :health automatically
# (register only your application events)

[NOTE!sync/RESERVED EVENT] :health is answered by the bridge before your router runs. Registering it in a WorkerConfigHandlers router has no effect and only adds dead code.

Reference

Worker Messages from the Configuration UI (Svelte side)
C# worker messages guide
TypeScript worker messages guide
sdk/python/industream/flowmaker/sdk/worker_bridge.py
sdk/python/industream/flowmaker/sdk/flowmaker_worker_builder.py