---
question: How do I run a workflow on a cron, an interval or an event?
kind: task
via_web: na
via_cli: na
via_python: available
via_api: na
via_config: na
ref_python: Cron / Every / OnEvent
prereq: You need ABI 3.x, a NATS project (the default of `abi new project` on 3.x) and a workflow. [Write a durable workflow](/v2/capabilities/build/workflows/write-a-durable-workflow) or [Create a workflow](/v2/capabilities/build/agents/creating-a-workflow) first.
next:
  - /v2/capabilities/build/workflows/monitor-and-cancel-a-run
  - /v2/capabilities/build/orchestrations/schedule-a-job
  - /v2/capabilities/build/orchestrations/react-to-an-event
---

> **You are reading the v2.x docs (engine modules). New projects use v3.** [Read this page in v3](/capabilities/build/workflows/schedule-a-workflow) · [Migrate from 2.x to 3.0](/capabilities/upgrade/migrate-to-3)

# Schedule a workflow

A workflow runs on a schedule, or when an event happens, through its triggers. They are the triggers of [module jobs](/v2/capabilities/build/modules/module-jobs): JetStream fires them and one replica of the module runs each one. A workflow adds an input to each schedule, so every tick starts a run with known parameters.

Where the triggers go depends on the form of the workflow:

| Workflow | Triggers |
|---|---|
| A `@workflow` method | `triggers=(...)` on the decorator |
| A workflow class | Not on the class: a `@workflow` on the module that calls it |

Triggers are ABI 3.x with NATS. On ABI 2.x, and on 3.x without NATS, a workflow runs on a schedule through a Dagster orchestration that calls its `run`: see [Dagster](/v2/capabilities/build/applications/dagster) (legacy on 3.x).

## Web UI

Not applicable. No screen sets a schedule: triggers are declared in code. System › Jobs shows each workflow's schedule, the input each trigger sends, and the next tick, and starts a run by hand (**Run now**). See [Monitor and cancel a run](/v2/capabilities/build/workflows/monitor-and-cancel-a-run?via=web).

## CLI

Not applicable. No `abi` command schedules a workflow. Declare the triggers in Python.

## Python

### Schedule a `@workflow`

Pass the triggers to the decorator. `input` is the run's input, checked against the input model:

```python title="src/reports/__init__.py"
from datetime import timedelta

from naas_abi_core.module.jobs import Cron
from naas_abi_core.module.Module import BaseModule
from naas_abi_core.module.workflows import SyncFlowContext, workflow


class ABIModule(BaseModule):
    @workflow(
        triggers=(
            Cron("0 0 6 * * *", time_zone="Europe/Paris", input={"regions": ["emea", "amer"]}),
        ),
        timeout=timedelta(minutes=30),
        max_attempts=3,
    )
    def daily_report(self, ctx: SyncFlowContext, inp: ReportInput) -> ReportOutput:
        ...
```

The input leaves `day` out, so each tick reports on the day before it: the run works it out in a step of its own ([Write a durable workflow](/v2/capabilities/build/workflows/write-a-durable-workflow?via=python#write-the-method)). Give a trigger only what is the same on every tick; a fixed date would make every tick write the same report.

`Cron`, `Every` and `OnEvent` come from `naas_abi_core.module.jobs`, `workflow` from `naas_abi_core.module.workflows`.

- `Cron` takes six fields, seconds first, or an alias such as `@daily`, and an optional `time_zone` (UTC without one). See [Schedule a job](/v2/capabilities/build/orchestrations/schedule-a-job?via=python) for the syntax.
- `Every` takes a duration (`"30s"`, `"6h"`) or a `timedelta`, at least one second.
- `input` is a JSON object, `{}` when it is left out. Changing it republishes that schedule only.
- A workflow takes up to 16 triggers. Every workflow can also be started by hand, with or without triggers.

### Start it on an event

`OnEvent(event_type=..., filter={...})` or `OnEvent("<subject>")` starts a run when the event arrives, as for a job ([React to an event](/v2/capabilities/build/orchestrations/react-to-an-event?via=python)). A `@workflow`'s input is then the event as a job gets it, `{"subject": ..., "data": {...}}`: type the input as `dict`, or as a model with those two fields.

Events are at most once: an event published while no replica of the module runs is not delivered.

### Schedule a workflow class

The engine publishes a workflow class as `workflow.<slug>` with no trigger, one attempt and one run at a time. To schedule it, declare a `@workflow` on the module that calls `run()` in a step:

```python title="src/my_module/__init__.py"
from naas_abi_core.module.jobs import Cron
from naas_abi_core.module.Module import BaseModule
from naas_abi_core.module.workflows import SyncFlowContext, workflow


class ABIModule(BaseModule):
    @workflow(triggers=(Cron("0 0 2 * * *", input={"entity_id": "42"}),))
    def nightly_customers(self, ctx: SyncFlowContext, inp: dict) -> dict:
        from my_module.workflows.CustomersWorkflow import (
            CustomersWorkflow,
            CustomersWorkflowConfiguration,
            CustomersWorkflowParameters,
        )

        workflow = CustomersWorkflow(CustomersWorkflowConfiguration())
        return ctx.step("run", workflow.run, CustomersWorkflowParameters(**inp))
```

The import stays inside the method when the workflow class reads its services through `ABIModule.get_instance()`, as the class on [Create a workflow](/v2/capabilities/build/agents/creating-a-workflow) does: its file imports the module, and a top-level import here would be circular. A class that imports nothing from the module can be imported at the top. A plain `@job` that calls `run()` works too, without the input check and the stored step: see [Module jobs](/v2/capabilities/build/modules/module-jobs).

### Restart the engine

The module publishes its schedules when the engine loads it. The workflow then shows its triggers in System › Jobs.

## REST API

Not applicable. No route sets a schedule. An outside scheduler can start runs with `POST /api/workflows/{module_id}/{name}/runs`: see [Run a workflow](/v2/capabilities/build/workflows/run-a-workflow?via=api).

## Config

Not applicable. `config.yaml` has no per-workflow key: triggers live in code. A `nats:` block is needed for any trigger to fire. `nats.workflows.bridge: false` stops the engine from publishing its modules' classes as `workflow.<slug>` and `pipeline.<slug>`.

## When startup fails

A trigger list over 16, an interval under one second or a trigger input that is not a JSON object is refused when the decorator is read. The module's descriptor is capped at 64 KiB, each schema at 16 KiB: a module with many workflows or large models can be refused at registration, with the limit in the message.

A `@workflow`'s trigger input is not checked at startup: a tick whose input does not fit the input model fails that run, without a retry.

## Missed ticks and repeats

A missed cron tick fires once when the broker comes back, not once per missed tick. Delivery is at least once and `max_attempts` may retry: a scheduled run must be safe to repeat. See [Make a run safe to repeat](/v2/capabilities/build/workflows/write-a-durable-workflow#make-a-run-safe-to-repeat).