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)
    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:

Opens the WebSocket, sends hello with engine and scheduler capabilities, waits for hello_ack.
Installs patches on each supplied engine adapter to capture lifecycle events.
Starts the heartbeat loop.
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.

Not every argument is environment-backed. engines, schedulers, and framework are live Python objects and are code-only. tags, 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. 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 match statements, StrEnum, and current asyncio idioms.

Wire-level reference

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