Bare Python

Requires Python 3.11+. See the compatibility matrix for engine and scheduler pin strings.

When to use `z4j-bare` directly

Standalone workers with no web framework (cron scripts, pure RQ workers, Dramatiq workers).
Custom frameworks z4j does not provide an adapter for.
Library or SDK code that wants to ship its own agent.

Public surface

from z4j_bare import (
    install_agent,           # function: install + start the runtime
    AgentRuntime,            # class: returned by install_agent
    RuntimeState,            # enum
    BareFrameworkAdapter,    # framework adapter for no-framework apps
    BufferStore,             # local SQLite buffer (advanced inspection)
    BufferEntry,             # one buffered row, as BufferStore yields it
    post_fork,               # re-establish the agent in a forked worker
    safe_boundary, safe_call,  # safety helpers for custom adapters
)

Agent lifecycle

install_agent(...) returns a started AgentRuntime driving the WebSocket on a background thread. Hold the reference and call runtime.stop() on shutdown.

from z4j_bare import install_agent
from z4j_celery import CeleryEngineAdapter

runtime = install_agent(
    engines=[CeleryEngineAdapter(celery_app=celery_app)],
    # brain_url / token / hmac_secret / project_id default to the
    # matching Z4J_* env vars; pass them explicitly only if you do
    # not want the env-var lookup.
)

try:
    your_worker_loop()       # sync or async; the runtime is independent
finally:
    runtime.stop()

What install_agent does:

Calls connect_signals on each supplied engine and scheduler adapter to capture lifecycle events, and starts the tasks that drain their event queues into the local buffer.
Opens the WebSocket, sends hello with engine and scheduler capabilities, waits for hello_ack.
Starts the heartbeat loop alongside the send and receive loops.
On runtime.stop(), signals the background loop, closes the transport and local buffer, and waits up to the stop timeout. Unsent rows remain in the durable buffer for a later start; shutdown does not guarantee delivery before process exit.

Multiple engines

install_agent(engines=[...]) accepts any number of adapters. The runtime fans events out for each:

from z4j_celery import CeleryEngineAdapter
from z4j_rq import RqEngineAdapter
from redis import Redis
from rq import Queue

rq_queue = Queue("default", connection=Redis.from_url("redis://localhost:6379/0"))

runtime = install_agent(
    engines=[
        CeleryEngineAdapter(celery_app=primary_celery_app),
        RqEngineAdapter(rq_app=rq_queue),
    ],
)

Each uniquely named adapter appears separately in the dashboard with its own queue namespace. The runtime stores adapters by their name, so supplying two adapters for the same engine name replaces the earlier entry rather than creating two dashboard entries.

Argument reference

The credential and connection fields brain_url, token, hmac_secret, and project_id may come from explicit arguments or their matching Z4J_* variables. Explicit arguments win over environment values, then model defaults apply. install_agent raises RuntimeError when Z4J_RUNTIME_DIR, Z4J_BUFFER_DIR or Z4J_BUFFER_PATH is set; relocate state with Z4J_HOME instead.

Not every argument is environment-backed. engines, schedulers, and framework are live Python objects and are code-only. buffer_path and dev_mode are also explicit code settings; Z4J_DEV_MODE is intentionally ignored so an inherited environment cannot opt a process into plaintext transport. tags may also come from Z4J_TAGS as key=value,key2=value2. The required engines argument may be an empty list for a scheduler-only process, but at least one adapter across engines and schedulers is required. See help(install_agent) and the environment reference for the complete mapping.

Verify with doctor

python -m z4j_bare doctor checks the buffer directory and the brain's DNS / TCP / TLS path, then starts a temporary WebSocket runtime using Z4J_* env vars or CLI flags. This is the canonical doctor implementation; the framework adapters (Django / Flask / FastAPI) wrap the same logic.

# Use env vars (typical):
Z4J_BRAIN_URL=https://tasks.example.com \
  Z4J_TOKEN=... \
  Z4J_HMAC_SECRET=... \
  Z4J_PROJECT_ID=ml-pipeline \
  python -m z4j_bare doctor

# Or flags (useful for ad-hoc checks):
python -m z4j_bare doctor \
  --brain-url https://tasks.example.com \
  --token ... \
  --hmac-secret ... \
  --project-id ml-pipeline

# Skip the WS upgrade probe:
python -m z4j_bare doctor --no-websocket

# JSON for scripting:
python -m z4j_bare doctor --json

Exit 0 on all-green, 1 on any failure. The WebSocket probe returns when the background runtime starts and does not wait for the authenticated hello_ack; use agent and brain logs to confirm token, project, and HMAC authentication. The same probes are exposed programmatically via z4j_bare.diagnostics so custom adapters can build their own doctor commands.

See service-user deployments for the buffer-path failure mode the doctor catches.

Minimum Python

3.11+. The runtime uses StrEnum, datetime.UTC, and asyncio.TaskGroup with except*.

Wire-level reference

See websocket protocol for the actual frame schema. The runtime is the canonical reference implementation of the protocol.