python_zmq_framework

Python node library for a flow-based, language-agnostic runtime — Node-RED without the UI. The reference runtime lives in ruby_zmq_framework; this library lets a Python process join that flow as a full citizen, or run entirely standalone with no Ruby involved.

• Nodes are independent OS processes. Each one does one job, lives in one file, and knows nothing about any other node — not their ports, not their names, not their language.
• Wires are pub/sub topics carrying JSON, over ZeroMQ.
• The graph is data: flow.yml (in the Ruby repo) is the only artifact that knows the topology. bin/flowctl reads it, computes the wiring, and runs every node — Python or Ruby — with that wiring in its environment.
• The contract is one page: PROTOCOL.md is everything a node in any language needs to join.

Only dependency is pyzmq. Nothing here requires Ruby to install or run — boot() also works standalone with no environment set, for local iteration or running a single node on its own.

Install

pip install -e .
# or, without installing the package:
pip install -r requirements.txt

Writing a node

from python_zmq_framework import FrameworkNode, boot

class RpmSmoother(FrameworkNode):
    def __init__(self, bus):
        super().__init__(bus)
        self.window = []

    def handle_message(self, topic, payload):
        self.window = (self.window + [payload["rpm"]])[-5:]
        self.broadcast("engine_data_smooth", {"rpm": sum(self.window) / len(self.window)})

RpmSmoother_instance = boot(RpmSmoother)

Note what's absent: no ports, no peers, no .subscribe() calls. Wiring comes entirely from environment variables — BUS_PORT, BUS_PEERS, BUS_SUBSCRIBES, NODE_NAME — set by whatever launches the process (see PROTOCOL.md). A flow manifest entry in the Ruby repo's flow.yml computes and injects them:

  rpm_smoother:
    cmd: python3 ../python_zmq_framework/nodes/rpm_smoother.py
    subscribes: [engine_data]
    publishes: [engine_data_smooth]

Run standalone (no environment needed — it binds an ephemeral port) to poke at a node in isolation: python3 nodes/dbc_decoder.py.

Every node automatically heartbeats every 5 seconds. FrameworkNode.handle_message is @abstractmethod, so Python's ABC machinery raises TypeError the moment you try to instantiate a subclass that forgot it — before __init__ runs at all, so no heartbeat thread can leak from a failed construction. That's exactly the fast, specific feedback an iterating LLM agent needs.

How It Works

1. ZeroMQBus binds a PUB socket (pass 0 for an OS-assigned ephemeral port, readable back via .port) and connects a SUB socket to every peer. Peers may be ints (loopback ports), "host:port" strings, or full ZeroMQ endpoints; pass bind_host="0.0.0.0" to accept peers from other machines. There is no dynamic discovery — every node's address has to be listed in every peer's peer_ports up front (or computed for you by bin/flowctl). Messages published on a bus are also delivered synchronously to subscribers on that same bus.
2. FrameworkNode is the base class for a node on the bus. Subclass it and implement handle_message(topic, payload). You get an automatic heartbeat broadcast (node_name, status, timestamp) every 5 seconds for free. node_name defaults to the class name; set self.node_name = "..." before calling super().__init__() to give distinct instances distinct identities.
3. boot(NodeClass) builds a node entirely from the environment contract in PROTOCOL.md — the same helper the Ruby gem exposes as RubyZmqFramework.boot.
4. Resilience: the listener survives anything the network throws at it — non-framework frame layouts and malformed JSON are dropped with a warning, and each subscriber's handle_message is caught individually so one raising handler can't starve the others or kill the listener. Handlers on one bus never run concurrently.
5. Clean shutdown: bus.close() stops the listener and releases the sockets; node.stop_heartbeat() ends the heartbeat thread. Stop your node first, then close the bus — publishing on a closed bus raises ZeroMQBusError.
6. Thread safety: a ZeroMQ PUB socket must not be written to concurrently from multiple threads. The heartbeat thread and whatever thread calls broadcast()/publish() both use the same socket, so ZeroMQBus.publish takes a lock around every send.

Demo nodes

pip install -r requirements.txt
python3 nodes/dbc_decoder.py        # broadcasts engine_data
python3 nodes/telemetry_logger.py   # logs whatever BUS_SUBSCRIBES names

nodes/dbc_decoder.py is a stand-in for a real cantools-based DBC decoder — a pure producer. nodes/telemetry_logger.py is a consumer: it calls no .subscribe() itself, so wiring it to engine_data is purely an environment-variable exercise:

BUS_PORT=5561 NODE_NAME=dbc_decoder python3 nodes/dbc_decoder.py &
BUS_PORT=5562 BUS_PEERS=127.0.0.1:5561 BUS_SUBSCRIBES=engine_data,heartbeat \
  NODE_NAME=logger python3 nodes/telemetry_logger.py

Tests

python -m unittest discover -s tests

Interop with the Ruby repo (optional)

The wire protocol and environment contract are shared — see PROTOCOL.md — so a Python node and a Ruby node interoperate automatically whenever both are wired into the same flow.yml (in ruby_zmq_framework), or by hand:

bus = ZeroMQBus(5561, peer_ports=[5558])  # 5558 = a Ruby StateRegistry node

This is entirely optional — nothing here requires the Ruby gem to be installed or running, and boot() works with only Python processes present.