Code Workflows: AI Agent Steps in Python

Write recurring work as Python steps that call the model only where judgment helps, then publish reports, take inputs and let people run it from chat.

A code workflow is a Python function in your Hub's workflows/ folder. Code decides each step. The model is called only where you ask it to, with hub.call_llm or hub.call_agent. Each step's result is saved, so a run that stops resumes after its last finished step.

In this guide you create a workflow in the my-hub starter, run it once, turn it into a weekly brief that publishes a report with one lasting link, and put it on a schedule.

Video lesson 7 · 3:03The daily report: an AI agent workflow on a scheduleRun a daily report manually, inspect it, then schedule delivery.

When to use a code workflow

Use one 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 and share one run history.

Code workflows start on their own only where you allow it: HUBZOID_SCHEDULES=1 in the Hub's .env, or a deployment that hubzoid run starts from hubzoid.yaml or with several Hubs. Markdown tasks need no such setting.

What you need

The my-hub starter from the quickstart, with a model that answers in chat.
A terminal in the folder that holds my-hub/.

Create the workflow

Terminal

hubzoid new workflow weekly-brief my-hub

This creates my-hub/workflows/weekly-brief/main.py and prints the command to run it once. The workflow's name is the function name, weekly_brief:

my-hub/workflows/weekly-brief/main.py

"""The weekly-brief workflow. A manual example with durable steps and state.

A workflow coordinates steps, calls agents, retains state, retries, and resumes
after a restart. Add a schedule only when ready. Read a secret INSIDE a step,
never at the top (step inputs are checkpointed).
"""
from hubzoid import workflow, step, hub


@workflow()  # Manual first. Add a schedule and timezone when ready.
def weekly_brief():
    # example: durable, idempotent work
    seen = hub.state.get("seen", 0)
    result = {"message": "Your workflow is ready"}
    do_something(result)
    hub.state["seen"] = seen + 1
    return seen + 1


@step   # a durable side effect — make it idempotent (at-least-once)
def do_something(result):
    print(f"[weekly-brief] {result}")

It needs no model and no outside service. hub.state keeps a counter between runs.

Run it once

Terminal

hubzoid schedule run my-hub weekly_brief

The command ends by printing what the function returned. On the first run that is 1, as in ✓ workflow weekly_brief returned: 1. Run it again and the number goes up, because hub.state kept the count. When hubzoid run my-hub is already running, the command hands the run to it and prints Workflow weekly_brief: 2.

Make it do real work

Replace the file with a workflow that reads the house rules, asks the model for a brief and publishes it as a report:

my-hub/workflows/weekly-brief/main.py

from pathlib import Path

from hubzoid import hub, step, workflow

HUB_DIR = Path(__file__).resolve().parents[2]


@step
def read_rules() -> str:
    return (HUB_DIR / "knowledge" / "house-rules.md").read_text(encoding="utf-8")


@workflow()
def weekly_brief():
    """A short brief of the house rules for the support team."""
    rules = read_rules()
    brief = hub.call_llm(
        "Write a short brief for the support team: support hours, the refund "
        "window and the large-order rule, one line each.\n\n" + rules,
        system="Use only the rules given.",
    )
    page = hub.run_dir / "brief.md"
    page.write_text(brief, encoding="utf-8")
    report = hub.publish_artifact(page, title="Weekly brief", key="weekly-brief")
    return {"report": report["latest_url"]}

@step saves read_rules()'s result, so a resumed run does not read again.
hub.call_llm is one model call with no tools, also saved as a step.
hub.run_dir is a private folder for this run.
hub.publish_artifact publishes the file as a report. With key= it becomes a live report: one link that always shows the newest copy.

Run it again with hubzoid schedule run my-hub weekly_brief. The command prints the result, then Report: and the live report's link.

Put it on a schedule

Give the decorator a schedule:

my-hub/workflows/weekly-brief/main.py

@workflow("every monday 09:07")
def weekly_brief():
    ...

Then allow code workflows to start on their own on this machine, and restart the Hub:

my-hub/.env

HUBZOID_SCHEDULES=1

The Hub loads workflow code when it starts, so restart it after every change to a workflow.

Check it is live

Open Settings → Agents → My Hub → Workflows. The workflow is listed with a status line that says when it runs next, the account it runs as, Run now and View runs. hubzoid schedule list my-hub shows the same from the terminal.

If the status line says schedules are off, HUBZOID_SCHEDULES=1 is missing from the .env of the Hub that is running, or the Hub has not restarted since you added it.

Declare a workflow

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 _. The workflow API reference lists every option of @workflow.

The most used @workflow options

Prop

Type

Rules the loader enforces

Always call the decorator: @workflow() for a manual workflow, @workflow("every 15 minutes") for a scheduled one.
The function name is the workflow's name. It must be unique in the Hub, ignoring case. hubzoid schedule run also accepts it with hyphens, so weekly-brief works.
The function takes no parameters. Read inputs from hub.input, hub.setting, hub.state, files or your own steps.
Write schedule, timezone, run_as, on_webhook, as_tool, schedulable, sensitive and surfaces as literal values. Hubzoid reads them from the source to list workflows without running your code.
The return value is the run's result, shown in Settings. Return plain data such as a dict, and never a secret.

Where definition errors show, and why one broken file stops every workflow

An unknown phrase, an invalid cron or an unknown time zone is reported by hubzoid doctor under workflows.definitions, and Settings shows it after "Can't run". Hubzoid checks every workflow file before it imports any of them, so one broken file stops all code workflows in the Hub from loading until it is fixed and the Hub restarts. Markdown tasks keep running.

Schedule phrases

A schedule is a plain phrase such as every monday 08:30, or a 5-field cron. The full grammar is in the workflow API reference.

Every phrase form, with its cron

Phrase	Cron for the first example	Notes
`every 15 minutes`, `every 5 min`, `every 2m`	`/15 * * *`	The interval must divide 60
`every 3 hours`, `every 1 hour`, `every 6h`	`0 /3 * *`	The interval must divide 24
`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, full name or three letters
`30 6 * * 1-5`	`30 6 * * 1-5`	A cron passes through unchanged

Phrases ignore case. Times are H:MM or HH:MM on a 24-hour clock, or an hour with am or pm. For several weekdays, use a cron.

Steps and model calls

@step marks a function whose result the engine saves. A finished 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.

Retried and single-attempt steps

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 shown in Settings, so read secrets inside the step that uses them, with hub.secret("name"), and never return one.

Call the model with hub.call_llm, one model call with no tools, or hub.call_agent, the Hub's full agent with its tools, skills and knowledge.

Model calls, retries, timeouts and cost

Call	What it does
`hub.call_llm(prompt, system=..., response_model=...)`	One model call with no tools. Returns text, a dict with `response_format="json"`, or a validated Pydantic instance with `response_model`. Retried once. A reply that does not match raises `ModelOutputError`, with the raw reply in `.raw`.
`hub.call_agent(task, response_model=...)`	The Hub's full agent, with its tools, skills and knowledge, acting as the run's account. Not retried unless `agent_max_attempts` is set in `workflows/settings.yaml`, because a retry can repeat a write. A failure raises `AgentRunError`.
`hub.call_jev(state, questions)`	Experimental typed decisions from TypeSafe's Jev through OpenRouter. Needs `JEV_OPENROUTER_API_KEY`.

Both main calls use the Hub's model unless you pass model, and take a timeout: 120 seconds for hub.call_llm and 10 minutes for hub.call_agent by default. A finished call is not made again when a run resumes. A call interrupted part way can run again and is charged again. Every call counts toward the run's account in Settings → Usage.

State, settings and secrets

A run reads lasting state, settings and secrets through hub.

Every state, folder, setting and secret call

Call	Reads
`hub.state[key]`	A lasting store for this workflow and the run's account. Values are JSON. The same workflow run for someone else starts empty.
`hub.shared_state[key]`	The same store, shared by every account that runs the workflow. Keep personal data out of it.
`hub.run_dir`	A private folder for this run, for files you then publish. Agent file tools cannot read it.
`hub.data_dir`	The Hub's private data folder, `.hubzoid/data/`. Backups include it.
`hub.setting(key, default=None)`	A top-level key of `workflows/settings.yaml`, read at the start of each run.
`hub.secret(name, default=None)`	The environment variable `name` in upper case, including the Hub's `.env`.

A write to hub.state saves at once and is not part of a step. Store a marker per finished item and skip marked items, rather than a counter you read, change and write back, because a resumed run can count twice.

How a run tells how it started

A run can tell how it started. hub.trigger is schedule, personal_schedule, tool, operator, cli, webhook or update. hub.started_by is the account that started it, or None for a schedule or a webhook. hub.team_run is true when the run acts as the workflow's own account.

Take inputs

A workflow can take typed inputs from a Pydantic model. The run reads them, validated, as hub.input.

An example with inputs, and where inputs come from

my-hub/workflows/stock-report/main.py

from typing import Literal

from pydantic import BaseModel, Field

from hubzoid import hub, workflow


class StockReportInputs(BaseModel):
    store: Literal["north", "south"] = Field(description="The store.")
    days: int = Field(7, description="How many days back to look.")


@workflow("daily at 6am", timezone="Europe/London", run_as="reports@example.com",
          inputs=StockReportInputs, schedule_input={"store": "north"})
def stock_report():
    """Stock movement for one store over the last few days."""
    page = build_page(hub.input.store, hub.input.days)   # your own step
    return hub.publish_artifact(page, title=f"Stock report: {hub.input.store}",
                                key=f"stock-{hub.input.store}")

Where the run starts	Where the inputs come from
The declared schedule	`schedule_input`, required when the model has required fields.
Run now in Settings	A form built from the model, on the agent's Workflows tab.
The terminal	`hubzoid schedule run my-hub stock_report --input store=south --input days=30`. A value that looks like JSON, such as a number, is parsed.
Chat and MCP	The person's request, when the workflow is offered there. See Run for me.
A webhook	Nothing, so a webhook workflow's fields need defaults.

Inputs are stored with the run and shown in its history, so keep them short and never secret.

Masking and size limits

Field(json_schema_extra={"hidden": True}) masks a value in lists. Inputs are limited to 4 KB. max_input_kb raises that to at most 256 KB for a large value such as a diff, but not on a workflow people can schedule.

Run for me

as_tool=True offers the workflow as a tool in chat and over MCP, named after the workflow and described by its docstring. It runs as the person asking, with their access, and returns their report links.

Python

@workflow(inputs=StockReportInputs, as_tool=True, schedulable=True)
def stock_report():
    """Stock movement for one store over the last few days."""
    ...

Nobody can run it until they are granted Run for that workflow on the agent's Access tab.

Access, waiting, answers, options and where it runs

Access. Each such workflow adds a capability on the agent's Access tab, under Hubzoid tools, then Workflows: Run followed by its title, such as Run Stock report (wf_stock_report). Nobody has it until it is granted.
Waiting. The tool waits up to tool_wait for the result: 60 seconds by default, at most 10 minutes. A longer run keeps going, and the person asks "is it ready?" later.
Answers. Return {"answer": "..."} to show text in chat after any links, up to 8,000 characters. Return {"refused": "..."} to show a refusal as written.
Options. confirm=True asks before running. surfaces=("web", "mcp") limits where the tool appears. sensitive=True marks the grant Sensitive in Settings.
Where it runs. Code workflows must run on the deployment, as described in Put it on a schedule. It is never offered on Slack or inside a run. HUBZOID_WORKFLOW_TOOLS=false in a Hub's .env removes it from that agent.

Personal schedules

With schedulable=True, people granted Schedule reports for themselves can ask in chat, for example "run the stock report for me every Monday at 9". The agent shows what will run, as whom and the next runs, and saves the schedule only after a yes. The first run starts at once. Each run acts as that person, so it stops when their access does.

Pauses, management, email and limits

A schedule pauses, with the reason shown, after three failures in a row, or when the person's access or the workflow's inputs change.
People manage their schedules in chat or in Settings → My schedules. Administrators see everyone's in Settings → Team schedules, with each one's model cost over 30 days, and can pause or stop them. Only an administrator resumes what an administrator paused.
Nobody is emailed about personal schedules unless workflows/settings.yaml sets notify_by_email: true and email delivery is set up.
min_interval="6h" on the decorator raises the shortest cadence for one workflow. The Hub limits are in workflows/settings.yaml:

my-hub/workflows/settings.yaml

personal_schedules:
  min_interval: 1h          # shortest cadence, the default
  per_person: 10            # schedules per person per agent, the default
  pause_after_failures: 3   # the default
notify_by_email: true       # off by default

Publish reports and email them

A run can publish a file it made as a report owned by the run's account, and email a link:

Python

@workflow("daily at 8am", timezone="Europe/London", run_as="reports@example.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", key="daily-report")
    hub.send_email("Your daily report", "It is ready.", artifacts=[report])

Audiences, email delivery and resumed runs

hub.publish_artifact returns the report's id and viewer link, /portal/artifacts/<id>. It is private to the owner unless you pass audience="hub" (everyone who can use the agent), audience="people" with share_with, or audience="domain". A workflow never creates a public link.
hub.send_email 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. raise_on_failure=False returns the result instead.
Both are saved steps, so a resumed run does not publish twice or resend an accepted email.

Live reports

key= (lowercase letters, digits, - and _) makes the publish a copy in a live report, with one link, latest_url, that always shows the newest copy.

Team and personal copies, unchanged copies and Update now

A team run writes the agent's report, at /portal/latest/<agent>/<key>. Run for me and personal schedules write that person's own copy, private to them until they share it.
When the file, or digest= (a short hash of the data behind it), matches the newest copy, nothing new is stored and the call returns that copy with "unchanged": true.
The page shows when the report was updated and when it updates next. The owner of a personal report, and people with Run and control workflows for a team one, have Update now, at most once every 5 minutes.
When a scheduled team run stops publishing a key it published before, that page says "No longer updating" and keeps its last copy.
A key belongs to one workflow in the agent. series_keep: 50 in workflows/settings.yaml keeps the last 50 copies. Without it, every copy is kept.

Company links

audience="domain" gives a team report a company link, /portal/r/<token>, that anyone at your company's domains can open after signing in, without access to the agent. They get a reader-only account. A person shares their own report the same way from the viewer's Share.

Sign-in, settings and the permission it needs

It needs company sign-in with Google, Microsoft or OpenID Connect (see authentication) and these deployment settings:

.env

HUBZOID_SHARE_DOMAINS=example.com     # who may open company links
OAUTH_ALLOWED_DOMAINS=example.com     # list each share domain here too
ENABLE_OAUTH_SIGNUP=true              # readers get an account on first visit

The publishing account needs Share reports with the company domain. Taking that away, or turning the link off, stops the link at once. Administrators find every live report in Settings → Team schedules, under Live reports, and copy, replace or turn off company links there.

Email lists and Markdown email

A team run can email each address on a list in workflows/settings.yaml, and send the body as Markdown.

An example list and its limits

my-hub/workflows/settings.yaml

email_lists:
  store-managers: [asha@example.com, ravi@example.com]

Python

hub.send_email("Weekly stock report",
               f"## This week\n\n- Open the [live report]({report['latest_url']})",
               email_list="store-managers", body_format="markdown")

email_list= emails each address on a list in workflows/settings.yaml, one message each, from a team run only. Every address must belong to an account on the deployment or to a share domain. A list holds at most 50 addresses and the body at most 2,000 characters, so link the report rather than pasting it.
body_format="markdown" sends headings, lists, emphasis and links with fixed styles. Raw HTML is escaped, images are removed and a plain-text part is kept.

See the workflow API reference for every option.

Who a run acts as

Team runs (the declared schedule, a webhook, Run now and the terminal) act as run_as, else HUBZOID_WORKFLOW_USER, else the owner recorded at setup. Run for me and personal schedules act as the person. hub.user.email returns the run's account, and hub.user.can(permission) checks whether it holds a permission now. See who a run acts as.

Restricted tools inside a run

Restricted tools reached through hub.call_agent are allowed only when the run's account holds the tool's permission. Grant it on the agent's Access tab in Settings, or with hubzoid grant priya@example.com <permission> my-hub. Grants that older releases made to workflow:<name> are kept but no run uses them. See restrict tools.

When a run fails

Any exception that leaves the workflow function fails the run. The error appears in hubzoid schedule status and in the run's history in Settings. The Hub's managers see a failure summary, and the run's account sees the full error.

To hear about failures without watching, set alert destinations in workflows/settings.yaml, or per workflow with alert_to=. See alerts.

Which failures raise alerts

A failed run, an overdue run and a streak of failures each raise an alert, and a schedule whose scheduled runs fail 20 times in a row pauses itself.

A run's deadline comes from timeout= or workflow_timeout. Running Python cannot be stopped at its deadline: the run raises an overdue alert and keeps going until the code returns. Put your own timeouts on network and file calls.

Test without the engine

run_local runs a workflow once, in a context like a real run, without the engine. Publishing and email are captured, never stored or sent.

An example test and what run_local covers

test_stock_report.py

from pathlib import Path

from hubzoid.workflows.runtime import load_definitions
from hubzoid.workflows.testing import run_local

hub_dir = Path("my-hub").resolve()
wf = next(w for w in load_definitions(hub_dir) if w.name == "stock_report")
out = run_local(wf.fn, hub_dir=hub_dir, inputs={"store": "south"},
                as_user="priya@example.com")
print(out.result, out.artifacts, out.emails)   # nothing was published or sent

Inputs are validated, and hub.input, hub.user, hub.trigger, hub.state and hub.setting work. Steps run directly.
By default it runs like a Run for me request from as_user. trigger="schedule", "cli" or "webhook" makes it a team run, and event= passes a webhook event.
Model and agent calls are not set up for you. Test a workflow that calls the model with hubzoid schedule run.

If something goes wrong

What you see	What to do
`no task or workflow 'weekly_brief' under ...`	Check the function name. The command lists the workflows it found.
`✗ workflow weekly_brief failed:` and an error	The run raised an exception. Fix it and run again. The failed run stays in the history.
The Workflows tab says schedules are off	Add `HUBZOID_SCHEDULES=1` to the Hub's `.env` and restart the Hub.
A change to the code has no effect	The Hub loads workflow code when it starts. Restart it.
Can't run on a workflow	Run `hubzoid doctor my-hub` and fix the file named under `workflows.definitions`, then restart.
A Run for me tool is missing in chat	Code workflows are not running on the deployment, the person lacks Run for that workflow, or `HUBZOID_WORKFLOW_TOOLS=false` is set.

What you see	What to do
`already exists:` from `hubzoid new workflow`	A folder with that name is already in `workflows/`. Pick another name.
`Workflows do not accept --timeout, --max-rounds or --model.`	Those options are for markdown tasks. Set limits in the workflow code.
`ModelOutputError`	The model's reply did not match the requested shape. Tighten the prompt or the Pydantic model.

Code workflows

When to use a code workflow

What you need

Create the workflow

Run it once

Make it do real work

Put it on a schedule

Check it is live

Declare a workflow

Schedule phrases

Steps and model calls

State, settings and secrets

Take inputs

Run for me

Personal schedules

Publish reports and email them

Live reports

Company links

Email lists and Markdown email

Who a run acts as

When a run fails

Test without the engine

If something goes wrong

Next steps

Workflow API reference

Webhooks

Operating runs

Runs in Settings

On this page

Code workflows

When to use a code workflowCopy Anchor Link

What you needCopy Anchor Link

Create the workflowCopy Anchor Link

Run it onceCopy Anchor Link

Make it do real workCopy Anchor Link

Put it on a scheduleCopy Anchor Link

Check it is liveCopy Anchor Link

Declare a workflowCopy Anchor Link

Schedule phrasesCopy Anchor Link

Steps and model callsCopy Anchor Link

State, settings and secretsCopy Anchor Link

Take inputsCopy Anchor Link

Run for meCopy Anchor Link

Personal schedulesCopy Anchor Link

Publish reports and email themCopy Anchor Link

Live reportsCopy Anchor Link

Company linksCopy Anchor Link

Email lists and Markdown emailCopy Anchor Link

Who a run acts asCopy Anchor Link

When a run failsCopy Anchor Link

Test without the engineCopy Anchor Link

If something goes wrongCopy Anchor Link

Next stepsCopy Anchor Link

Workflow API reference

Webhooks

Operating runs

Runs in Settings

On this page

When to use a code workflow

What you need

Create the workflow

Run it once

Make it do real work

Put it on a schedule

Check it is live

Declare a workflow

Schedule phrases

Steps and model calls

State, settings and secrets

Take inputs

Run for me

Personal schedules

Publish reports and email them

Live reports

Company links

Email lists and Markdown email

Who a run acts as

When a run fails

Test without the engine

If something goes wrong

Next steps