Quickstart - FastAPI

1. Install

pip install z4j-fastapi z4j-arq           # or z4j-taskiq, 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. Attach to the lifespan (recommended)

main.py

from fastapi import FastAPI
from z4j_fastapi import z4j_lifespan

# Reads Z4J_BRAIN_URL / Z4J_TOKEN / Z4J_HMAC_SECRET / Z4J_PROJECT_ID
# from the environment. Pass them as kwargs to z4j_lifespan(...) only
# if you want explicit overrides.
app = FastAPI(lifespan=z4j_lifespan())

z4j_lifespan() returns an asynccontextmanager you hand to FastAPI(lifespan=...). The agent starts when FastAPI starts up and stops cleanly when it shuts down.

If the FastAPI process itself needs an arq adapter for discovery and actions, pass either arq's RedisSettings or a pre-built arq Redis pool:

from arq.connections import RedisSettings

app = FastAPI(lifespan=z4j_lifespan(
    arq_redis_settings=RedisSettings(host="redis", port=6379),
))

arq_redis_settings does not accept an SQLAlchemy session factory. Other engine handles have separate keyword arguments such as rq_app, dramatiq_broker, huey, and taskiq_broker. Passing a handle here only constructs the adapter in this FastAPI process; an arq worker is a separate process and still needs its own runtime and hook wiring.

4. Manual install (apps that cannot use lifespan)

from z4j_fastapi import install_z4j
from fastapi import FastAPI

app = FastAPI()

# `app` is required. The helper registers both a FastAPI shutdown handler
# and an atexit fallback, so a second manual stop hook is normally unnecessary.
runtime = install_z4j(app)

# Startup failures and Z4J_DISABLED return None while FastAPI keeps running.
if runtime is None:
    # Inspect `z4j.host.fastapi.extension` logs.
    ...

5. For arq / taskiq worker processes

The worker processes also need their own agent runtime. The adapter instance whose event queue the runtime drains must be the same instance supplied to the worker hook or middleware; creating a second throwaway adapter silently sends events to an undrained queue.

• arq: ArqEngineAdapter requires redis_settings. The attach_to_worker_settings helper also needs an existing capture, or that same adapter plus a live event loop. Its class form must run before arq builds the Worker; its instance form can run from on_startup. Do not copy the old ArqEngineAdapter() one-liner: it fails immediately, and constructing a separate adapter would not feed the installed runtime.
• taskiq: construct one TaskiqEngineAdapter(broker=broker), pass it to the worker's install_agent(engines=[adapter]), then call attach_to_broker(broker, adapter=adapter). The one-argument attach_to_broker(broker) form is rejected.
• Celery: import z4j_celery in the worker process. That registers a worker_ready handler (not worker_init). Auto-bootstrap also requires the four Z4J_* credentials in the worker environment and a Celery argv shape the detector recognizes; the exact celery -A app worker form works.

See arq, taskiq, or Celery for the exact pattern.

6. Verify

Run the doctor first; it checks local configuration, buffer access, and the DNS/TCP/TLS/transport path:

python -m z4j_fastapi doctor

Run as the same user the service runs under. Its WebSocket probe returns after the background runtime starts and does not wait for agent authentication, so all [OK] rows can still accompany a bad token. Confirm the worker slot online and inspect auth warnings as well.

Then boot your API and worker and confirm their worker slots online. Enqueue and run a task, then confirm its lifecycle in Tasks. arq emits its first z4j event when a worker starts the job, not when enqueue_job queues it, so a deferred or waiting arq job has no task row yet and there is no 100 ms enqueue guarantee.

Troubleshooting

See framework: FastAPI.