Skip to content
Docs

Documentation navigation

Scheduling

Cron, interval, and at-time schedules with retry, jitter, and process-wide job registry helpers.

#Scheduling and durable triggers

Agent and Swarm expose cron() and interval() builders. Their default mode="remote" creates a durable server-side trigger; mode="local" opts into the in-process compatibility scheduler.

#Remote mode (default)

A durable registration needs a stable key and an immutable deployed-agent target before invoke() registers it:

Python 
from maivn import Agent

agent = Agent(name="daily-briefing", api_key="...")
trigger = (
    agent.cron("0 9 * * MON-FRI", tz="America/Chicago")
    .key("daily-briefing")
    .named("Daily briefing")
    .target("agent-123", version=4)
    .with_options(max_attempts=3, backoff="exponential")
    .invoke("Compose the daily operations briefing")
)

print(trigger.scenario_id, trigger.trigger_id, trigger.status)

Remote interval triggers accept a whole-second timedelta. The builder also supports where() for a JSONPath admission predicate and emit() for event output. with_options() configures concurrency, retries, deduplication, lineage/fanout bounds, shadow mode, and the fire budget.

An attached Trigger supports history(), replay(), pause(), resume(), delete(), fire(), and fire_stream() plus async equivalents. These methods perform server-side operations and require an authorized deployment.

#Local compatibility mode

Use local mode only when the process itself should own the scheduler lifetime:

Python 
from maivn import Agent

agent = Agent(name="local-briefing", api_key="...")
job = agent.cron("0 9 * * *", mode="local").invoke("Compose the briefing")
job.on_fire(lambda receipt: print(receipt.fired_at))
job.on_error(lambda receipt: print(receipt.error))

Local mode returns ScheduledJob and retains the compatibility callbacks and local lifecycle controls. It is not durable across process restarts.

#Local lifetime and one-shot jobs

Local .invoke(...) starts the job immediately. The scheduler uses background threads; a short script must keep its process alive until shutdown:

Python 
import time

try:
    while not job.is_done:
        time.sleep(1)
except KeyboardInterrupt:
    pass
finally:
    job.stop(drain=True, timeout=30)

agent.at(when).invoke(...) is a local one-shot job. Unlike cron and interval, at has no remote-mode option. Use a timezone-aware datetime. For local retries pass Retry(...); durable remote retries use the trigger builder's .with_options(...) instead.