intrabus 0.2.2

General-purpose ZeroMQ communication node with pub/sub, request/reply, registry, heartbeats, stats, and diagnostics

intrabus

intrabus is a lightweight, general-purpose ZeroMQ communication node for Python applications.

It gives independent modules a simple way to communicate over:

• publish/subscribe events
• request/reply messages
• module registration
• heartbeats and liveness tracking
• runtime statistics
• diagnostics
• node health queries

The goal is to keep setup simple: install the library, start a CommunicationNode, connect modules with BusInterface, and communicate.

pip install intrabus

Current release target: 0.2.2

intrabus is still alpha software. The public API is usable, but the project may still evolve before 1.0.

---

Why intrabus?

Use intrabus when you want multiple Python components to communicate inside one machine or between local processes without deploying external infrastructure such as Redis, RabbitMQ, Kafka, or HTTP services.

Typical use cases:

• test automation systems
• modular desktop applications
• robotics or hardware-control tools
• local monitoring/control systems
• plugin-like Python applications
• prototypes that may later grow into distributed systems

intrabus is not intended to replace large production message brokers. It is designed to be small, embeddable, and easy to reason about.

---

Quick start

from intrabus import BusInterface, CommunicationNode


def handle_request(message: dict) -> dict:
    return {
        "pong": True,
        "received": message,
    }


with CommunicationNode("local"):
    server = BusInterface("server", request_handler=handle_request)
    client = BusInterface("client")

    reply = client.send_request("server", {"ping": True}, timeout=2)
    print(reply)

    health = client.send_request(
        "intrabus.node",
        {"command": "node.get_health"},
        timeout=2,
    )
    print(health["status"])

    server.stop()
    client.stop()

---

Core concepts

CommunicationNode

CommunicationNode owns the core intrabus runtime:

• TopicBroker for pub/sub traffic
• CentralBroker for request/reply traffic
• ModuleRegistry for known modules
• StatsCollector for runtime metrics
• DiagnosticsManager for runtime diagnostics

Most users should start here:

from intrabus import CommunicationNode

with CommunicationNode("local") as node:
    print(node.is_running)

BusInterface

BusInterface is what application modules use to communicate.

from intrabus import BusInterface

module = BusInterface("sensor")

A BusInterface can:

• publish events
• subscribe to events
• send requests
• handle incoming requests
• auto-register with the local node
• send heartbeats to the local node

Module registry

When a module starts, it can automatically register with the node:

worker = BusInterface("worker")

The node tracks:

• module name
• node ID
• status: online or offline
• connected time
• last seen time
• capabilities
• metadata

Heartbeats and stale detection

By default, modules send periodic heartbeats to the node.

The node can mark modules offline when they stop sending heartbeats:

reply = client.send_request(
    "intrabus.node",
    {
        "command": "node.get_health",
        "timeoutSeconds": 5,
    },
)

---

Publish/subscribe

from intrabus import BusInterface, CommunicationNode


with CommunicationNode():
    publisher = BusInterface("publisher")
    subscriber = BusInterface("subscriber")

    def on_temperature(topic: str, message: dict) -> None:
        print(topic, message)

    subscriber.subscribe("temperature", on_temperature)
    publisher.publish("temperature", {"value": 23.5})

    publisher.stop()
    subscriber.stop()

---

Request/reply

from intrabus import BusInterface, CommunicationNode


def echo(message: dict) -> dict:
    return {"echo": message}


with CommunicationNode():
    server = BusInterface("server", request_handler=echo)
    client = BusInterface("client")

    reply = client.send_request("server", {"hello": "world"}, timeout=2)
    print(reply)

    server.stop()
    client.stop()

BusInterface reserves a few request/reply payload fields for protocol metadata:

• correlationId
• sender
• __intrabus_type

The __intrabus_type marker is added automatically as "request" or "reply". It prevents late replies from being treated as new application requests after node restarts or transient disconnects.

---

Node management commands

The internal node service is available at:

intrabus.node

Send requests to it with BusInterface.send_request().

node.get_registry

Returns the current module registry.

client.send_request("intrabus.node", {"command": "node.get_registry"})

node.get_stats

Returns current runtime statistics.

client.send_request("intrabus.node", {"command": "node.get_stats"})

node.get_diagnostics

Returns current diagnostics derived from registry and stats.

client.send_request("intrabus.node", {"command": "node.get_diagnostics"})

node.get_health

Returns a compact node health payload suitable for monitoring:

client.send_request(
    "intrabus.node",
    {
        "command": "node.get_health",
        "timeoutSeconds": 5,
    },
)

The health response contains:

{
    "ok": True,
    "nodeId": "local",
    "status": "healthy",  # healthy | degraded | unhealthy
    "staleModules": [],
    "summary": {
        "modules": {
            "total": 2,
            "online": 2,
            "offline": 0,
        },
        "stats": {
            "totalMessages": 10,
            "totalRequests": 5,
            "totalReplies": 4,
            "totalTimeouts": 0,
            "totalErrors": 0,
            "totalEvents": 1,
            "totalDeliveryFailures": 0,
        },
        "diagnostics": {
            "count": 0,
            "severityCounts": {},
        },
    },
    "diagnostics": {...},
}

To include the full registry and raw stats payload, including recent events and recent messages, pass includeDetails:

client.send_request(
    "intrabus.node",
    {
        "command": "node.get_health",
        "includeDetails": True,
    },
)

node.reset_stats

Clears runtime statistics.

client.send_request("intrabus.node", {"command": "node.reset_stats"})

node.clear_diagnostics

Clears currently stored diagnostics.

client.send_request("intrabus.node", {"command": "node.clear_diagnostics"})

---

Backward-compatible node commands

The following older command names are still supported:

Older command	Preferred command
registry.get	node.get_registry
stats.get	node.get_stats
diagnostics.get	node.get_diagnostics
stats.reset	node.reset_stats
diagnostics.clear	node.clear_diagnostics

Internal module lifecycle commands are also supported:

• module.register
• module.unregister
• module.heartbeat
• registry.health

Most application code should use the node.* commands.

---

Runtime statistics

StatsCollector tracks bounded in-memory application traffic statistics:

• total messages
• requests
• replies
• events
• timeouts
• errors
• delivery failures
• latency samples
• average latency
• max latency
• per-module stats
• per-topic stats
• recent events
• recent message metadata

User-facing traffic stats intentionally exclude internal node-management traffic, including module registration, module heartbeat, registry polling, health checks, diagnostics polling, and stats polling. This keeps monitoring clients and web dashboards from inflating application message totals.

Recent messages are bounded and do not include full payloads by default. This avoids unbounded RAM growth and reduces the risk of accidentally storing sensitive data.

---

Diagnostics

DiagnosticsManager currently derives diagnostics from registry and stats. Repeated recent stats events are grouped into higher-level diagnostics to keep monitoring output readable.

Supported diagnostic codes:

Code	Meaning
MODULE_OFFLINE	A registered module is offline
REQUEST_TIMEOUT	A request timed out
HANDLER_ERROR	A request handler raised an exception
DELIVERY_FAILURE	A message could not be delivered by the broker

Diagnostic severities:

• info
• warning
• error
• critical

node.get_health maps diagnostics to node status:

Status	Meaning
healthy	No warnings or errors
degraded	At least one warning
unhealthy	At least one error or critical diagnostic

---

MessageEnvelope

MessageEnvelope is a dataclass for future-compatible standardized messages.

It includes fields for future multi-node support:

• messageId
• correlationId
• sender
• target
• topic
• sourceNode
• targetNode
• ttl
• hopPath
• payload

The current brokers still use lightweight JSON payloads internally, but MessageEnvelope is available as the standard message shape for higher-level integrations.

---

Advanced: manual brokers

Most users should use CommunicationNode.

For low-level control, you can still start brokers manually:

from intrabus import run_central_broker, run_topic_broker

run_topic_broker()
run_central_broker()

And stop singleton brokers:

from intrabus import stop_central_broker, stop_topic_broker

stop_topic_broker()
stop_central_broker()

This is mainly useful for simple scripts and compatibility with older examples.

---

Development

Install development dependencies:

uv sync --dev

Run tests:

uv run pytest -q

Run linting and formatting checks:

uv run ruff check .
uv run ruff format --check .

Format code:

uv run ruff format .

Build the package:

uv run python -m build

Check distributions:

uv run twine check dist/*

---

Roadmap

Planned future work:

• event/message sinks for streaming to external consumers
• demo project using this package
• monitoring dashboard as a separate project
• richer diagnostics
• multi-node routing
• optional dashboard/websocket integrations

The library itself will stay focused on being a general-purpose communication node.

---

License

MIT