Code workflows

A code workflow is a Python function under workflows/<name>/ 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 thisCopy Anchor Link

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 is simpler. Both run on the same engine, as described in Workflows and schedules.

Start from the Watchtower sampleCopy Anchor Link

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 as chat.

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.

The workflow fileCopy Anchor Link

Detection is plain Python in steps. Only the explanation comes from the model, as a validated Pydantic object.

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/<window>.md and copies it to latest.md. The workflow itself ties them together:

@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.

# 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 (opens in a new tab).

Declare a workflowCopy Anchor Link

Put workflow files in a folder under workflows/, such as workflows/<folder>/main.py. Hubzoid imports every .py file one level down and skips files whose names start with _.

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 phrasesCopy Anchor Link

Phrases are case-insensitive. Hubzoid translates each one to a cron expression and evaluates it in the workflow's timezone.

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.

StepsCopy Anchor Link

@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.

@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:

@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 modelCopy Anchor Link

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.

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:

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.

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 secretsCopy Anchor Link

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:<service>:<window_end>, 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:

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 itCopy Anchor Link

A run can publish a file it generated as a private artifact owned by the run's account, and email that account a link:

@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/<id>. 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 for every parameter.

AccessCopy Anchor Link

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. 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:

hubzoid grant priya@company.com <permission> my-watchtower

hubzoid schedule list shows who each workflow runs as, or why it cannot run. Grants made to the legacy identity workflow:<name> before 1.0.1 stay but are not used. See Restrict tools.

Handle failureCopy Anchor Link

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:

@workflow("every 15 minutes", on_failure="https://alerts.example.com/hubzoid")
def watchtower(): ...

{"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:

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 workflowCopy Anchor Link

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 schedulesCopy Anchor Link

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.

echo "HUBZOID_SCHEDULES=1" >> .env
hubzoid run .

Check that the schedule is live:

• hubzoid schedule list . shows workflow watchtower · scheduled · UTC · next ... · runs as <account> (<where it came from>), 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.

Next stepsCopy Anchor Link

Read this page as Markdown