Skip to main content

triggers.cron

DBOS cron-based trigger support for durable agents.

Uses DBOS scheduled workflows to run agents on a cron schedule. When the trigger fires, it retrieves the registered agent and executes it with a configurable input message.

CronTriggerConfig Objects​

@dataclass
class CronTriggerConfig()

Configuration for a cron-based trigger.

Attributes:

  • agent_id - The agent to trigger.
  • cron_expression - Cron expression (e.g. "0 6 1 * *" for 6am on 1st of each month).
  • input_message - Message to send to the agent when triggered.
  • enabled - Whether the trigger is active.
  • description - Human-readable description of the schedule.
  • metadata - Additional metadata for the trigger.

CronTriggerResult Objects​

@dataclass
class CronTriggerResult()

Result of a cron trigger execution.

CronTrigger Objects​

class CronTrigger()

Manages a DBOS-backed cron trigger for an agent.

Registers a DBOS scheduled workflow that runs the agent on the specified cron schedule. The workflow is durable — if the process crashes, DBOS will replay from the last committed step.

Example:

```python
trigger = CronTrigger(CronTriggerConfig(
    agent_id="monthly-report-agent",
    cron_expression="0 6 1 * *",
    input_message="Generate the monthly report",
))
trigger.start()
```

start​

def start() -> None

Register the DBOS scheduled workflow and start the trigger.

stop​

def stop() -> None

Stop the cron trigger.

register_cron_trigger​

def register_cron_trigger(config: CronTriggerConfig) -> CronTrigger

Register and start a cron trigger for an agent.

Arguments:

  • config - Cron trigger configuration.

Returns:

The started CronTrigger instance.

unregister_cron_trigger​

def unregister_cron_trigger(agent_id: str) -> bool

Stop and unregister a cron trigger.

Arguments:

  • agent_id - The agent whose trigger to remove.

Returns:

True if a trigger was found and removed, False otherwise.

get_cron_trigger​

def get_cron_trigger(agent_id: str) -> CronTrigger | None

Get the cron trigger for an agent, if any.

list_cron_triggers​

def list_cron_triggers() -> list[CronTrigger]

List all registered cron triggers.