--- title: Code workflows description: Build a durable Python workflow with @workflow, @step and hub from the packaged Watchtower sample, choose the account it runs as, publish and email a report, and enable its schedule on the server that should run it. canonical_url: https://hubzoid.com/docs/guides/code-workflows last_updated: 2026-09-27 --- # Code workflows Build a durable Python workflow with @workflow, @step and hub from the packaged Watchtower sample, choose the account it runs as, publish and email a report, and enable its schedule on the server that should run it. A code workflow is a Python function under `workflows//` that runs a defined sequence of steps on a schedule or on demand. Code decides what happens. The model is called only where judgment helps, through `hub.call_llm` or `hub.call_agent`. Each step's result is saved, so a run that stops resumes after its last completed step. ## When to use this Use a code workflow when the steps must be exact: a threshold decides whether to alert, a loop walks a list of records, or you call your own APIs. For work you can describe in a paragraph, a [markdown task](https://hubzoid.com/docs/guides/markdown-tasks) is simpler. Both run on the same engine, as described in [Workflows and schedules](https://hubzoid.com/docs/concepts/workflows-and-schedules). ## Start from the Watchtower sample Watchtower checks service metrics against thresholds every 15 minutes. When a service stays over a threshold, it asks the model to explain the breach and writes a report that people can ask the Watchtower agent about in chat. The metrics are bundled sample data. The model call and the engine are real, so a run needs the same [model credentials](https://hubzoid.com/docs/concepts/agents-and-models) as chat. ```bash hubzoid init my-watchtower --template watchtower cd my-watchtower hubzoid schedule run . watchtower cat output/watchtower/latest.md ``` The report names the failing service, its peak numbers, the likely cause and what to check first. Run the workflow again and it reports nothing new, because it remembers the breach it already explained. ```text my-watchtower/ AGENTS.md README.md knowledge/ watchtower.md raw_data/ events/ metrics.jsonl samples/ broken.jsonl workflows/ settings.yaml watchtower/ main.py ``` ### The workflow file Detection is plain Python in steps. Only the explanation comes from the model, as a validated Pydantic object. ```python title="workflows/watchtower/main.py" from __future__ import annotations import json from datetime import datetime from pathlib import Path from pydantic import BaseModel, Field from hubzoid import hub, step, workflow HUB_DIR = Path(__file__).resolve().parents[2] DEFAULTS = {"p95_ms": 800, "error_rate": 0.02, "window_minutes": 15, "min_points": 3} class Finding(BaseModel): severity: str = Field(description="warning or critical") summary: str = Field(description="One sentence: what is wrong, with the numbers") likely_cause: str = Field(description="The most likely cause, citing an event if one fits") next_step: str = Field(description="One concrete thing an operator should check first") @step() def load_events() -> list[dict]: """Every event line under raw_data/events/. A malformed line fails the run with its file and line number, so the failure is easy to find and fix.""" events = [] for path in sorted((HUB_DIR / "raw_data" / "events").glob("*.jsonl")): for n, line in enumerate(path.read_text().splitlines(), 1): if not line.strip(): continue try: events.append(json.loads(line)) except json.JSONDecodeError as exc: raise ValueError(f"{path.name} line {n} is not valid JSON: {exc.msg}") from None return events ``` Two more steps complete the pipeline. `find_breaches(events, limits)` returns the services whose last `window_minutes` of samples had at least `min_points` over a threshold, with any deploy events attached. `write_report(window_end, results)` writes `output/watchtower/.md` and copies it to `latest.md`. The workflow itself ties them together: ```python title="workflows/watchtower/main.py" @workflow(schedule="every 15 minutes") def watchtower(): limits = {**DEFAULTS, **(hub.setting("watchtower") or {})} breaches = find_breaches(load_events(), limits) new = [b for b in breaches if not hub.state.get(f"explained:{b['service']}:{b['window_end']}")] if not new: return {"breaches": len(breaches), "new": 0, "report": None} results = [] for b in new: finding = hub.call_llm( "A service crossed its alert thresholds. Explain it for the on-call operator.\n" f"Thresholds: p95 latency {limits['p95_ms']} ms, error rate {limits['error_rate']:.1%}.\n" f"Measurements for the last {limits['window_minutes']} minutes:\n{json.dumps(b, indent=2)}", system="You are a careful site reliability engineer. Use only the data given.", response_model=Finding, ) results.append({"breach": b, "finding": finding.model_dump()}) report = write_report(new[0]["window_end"], results) for b in new: hub.state[f"explained:{b['service']}:{b['window_end']}"] = report return {"breaches": len(breaches), "new": len(new), "report": report} ``` The thresholds live in `workflows/settings.yaml`. Each run reads them again, so a change needs no restart. ```yaml title="workflows/settings.yaml" # Settings for this hub's workflows, read with hub.setting(...). watchtower: p95_ms: 800 # p95 latency threshold, milliseconds error_rate: 0.02 # error-rate threshold, 0.02 = 2% window_minutes: 15 # how far back each check looks min_points: 3 # samples over a threshold before it counts as a breach ``` The full source is in the [Watchtower template](https://github.com/hubzoid/hubzoid/blob/main/hubzoid/templates/watchtower/workflows/watchtower/main.py). ## Declare a workflow Put workflow files in a folder under `workflows/`, such as `workflows//main.py`. Hubzoid imports every `.py` file one level down and skips files whose names start with `_`. | Name | Type | Default | Description | | --- | --- | --- | --- | | `schedule` | `str \| None` | `None` | A schedule phrase or a raw 5-field cron. Leave it out for a workflow you only run by hand. Can be passed as the first positional argument. | | `timezone` | `str \| None` | `UTC` | IANA timezone the schedule is evaluated in, such as Europe/London. Daylight saving is handled, so daily at 06:00 stays 06:00 local. | | `on_failure` | `str \| None` | `None` | An http or https URL that receives a POST when a run raises. Any other value writes an error line to the server log. | | `run_as` | `str \| None` | `None` | Email of the account every run acts as. It chooses whose permissions, state, artifacts, email and connections the run uses, and grants nothing. None uses HUBZOID\_WORKFLOW\_USER, else the owner recorded at setup. | Rules the loader enforces: - Always call the decorator, as `@workflow()` for a manual workflow or `@workflow("every 15 minutes")` for a scheduled one. - The workflow's name is the function name. It must be unique within the hub, and it is the name you pass to `hubzoid schedule run`, `pause` and `resume`. - The function takes no parameters. Read inputs from `hub.setting`, `hub.state`, files or your own steps. - Write `schedule`, `timezone` and `run_as` as string literals. Hubzoid reads them from the source to list workflows in the Console and in `hubzoid schedule list` without running your code. - The return value becomes the run's result, shown in the Console. Return plain data such as a dict. The schedule is checked when the file loads. An unknown phrase, an invalid cron or an unknown timezone is reported by `hubzoid doctor` as `workflows.definitions` and shown in the Console as a definition error. Hubzoid checks every workflow file before it imports any of them, so one definition error keeps all code workflows in the hub from loading until it is fixed and the hub restarts. Markdown tasks keep running. ## Schedule phrases Phrases are case-insensitive. Hubzoid translates each one to a cron expression and evaluates it in the workflow's timezone. | Phrase | Cron for the first example | Notes | | --- | --- | --- | | `every 15 minutes`, `every 5 min`, `every 2m` | `*/15 * * * *` | The interval must divide 60: 1, 2, 3, 4, 5, 6, 10, 12, 15, 20 or 30 | | `every 3 hours`, `every 1 hour`, `every 6h` | `0 */3 * * *` | The interval must divide 24: 1, 2, 3, 4, 6, 8 or 12 | | `daily at 6am`, `daily 06:30`, `every day at 18:00` | `0 6 * * *` | Once a day | | `every monday 08:30`, `every fri at 5pm` | `30 8 * * 1` | One weekday per phrase, full name or three letters | | `30 6 * * 1-5` | `30 6 * * 1-5` | A raw cron with digits and `* / , -` passes through | Times are `H:MM` or `HH:MM` on a 24-hour clock, or an hour with an optional `am` or `pm`, such as `6am` or `6 pm`. `12am` is midnight and `12pm` is noon. For minutes in the afternoon, use the 24-hour form, such as `17:30`. For several weekdays, use a raw cron. The dispatcher checks once a minute. The full grammar is in the [workflow API reference](https://hubzoid.com/docs/reference/workflow-api#schedule-grammar). ## Steps `@step` marks a function whose result the engine saves. A completed step is not run again when a run resumes after a restart. A step that was interrupted runs again, so make steps that change something outside the hub safe to repeat. ```python @step # tried once def fetch_orders(): ... @step(max_attempts=3) # retried, for calls that are safe to repeat def post_summary(text): ... ``` A step's result is stored in the run history and shown in the Console, so never return a secret. Read secrets inside the step that uses them instead of passing them in: ```python @step(max_attempts=3) def post_summary(text: str) -> str: token = hub.secret("chat_webhook_token") # read here, inside the step ... return "posted" ``` ## Call the model `hub.call_llm` is one model call with no tools. It returns text by default, a dict with `response_format="json"`, or a validated instance with a Pydantic `response_model`. ```python note = hub.call_llm("Summarize these counts in two sentences: ...", system="You are a careful analyst. Use only the data given.") data = hub.call_llm("Return a JSON object with keys store and change for ...", response_format="json") # a dict finding = hub.call_llm(prompt, response_model=Finding) # a Finding ``` - It uses the hub's model unless you pass `model`, which accepts any LiteLLM model id, `claude-local` or `codex-local`. - For JSON, Hubzoid appends an instruction (with the schema when you pass `response_model`) and uses the provider's JSON mode on LiteLLM models. On `claude-local` and `codex-local` it runs one turn with no tools. - The reply is parsed from the whole text, a fenced block or the outermost braces. A reply that is not a JSON object, or does not match `response_model`, raises `ModelOutputError` with the raw reply in `.raw`. - The call is a saved step and is retried once. A resumed run does not pay for the same call twice. `hub.call_agent` runs the hub's full agent, with its tools, skills and knowledge: ```python draft = hub.call_agent("Read output/watchtower/latest.md and draft a two-line note for the on-call channel.") ``` It returns the agent's reply, or a validated instance when you pass `response_model`, in which case the agent is asked to finish with JSON matching the schema. A failed agent run raises `AgentRunError` and fails the workflow run. Because the agent can call tools that write, it is not retried unless `workflows/settings.yaml` sets `agent_max_attempts: N`. The agent acts as the run's account, so it uses that person's grants and their own Open WebUI connections, never the author's. `hub.call_jev` is experimental. It asks TypeSafe's Jev, through OpenRouter, for typed yes or no, choice and score decisions, and needs its own `JEV_OPENROUTER_API_KEY`. See the [workflow API reference](https://hubzoid.com/docs/reference/workflow-api). Every model call writes a usage row for the run's account, so tokens and estimated cost appear on the agent's card on the Console's **Agents** page. ## State, settings and secrets | Call | Reads | Notes | | --- | --- | --- | | `hub.state[key]`, `hub.state.get(key)` | Durable key and value store for this workflow and the run's account | Values are JSON. Survives restarts and upgrades. Keyed by hub, workflow, account and key, so workflows never clash and the same workflow run for someone else starts empty. | | `hub.shared_state[key]` | The same store, shared by every account that runs the workflow | For data that is not about one person. Keep personal data out of it. | | `hub.run_dir` | A private folder for this run under `.hubzoid/runs/` | Write generated files here, then publish them. Agent file tools cannot read it. | | `hub.setting(key, default=None)` | Top-level keys of `workflows/settings.yaml` | Read at the start of each run. | | `hub.secret(name, default=None)` | The environment variable named `name` in upper case | Includes values from the hub's `.env` and `restricted/.env`. | A write to `hub.state` or `hub.shared_state` commits at once and is not part of the step checkpoint. Store a marker per finished item, as Watchtower does with `explained::`, and skip marked items. Do not keep a counter that you read, change and write back, because a replayed run can count twice. State written before 1.0.1 is kept but belongs to no account, so the first run as a person after an upgrade starts with empty `hub.state`. Two top-level keys in `workflows/settings.yaml` configure the engine and are read when the hub starts: ```yaml title="workflows/settings.yaml" agent_max_attempts: 2 # retry a failed hub.call_agent once (default 1) max_concurrent_workflows: 3 # at most 3 code workflows at once in this hub (no cap by default) ``` ## Publish a report and email it A run can publish a file it generated as a private artifact owned by the run's account, and email that account a link: ```python @workflow("daily at 8am", timezone="Asia/Kolkata", run_as="priya@company.com") def daily_report(): page = hub.run_dir / "report.html" # a private folder for this run page.write_text(build_html()) # your own generator report = hub.publish_artifact(page, title="Daily report") hub.send_email("Your daily report", "It is ready.", artifacts=[report]) ``` - `hub.publish_artifact` stores a new artifact on every call and returns its id and viewer link, `/portal/artifacts/`. It is private to the owner unless you pass `audience="hub"` or `audience="people"` with `share_with`. A workflow can never create a public link. - `hub.send_email` always goes to the run's own account, over SMTP (`HUBZOID_SMTP_*`) or to a preview outbox with `HUBZOID_EMAIL_DELIVERY=preview`. It raises unless the server accepted the message or it was previewed. - Both are recorded steps, so a resumed run does not publish twice or resend an accepted email. Both need the run to act as a person, not a legacy service identity. See the [workflow API reference](https://hubzoid.com/docs/reference/workflow-api#hubpublish_artifact) for every parameter. ## Access A code workflow runs as an ordinary account, chosen by `run_as`, else `HUBZOID_WORKFLOW_USER`, else the owner recorded at setup. The resolution and its rules are in [Workflows and schedules](https://hubzoid.com/docs/concepts/workflows-and-schedules#who-a-run-acts-as). `hub.user.id` and `hub.user.email` return that account, and `hub.user.can(permission)` checks whether it holds a permission in the current hub now. Restricted tools reached through `hub.call_agent` are allowed only when that account has the tool's permission. Grant it in the Console with **Edit access**, or on the command line: ```bash hubzoid grant priya@company.com my-watchtower ``` `hubzoid schedule list` shows who each workflow runs as, or why it cannot run. Grants made to the legacy identity `workflow:` before 1.0.1 stay but are not used. See [Restrict tools](https://hubzoid.com/docs/guides/restrict-tools). ## Handle failure Any exception that leaves the workflow function fails the run. The engine marks it failed, and the error appears in `hubzoid schedule status` and on the agent's **Runs & schedules** tab in the Console. There the hub's managers see a failure summary, and the run's account sees the full error. With `on_failure` set to a URL, Hubzoid also sends a POST with a 10 second timeout: ```python @workflow("every 15 minutes", on_failure="https://alerts.example.com/hubzoid") def watchtower(): ... ``` ```json {"workflow": "watchtower", "hub": "my-watchtower", "error": "broken.jsonl line 2 is not valid JSON: Expecting value"} ``` The notification is best effort and never hides the failure itself. Watchtower includes a malformed file to show a failed run and its recovery: ```bash cp raw_data/samples/broken.jsonl raw_data/events/ hubzoid schedule run . watchtower # exits 1: broken.jsonl line 2 is not valid JSON hubzoid schedule status . # the failed run and its error rm raw_data/events/broken.jsonl hubzoid schedule run . watchtower # succeeds again ``` ## Scaffold a new workflow ```bash hubzoid new workflow review-prs my-hub hubzoid schedule run my-hub review_prs ``` `hubzoid new workflow` creates `workflows/review-prs/main.py` with a manual workflow named `review_prs` that reads and writes `hub.state` and calls one `@step`. It needs no model or outside service, and prints the command to run it once. Add a schedule and timezone after you have tested its effects. `hubzoid schedule run` takes the function name. Hyphens are read as underscores, so `review-prs` works too. ## Enable schedules Code workflows fire on schedule only on a deployment that is marked to run them, so a copy of the hub on a laptop never fires alongside the server. Manual runs with `hubzoid schedule run` work everywhere. ### Single hub ```bash echo "HUBZOID_SCHEDULES=1" >> .env hubzoid run . ``` ### Gateway `hubzoid gateway` sets `HUBZOID_GATEWAY=1` for every bridge it launches, which enables code workflow schedules without further configuration. See [Gateway](https://hubzoid.com/docs/deploy/gateway). Check that the schedule is live: - `hubzoid schedule list .` shows `workflow watchtower · scheduled · UTC · next ... · runs as ()`, or `cannot run:` with the fix. - The agent's **Runs & schedules** tab in the Console shows the workflow as **Scheduled** with its next run and **Runs as**. - `hubzoid doctor .` reports `scheduler.health`. It warns when the dispatcher has stopped sending its heartbeat. The hub loads workflow modules when it starts, so restart it after you change a workflow's code. Runs that were queued or interrupted under the old code are cancelled at that start, as described in [Operating runs](https://hubzoid.com/docs/guides/operating-runs#drain-before-a-deploy). ## Next steps - [Workflow API reference](https://hubzoid.com/docs/reference/workflow-api): Every decorator parameter, hub method, grammar rule and exception. - [Operating runs](https://hubzoid.com/docs/guides/operating-runs): Pause, resume, cancel, drain and troubleshoot scheduled work. - [Console runs](https://hubzoid.com/docs/console/runs): Read run status, results and steps across your agents. - [Configuration reference](https://hubzoid.com/docs/reference/configuration): Environment variables such as HUBZOID\_SCHEDULES and HUBZOID\_DBOS\_DB.