Quickstart - Bare Python

z4j-bare is the foundation package; every framework adapter depends on it. You can use it directly if you have no framework.

1. Install

pip install z4j-bare z4j-celery

2. Mint an agent

From the z4j dashboard, Agents, Mint. The response shows the bearer token and the per-project HMAC secret once. Save both: Z4J_TOKEN and Z4J_HMAC_SECRET. The agent refuses to start without the HMAC secret.

3. Start the agent at process boot

worker.py

import asyncio
from z4j_bare import install_agent
from z4j_celery import CeleryEngineAdapter

# my_celery_app = Celery("myapp", broker="redis://...")

runtime = install_agent(
    engines=[CeleryEngineAdapter(celery_app=my_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:
    asyncio.run(your_work())
finally:
    runtime.stop()  # graceful close on shutdown

install_agent returns a started AgentRuntime. Hold the reference and call .stop() on shutdown. The runtime drives the WebSocket connection on a background thread and an internal asyncio loop, so it composes with sync workloads.

Verify before launching the long-running agent

Before you run worker.py for the first time, run the doctor; it confirms config, brain reachability, and buffer-path writability without starting the persistent agent:

Z4J_BRAIN_URL=... Z4J_TOKEN=... Z4J_HMAC_SECRET=... Z4J_PROJECT_ID=... \
  python -m z4j_bare doctor

Run as the same user your worker process will run under. All probes should be [OK].

4. What install_agent does

• Opens the WebSocket to z4j (or falls back to longpoll if WebSocket is blocked).
• Sends hello with engine and scheduler capabilities.
• Installs patches on the supplied engine adapters to capture lifecycle events.
• Starts a heartbeat loop.
• On runtime.stop(), drains in-flight events, sends close, and joins the background thread.

5. Multiple engines

install_agent accepts any number of engine adapters; the runtime fans events out for each:

from z4j_bare import install_agent
from z4j_celery import CeleryEngineAdapter
from z4j_rq import RqEngineAdapter

runtime = install_agent(
    engines=[
        CeleryEngineAdapter(celery_app=celery_app),
        RqEngineAdapter(redis_url="redis://localhost:6379/0"),
    ],
)

Notes

• All install_agent keyword arguments map to a matching Z4J_* env var. Explicit keywords win over env vars, which win over defaults. See install_agent reference for the full argument list.
• The runtime is safe to start and stop multiple times in tests.