Framework - Bare Python

Package: z4j-bare - the foundation of every framework adapter.

When to use z4j-bare directly

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

Agent lifecycle

import asyncio
from z4j_bare import Agent

async def main():
    async with Agent(
        brain_url="wss://z4j.example.com/ws",
        token=os.environ["Z4J_TOKEN"],
        project_id="default",
        agent_name="worker-01",
    ):
        await your_worker_loop()

asyncio.run(main())

async with will:

1. Open the WebSocket, send hello, wait for ack.
2. Install patches on auto-detected engines (scans installed z4j-* packages).
3. Start the heartbeat task.
4. On exit, send goodbye, await ack, close.

Sync workers

If your worker is sync, run the agent in a dedicated thread:

import threading, asyncio
from z4j_bare import Agent

def run_agent():
    asyncio.run(Agent(brain_url=..., token=...).run_forever())

threading.Thread(target=run_agent, daemon=True).start()
your_sync_worker_loop()

Agent.run_forever() is a convenience for this case. It never returns; send SIGTERM to stop.

Explicit engine registration

If auto-discovery misbehaves (multiple Celery apps in one process, unusual import patterns):

from z4j_bare import Agent
from z4j_celery import CeleryAdapter

agent = Agent(...)
agent.register_engine(CeleryAdapter(app=my_primary_celery_app))
agent.register_engine(CeleryAdapter(app=my_analytics_celery_app))

Each registered adapter shows separately in the brain with its own queue namespace.

Dispatcher internals

The dispatcher (z4j_bare.dispatcher.Dispatcher) batches events, applies redaction, and manages the WS flush loop. You rarely touch it directly - but it’s public, so custom adapters can reuse it.

Test fixtures

z4j-bare ships pytest fixtures that stub the brain:

from z4j_bare.testing import stub_brain

def test_enqueue_emits_event(stub_brain):
    your_app.enqueue_something()
    assert stub_brain.events[0].event_type == "task_sent"

Minimum Python

3.10. Uses match statements and ParamSpec.

What’s in the box

• Agent - main entry point.
• Dispatcher - batching + redaction + flush.
• Redactor - pluggable secret filter.
• HelloBuilder - constructs the capability map.
• testing.stub_brain - test fixture.

See API § websocket-protocol for the wire-level reference.