Flask

z4j-flaskv1.10.0Apache-2.0released 2026-08-28

PyPISourceProject pageUpstream

Requires Flask 3.1.3+, Python 3.11+. See the compatibility matrix for the full pin string.

Install + wire

Z4J() and Z4J.init_app(app) read configuration from Flask's app.config or from the environment; they do not take a config dict argument.

Factory pattern with deferred init:

from flask import Flask
from z4j_flask import Z4J

z4j = Z4J()

def create_app():
    app = Flask(__name__)
    app.config.from_object("myapp.config.Config")
    # Required: Z4J_BRAIN_URL, Z4J_TOKEN, Z4J_HMAC_SECRET,
    # Z4J_PROJECT_ID. Optional: Z4J_AGENT_NAME and Z4J_AGENT_ID.
    z4j.init_app(app)
    return app

Or the one-shot form, with config already on app.config or in env:

Z4J(app)

Configuration sources

The extension accepts three configuration shapes. Environment variables have highest precedence, then flat Flask keys, then the nested Flask dict.

Source	Example
Flat Flask keys	app.config["Z4J_BRAIN_URL"] = "...", app.config["Z4J_TOKEN"] = "...", app.config["Z4J_HMAC_SECRET"] = "..."
Nested Flask dict	app.config["Z4J"] = {"brain_url": "...", "token": "...", "hmac_secret": "...", "project_id": "billing-prod"}
Environment variables	Z4J_BRAIN_URL, Z4J_TOKEN, Z4J_HMAC_SECRET, Z4J_PROJECT_ID, Z4J_AGENT_NAME

Z4J_HMAC_SECRET is required; the agent refuses to start without it. To skip the agent entirely (CI, local dev), set Z4J_DISABLED=1 (env) or app.config["Z4J_DISABLED"] = True.

Lifecycle

Flask's application context does not have a natural startup hook. z4j-flask starts the agent during init_app itself; the WebSocket connection lives in a background asyncio runtime that the extension manages. Request teardown hooks are registered automatically.

When TASKIQ_BROKER is configured, Flask discovery attaches z4j's TaskIQ middleware but does not guess a loop during synchronous app initialization. Initialize Z4J(app) before the component that starts TaskIQ; the broker's real startup hook then binds its owner loop. Until it is bound, TaskIQ submit and result probes from z4j fail closed. A separate TaskIQ worker needs its own agent.

For long-running workers that don't run under Flask's request loop (RQ worker, Celery beat scheduler):

rq_worker.py

from myapp import create_app
app = create_app()   # Z4J.init_app already ran inside; the agent is up.
# ... RQ worker loop ...

Multi-process

Each gunicorn / uWSGI worker process registers separately. z4j's worker-first protocol identifies each one by (agent_id, worker_id), so a gunicorn -w 4 deployment shows up as four workers under one agent in the dashboard.

Z4J_AGENT_NAME is host metadata on the shared agent row, not a per-worker identity. Keep it stable for all workers that share one token. If hosts need separate labels, mint a separate agent token for each host instead of varying the name by PID.

Blueprints

z4j does not register any Flask blueprints. It attaches as pure background behavior; your URL space is untouched.

CLI (flask ...)

Management commands run under the same init_app path, so the agent attempts to start when they boot. Set Z4J_DISABLED=1 for one-off commands you do not want surfaced as agent connects.

Verify with doctor

python -m z4j_flask doctor checks the buffer directory and brain DNS / TCP / TLS path, then starts a temporary WebSocket runtime using the Z4J_* env vars your service is configured with. It does not import the Flask app or read app.config.

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

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

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

Exit 0 on all-green, 1 on any failure. It catches service-user startup failures and DNS, firewall, or certificate issues. The WebSocket probe returns when the background runtime starts and does not wait for authenticated hello_ack, so a bad token, project value, or HMAC secret can still pass. Confirm authentication in the agent and brain logs. See service-user deployments.

Troubleshooting

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

• Gunicorn / uWSGI preload mode -- --preload imports your app (and installs the agent) ONCE in the arbiter, then fork()s the workers. Threads do not survive a fork, so without extra wiring the workers inherit a dead agent and capture nothing. Add a post-fork hook so each worker re-establishes its own agent:

  gunicorn.conf.py

  def post_fork(server, worker):
      from z4j_bare import post_fork as z4j_post_fork
      z4j_post_fork()

  For uWSGI use the @postfork decorator to call z4j_bare.post_fork(). If you do NOT preload (the gunicorn default), the app is imported in each worker after the fork and no hook is needed.

• PermissionError: ... /var/www/.z4j in service log -- the agent auto-relocates the buffer to $TMPDIR/z4j-{uid}. See service-user deployments.