FastAPI

Requires FastAPI 0.109.1+, Python 3.11+. 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_broker`	One taskiq broker instance; z4j attaches its middleware and TaskIQ startup binds the actual owner loop.

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.

For TaskIQ, an ambient ASGI loop is not treated as proof of broker ownership. Pass the broker before its host lifecycle starts; the adapter remains unbound and its asynchronous actions fail closed until TaskIQ startup records the real owner. If the broker is already live on another loop, use the direct TaskIQ adapter API with that verified broker_loop instead.

Manual install (apps that cannot use lifespan)

from fastapi import FastAPI
from z4j_fastapi import install_z4j, get_runtime

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

# Anywhere later in the process:
runtime = get_runtime()
# install_z4j registers FastAPI shutdown and process-exit handlers.

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 registers a worker_ready handler that boots the agent after Celery has forked its pool. It is deliberately not registered on worker_init. z4j-fastapi imports z4j_celery opportunistically when installed, provided the Celery worker imports the FastAPI application module.
arq: construct ArqEngineAdapter(redis_settings=RedisSettings(...)), then call attach_to_worker_settings(WorkerSettings, adapter=adapter) at module import time. See arq engine.
taskiq: construct adapter = TaskiqEngineAdapter(broker=broker), then call attach_to_broker(broker, adapter=adapter) before the TaskIQ CLI starts the 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 checks the buffer directory and the brain's DNS / TCP / TLS path, then starts a temporary WebSocket runtime 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. It catches uvicorn-under-service-user startup failures and DNS, firewall, or certificate issues. The WebSocket probe returns when the background runtime starts; it does not wait for the brain's authenticated hello_ack, so a bad token, project value, or HMAC secret can still pass this command. Confirm authentication in the agent and brain logs. 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(...) share the core brain_url, token, hmac_secret, project_id, and engine-handle keyword arguments; explicit kwargs win over Z4J_* env vars. z4j_lifespan(...) additionally owns lifespan composition and startup reconciliation options. See env vars for the full list.