FastAPI

Requires FastAPI 0.95+, Python 3.10+. See the compatibility matrix for the full pin string.

Public surface

from z4j_fastapi import (
    z4j_lifespan,             # function: returns an asynccontextmanager
    install_z4j,              # function: manual install for non-lifespan apps
    get_runtime,              # function: retrieve the live AgentRuntime
    FastAPIFrameworkAdapter,  # class: the framework adapter
    reconcile_schedules,      # function: declarative schedule reconciliation
)

Install + wire (lifespan, recommended)

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. Z4J_HMAC_SECRET is required; the runtime refuses to start without it.

To compose with your own startup logic, build a wrapper lifespan that yields the z4j lifespan inside:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with z4j_lifespan()(app):
        # your own setup here (db pools, caches, ...)
        yield
        # your own teardown

Engine handles via kwargs

z4j_lifespan(...) accepts a number of optional engine handles so the framework adapter can register them without import-order gymnastics. The most common:

Kwarg	Purpose
`arq_redis_settings`	`arq.RedisSettings` or pre-built pool used by the arq adapter.
`arq_function_names`, `arq_queue_name`	arq function discovery hints.
`rq_app`	Pre-built RQ Redis connection.
`taskiq_brokers`	Iterable of taskiq broker instances to attach.

Each handle is forwarded to the matching engine adapter only if that adapter is installed. If you pass arq_redis_settings without z4j-arq installed, the lifespan errors with a clear message.

Manual install (apps that cannot use lifespan)

from z4j_fastapi import install_z4j, get_runtime

# Call once at startup; the returned handle drives the agent loop.
runtime = install_z4j()

# Anywhere later in the process:
runtime = get_runtime()
runtime.stop()  # graceful close on shutdown

Why lifespan, not `@app.on_event`

@app.on_event("startup") is deprecated. The lifespan context manager is the current idiom; it cleanly runs the agent for the full server lifetime, including graceful shutdown.

Worker processes (arq / taskiq / Celery)

arq, taskiq, and Celery workers are separate processes from the FastAPI app and need their own agent.

Celery: importing z4j_celery is enough; the worker_init signal auto-boots the agent inside Celery workers. z4j-fastapi already imports z4j_celery opportunistically when it’s installed, so a typical FastAPI + Celery app needs no extra wiring on the worker side.
arq: use attach_to_worker_settings(WorkerSettings, adapter=ArqEngineAdapter()) at module import time. See arq engine.
taskiq: use attach_to_broker(broker). See taskiq engine.

Uvicorn reload

During uvicorn --reload, agents reconnect on every save. z4j tolerates this — rapid reconnects do not create duplicate agents because the worker-first protocol keys connections by (agent_id, worker_id).

Multi-worker (`uvicorn --workers N`)

Each worker is a separate process. The worker-first protocol identifies each one by (agent_id, worker_id) so a uvicorn -w 4 deployment shows as four workers under one agent in the dashboard. Set Z4J_AGENT_NAME to keep multi-host deploys readable.

Auth

z4j is not an authentication provider for your FastAPI app. The agent only authenticates to the brain. Your app’s own auth surface is untouched.

Verify with doctor

python -m z4j_fastapi doctor runs the same probes the agent runtime does (buffer dir writable, brain DNS / TCP / TLS, WebSocket upgrade) using the Z4J_* env vars your service is configured with.

# Always run as the same user the service runs under.
sudo -u app /srv/app/venv/bin/python -m z4j_fastapi doctor

# Skip the WS round-trip when z4j is intentionally offline:
python -m z4j_fastapi doctor --no-websocket

# Machine-readable for scripting:
python -m z4j_fastapi doctor --json

Exit 0 on all-green, 1 on any failure. Catches uvicorn-under-service-user silent startup failures, NAT / firewall / cert issues, and wrong-token / wrong-project problems with a specific failure reason. See service-user deployments.

Troubleshooting

First, run python -m z4j_fastapi doctor — it surfaces the most common failures with a specific reason.

PermissionError: ... /nonexistent/.z4j under uvicorn — the service user has an unwritable $HOME; the agent auto-relocates the buffer to $TMPDIR/z4j-{uid}. See service-user deployments.
Agent only registers under uvicorn, not under arq / taskiq workers — workers are separate processes and need the matching engine-wiring helper (see Worker processes above).

Configuration sources

z4j_lifespan(...) and install_z4j(...) accept the same brain_url, token, hmac_secret, project_id, agent_name keyword arguments; explicit kwargs win over Z4J_* env vars. See env vars for the full list.