Framework - FastAPI

Package: z4j-fastapi - FastAPI 0.105+.

Install + wire

from contextlib import asynccontextmanager
from fastapi import FastAPI
from z4j_fastapi import Z4JAgent

@asynccontextmanager
async def lifespan(app: FastAPI):
    agent = Z4JAgent(
        brain_url=os.environ["Z4J_BRAIN_URL"],
        token=os.environ["Z4J_TOKEN"],
        project_id=os.environ.get("Z4J_PROJECT_ID", "default"),
        agent_name=os.environ.get("HOSTNAME", "fastapi"),
    )
    async with agent:
        app.state.z4j = agent
        yield

app = FastAPI(lifespan=lifespan)

Why lifespan not on_event

FastAPI’s @app.on_event("startup") is deprecated. The lifespan context manager is the current idiom; it cleanly runs the agent for the full server lifetime.

Worker processes (arq / taskiq)

arq and taskiq workers are separate processes from the FastAPI app. They also need an agent. Pattern:

# arq worker
from arq import ArqRedis
from z4j_fastapi import Z4JAgent

async def on_startup(ctx):
    ctx["z4j"] = Z4JAgent(...)
    await ctx["z4j"].__aenter__()

async def on_shutdown(ctx):
    await ctx["z4j"].__aexit__(None, None, None)

class WorkerSettings:
    functions = [...]
    on_startup = on_startup
    on_shutdown = on_shutdown

Same shape for taskiq (startup / shutdown hooks).

Uvicorn reload

During uvicorn --reload, agents reconnect on every save. The brain tolerates this - rapid reconnects don’t create duplicate agents because tokens are unique.

Multi-worker (uvicorn --workers N)

Each worker is a separate agent process. Set agent_name with $PID if you want per-worker visibility.

Auth integration

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

Config keys

Same as Django/Flask - see Django.