---
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 a NATS project (the default of `abi new project`) and a workflow. [Write a durable workflow](/capabilities/build/workflows/write-a-durable-workflow) or [Create a workflow](/capabilities/build/agents/creating-a-workflow) first.
next:
  - /capabilities/build/workflows/monitor-and-cancel-a-run
  - /capabilities/build/orchestrations/schedule-a-job
  - /capabilities/build/orchestrations/react-to-an-event
---

# Schedule a workflow

A workflow runs on a schedule, or when an event happens, through its triggers. They are the triggers of [module jobs](/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, in any module | `triggers=(...)` on the decorator |
| A workflow class of a remote module | `job = JobSettings(triggers=(...))` on the class |
| A workflow class of an engine module | Not on the class: a `@workflow` on the module that calls it |

## 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](/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_sdk.jobs import Cron
from naas_abi_sdk.workflows import FlowContext, 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,
    )
    async def daily_report(self, ctx: FlowContext, 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](/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.

In an engine module, import `Cron`, `Every` and `OnEvent` from `naas_abi_core.module.jobs`, and `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](/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](/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. A workflow class takes the event's `data` as its parameters.

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

### Schedule a class of a remote module

Set `job` on the class. A Cron or Every `input` is the class's parameters:

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

from naas_abi_sdk.jobs import Cron, JobSettings
from naas_abi_sdk.workflow import Workflow, WorkflowParameters


class CustomersWorkflow(Workflow[CustomersWorkflowParameters]):
    """Summarize an entity from the knowledge graph."""

    job = JobSettings(
        triggers=(Cron("0 0 2 * * *", input={"entity_id": "42"}),),
        timeout=timedelta(minutes=10),
        max_attempts=3,
    )

    async def run(self, parameters: CustomersWorkflowParameters) -> dict:
        ...
```

`JobSettings` is `(triggers=(), timeout=None, max_attempts=1, max_concurrency=1)`. A class without `job` keeps those defaults: no trigger, one attempt, one run at a time.

### Schedule a class of an engine module

The engine publishes an engine module's class as `workflow.<slug>` with no trigger, one attempt and one run at a time, and reads no `job` setting on the class. 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](/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](/capabilities/build/modules/module-jobs).

### Restart the module

The module publishes its schedules when it starts. 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](/capabilities/build/workflows/run-a-workflow?via=api).

## Config

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

## When startup fails

A remote module raises before it connects when:

- a class's `job` is not a `JobSettings`;
- a class's `Cron` or `Every` input does not validate against its parameters model (give the trigger an `input=`, or the fields defaults);
- `workflow.<slug>` or `pipeline.<slug>` is also the name of a `@job`, a `@workflow` or an entry in `jobs`;
- the module has workflow or pipeline classes but no discovery, or no `document` service in its dependencies.

In any module, a trigger list over 16, an interval under one second or a trigger input that is not a JSON object is refused when the class or 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](/capabilities/build/workflows/write-a-durable-workflow#make-a-run-safe-to-repeat).