AI Agent Cron Jobs: Run an Agent on a Schedule

An AI agent cron job runs an agent on a schedule: a digest every weekday at 08:00, a health check every five minutes, a cleanup every night. The schedule itself is the easy part. The hard parts are time zones, runs that take longer than the interval, ticks that fall while the agent is down, and jobs that must not publish twice. This guide explains how a cron-triggered agent works and shows how to build one with Promptise Foundry's CronTrigger, including tests that don't wait for the clock.

Put the agent in a long-running process and give it a cron trigger. The trigger computes the next time that matches the cron expression, waits for it, and hands the agent an event with the scheduled time. In Promptise you add a TriggerConfig(type="cron", ...) to a ProcessConfig and start an AgentProcess:

TriggerConfig(type="cron", cron_expression="0 8 * * 1-5", cron_timezone="Europe/Zurich")

Then decide three things the schedule doesn't: what happens when a run is still going at the next tick, what happens to ticks missed while the process was down, and how a job avoids doing its work twice.

• Schedules are UTC unless you set `cron_timezone`. With a time zone, daylight saving time moves the run with local time.

• Ticks don't wait for runs. With concurrency=1, ticks that fire during a long run queue up and run back to back.

• Missed ticks are not caught up. If the process is stopped at 08:00, the 08:00 run never happens unless you add a catch-up step.

• Use the scheduled time as an idempotency key, so retries and duplicate ticks can't publish twice.

• Test with `process.inject()`, not by waiting for the clock.

A cron expression lists the minutes, hours, days of the month, months and weekdays a job runs on. 0 8 * * 1-5 means minute 0 of hour 8, Monday to Friday. A scheduler reads "now", finds the next time that matches, sleeps until then, and fires. Promptise uses the croniter library for the matching, so ranges (1-5), lists (1,15), steps (*/10) and weekday names (mon-fri) all work. A sixth field adds seconds: * * * * * */20 fires every 20 seconds. Macros such as @daily are rejected, and so is any expression with fewer than five or more than six fields.

Common schedules for agents:

One trap: when both the day-of-month and the weekday field are set, cron runs on either. 0 9 1 * 1 runs on the 1st of the month and on every Monday, not only on Mondays that fall on the 1st.

The trigger is one part of a pipeline. A listener task waits for the trigger, a shared queue holds events, and workers run the agent:

 CronTrigger                listener task          trigger queue        worker (concurrency=1)
 ───────────                ─────────────          ─────────────        ──────────────────────
 next match: 08:00 ──wait──▶ tick 08:00 ──filter──▶ [08:00]      ──────▶ run 08:00 ............
 next match: 08:05 ──wait──▶ tick 08:05 ──filter──▶ [08:05]              (still running)
 next match: 08:10 ──wait──▶ tick 08:10 ──filter──▶ [08:05, 08:10]       (still running)
                                                                 ──────▶ run 08:05 (late)
 Process stopped at 08:15 and started at 08:31: the next match is computed from 08:31,
 so 08:15, 08:20, 08:25 and 08:30 never fire.

Four consequences follow from this design, and the steps below show each one with real output:

• The trigger never waits for the agent. After each tick it computes the next one from the current time. Ticks keep firing while a run is in progress, and they wait in the queue.

• Nothing is stored between process starts. The trigger keeps no record of past ticks, so a restart starts counting from the moment it comes back.

• Each tick carries its scheduled time. The payload is {"scheduled_time": ..., "cron_expression": ..., "timezone": ...}, with the local UTC offset in the timestamp. That time names the slot the run belongs to, even when the run starts late.

• The filter runs before the queue. A filter_expression sees every tick and can drop one before it costs an LLM call.

The agent receives the tick as its user message, wrapped in a block that marks it as data from outside the system. The trigger overview shows that message format.

A schedule also isn't a good "is anything new?" check every minute when nothing changes for hours. Every tick is a full agent run that you pay for. Poll cheaply in a filter or a tool, and let the agent run only when there is work.

• Python 3.10 or later and Promptise Foundry 1.4.3 or later. croniter ships with it.

• An OpenAI API key in OPENAI_API_KEY. The examples use openai:gpt-5-mini, and a run costs well under a cent.

• About 20 minutes.

pip install "promptise>=1.4.3"

Check when a schedule fires before you trust it, especially in a time zone with daylight saving time. The trigger asks croniter for the next match after "now" in the trigger's time zone. This script makes the same call from start times you choose:

"""Preview when a cron schedule fires, with the same croniter call CronTrigger makes."""
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo

from croniter import croniter
from pydantic import ValidationError

from promptise.runtime import TriggerConfig


def preview(expression: str, tz: str, start: str, count: int) -> None:
    zone = ZoneInfo(tz)
    now = datetime.fromisoformat(start).replace(tzinfo=zone)
    print(f"{expression!r} in {tz}, from {now.isoformat()}")
    for _ in range(count):
        # Same call as CronTrigger._compute_next_fire: next match after "now".
        fire = croniter(expression, now).get_next(datetime)
        print("  fires at", fire.isoformat())
        # The trigger asks again just after firing. Step in UTC: local
        # arithmetic would lose which 02:30 we are in during the repeated hour.
        now = (fire.astimezone(timezone.utc) + timedelta(milliseconds=5)).astimezone(zone)


# Five fields: minute hour day-of-month month day-of-week
preview("0 9 * * 1-5", "Europe/Zurich", "2026-03-26 12:00", 3)  # across the March DST change
# Six fields: the sixth is seconds
preview("* * * * * */20", "UTC", "2026-10-19 08:00:00", 3)
# A job at 02:30 local time on the two DST change nights
preview("30 2 * * *", "Europe/Zurich", "2026-03-28 12:00", 2)  # 02:30 doesn't exist on Mar 29
preview("30 2 * * *", "Europe/Zurich", "2026-10-24 12:00", 3)  # 02:30 happens twice on Oct 25

# Bad schedules fail when the config is created, not at the first tick
for kwargs in (
    {"cron_expression": "0 9 * *"},
    {"cron_expression": "0 9 * * *", "cron_timezone": "Europe/Zurch"},
):
    try:
        TriggerConfig(type="cron", **kwargs)
    except ValidationError as exc:
        print("rejected:", exc.errors()[0]["msg"])

'0 9 * * 1-5' in Europe/Zurich, from 2026-03-26T12:00:00+01:00
  fires at 2026-03-27T09:00:00+01:00
  fires at 2026-03-30T09:00:00+02:00
  fires at 2026-03-31T09:00:00+02:00
'* * * * * */20' in UTC, from 2026-10-19T08:00:00+00:00
  fires at 2026-10-19T08:00:20+00:00
  fires at 2026-10-19T08:00:40+00:00
  fires at 2026-10-19T08:01:00+00:00
'30 2 * * *' in Europe/Zurich, from 2026-03-28T12:00:00+01:00
  fires at 2026-03-29T03:00:00+02:00
  fires at 2026-03-30T02:30:00+02:00
'30 2 * * *' in Europe/Zurich, from 2026-10-24T12:00:00+02:00
  fires at 2026-10-25T02:30:00+02:00
  fires at 2026-10-25T02:30:00+01:00
  fires at 2026-10-26T02:30:00+01:00
rejected: Value error, Invalid cron expression: '0 9 * *' (expected 5 fields 'minute hour day month weekday', or 6 with a trailing seconds field)
rejected: Value error, Unknown time zone 'Europe/Zurch' (use an IANA name such as 'Europe/Zurich'; on Windows install the 'tzdata' package)

Read the output closely:

• 0 9 * * 1-5 stays at 09:00 local time** across the March change. The offset in scheduled_time moves from +01:00 to +02:00. In UTC the run moves an hour earlier.

• The six-field form fires on second 0, 20 and 40 of every minute.

• A 02:30 job on the spring-forward night runs at 03:00, because 02:30 doesn't exist that night.

• A 02:30 job on the fall-back night runs twice, at 02:30 summer time and 02:30 winter time, an hour apart. A nightly job that must run once should not be scheduled between 02:00 and 03:00 local time, or should run in UTC.

• Bad expressions and misspelt time zones fail when the config is created, so a broken schedule never reaches production.

Without cron_timezone the expression is read in UTC, not in the server's local time. 0 9 * * * on a server in Zurich runs at 10:00 or 11:00 local time, depending on the season.

A scheduled job will sometimes run twice for the same slot: a retry after an error, a catch-up run after a restart, or a test. Make the tool that has the side effect idempotent. Here the digest's file name comes from the slot, which is the tick's scheduled_time, so a second publish for the same slot does nothing:

"""Tools for the digest agent. publish_digest is idempotent: the slot names the file.
SLOW_SECONDS slows list_open_tickets down, to make runs longer than the interval."""
import asyncio
import os
import re
import sys
from pathlib import Path

from promptise.mcp.server import MCPServer

DIGESTS = Path(os.environ.get("DIGEST_DIR", "digests"))
SLOW_SECONDS = float(os.environ.get("SLOW_SECONDS", "0"))
server = MCPServer("tickets")

TICKETS = [
    {"id": 4812, "priority": "high", "title": "Checkout fails for EU cards", "age_hours": 5},
    {"id": 4809, "priority": "high", "title": "Password reset email not sent", "age_hours": 19},
    {"id": 4801, "priority": "normal", "title": "CSV export has wrong date format", "age_hours": 40},
    {"id": 4797, "priority": "low", "title": "Typo on pricing page", "age_hours": 52},
]


def slot_file(slot: str) -> Path:
    return DIGESTS / (re.sub(r"[^0-9A-Za-z]+", "-", slot).strip("-") + ".md")


@server.tool(read_only_hint=True)
async def list_open_tickets() -> list[dict]:
    """List the open support tickets."""
    if SLOW_SECONDS:
        await asyncio.sleep(SLOW_SECONDS)
    return TICKETS


@server.tool()
async def publish_digest(slot: str, summary: str) -> dict:
    """Publish the digest for one schedule slot. Pass the payload's scheduled_time as slot."""
    path = slot_file(slot)
    if path.exists():
        print(f"publish_digest {slot}: already published", file=sys.stderr, flush=True)
        return {"status": "already_published", "slot": slot,
                "next_step": "This slot's digest exists. Do not publish it again; just stop."}
    DIGESTS.mkdir(parents=True, exist_ok=True)
    path.write_text(summary + "\n")
    print(f"publish_digest {slot}: published", file=sys.stderr, flush=True)
    return {"status": "published", "slot": slot, "file": str(path)}


if __name__ == "__main__":
    server.run(transport="stdio")

The already_published result also tells the model what to do next ("just stop"), which keeps a duplicate run from trying again. The same idea works for any side effect: an upsert keyed by slot, an email with a deduplication key, a ticket comment that checks for its own marker first.

The process holds the trigger, the agent and the delivery settings. This is the core of digest_agent.py, slightly shortened (the full file is in the complete example):

config = ProcessConfig(
    model="openai:gpt-5-mini",
    instructions=INSTRUCTIONS,
    servers={"tickets": StdioServerSpec(command=sys.executable, args=["tickets_server.py"])},
    triggers=[
        TriggerConfig(
            type="cron",
            cron_expression="0 8 * * 1-5",     # 08:00, Monday to Friday
            cron_timezone="Europe/Zurich",     # read in local time, DST-aware
            filter_expression=skip,            # step 5: drop ticks while one is waiting
        ),
    ],
    concurrency=1,                                          # one run at a time
    context=ContextConfig(conversation_history=False),      # step 7: every run starts fresh
    trigger_delivery=TriggerDeliveryConfig(max_retries=2),  # safe: publishing is idempotent
    max_invocation_time=600,                                # a hung run fails after 10 minutes
)
process = AgentProcess(name="ticket-digest", config=config)
await process.start()

The instructions tell the agent to pass the payload's scheduled_time as the slot. max_retries=2 re-runs a run that raised, with a 2-second backoff that doubles. That is only safe because step 2 made publishing idempotent: a retry re-runs the whole agent turn, including tools that already ran. After the retries, the event goes to process.dead_letters.

You don't need to wait until 08:00 to test a scheduled agent. Start the process with its real schedule and put a tick on the queue yourself with process.inject(). The event below has the same payload the cron trigger produces:

"""Test the scheduled agent without waiting for 08:00: inject the tick yourself."""
import asyncio
import shutil

from digest_agent import HERE, build_process
from promptise.runtime import TriggerEvent
from tickets_server import slot_file

SLOT = "2026-10-19T08:00:00+02:00"


def tick(slot: str) -> TriggerEvent:
    # The same payload CronTrigger produces.
    return TriggerEvent(trigger_id="cron-test", trigger_type="cron",
                        payload={"scheduled_time": slot, "cron_expression": "0 8 * * 1-5",
                                 "timezone": "Europe/Zurich"})


async def wait_for_runs(process, n: int, timeout: float = 120) -> None:
    async with asyncio.timeout(timeout):
        while process.status()["invocation_count"] < n:
            await asyncio.sleep(0.2)


async def main() -> None:
    shutil.rmtree(HERE / "digests", ignore_errors=True)
    process = build_process()  # the real schedule; it won't fire during the test
    await process.start()
    try:
        await process.inject(tick(SLOT))
        await wait_for_runs(process, 1)
        first = slot_file(SLOT).read_text()

        await process.inject(tick(SLOT))  # a duplicate tick for the same slot
        await wait_for_runs(process, 2)
        status = process.status()
    finally:
        await process.stop()

    assert slot_file(SLOT).read_text() == first, "the digest was published twice"
    assert len(list((HERE / "digests").iterdir())) == 1
    print("digest for", SLOT)
    print(first.strip())
    print("runs:", status["invocation_count"], "| digests:", 1,
          "| dead letters:", status["dead_letter_count"],
          "| conversation messages kept:", status["conversation_messages"])
    print("PASS")


if __name__ == "__main__":
    asyncio.run(main())

publish_digest 2026-10-19T08:00:00+02:00: published
publish_digest 2026-10-19T08:00:00+02:00: already published
digest for 2026-10-19T08:00:00+02:00
4812: Checkout fails for EU cards (high)
4809: Password reset email not sent (high)
4801: CSV export has wrong date format (normal)
4797: Typo on pricing page (low)
runs: 2 | digests: 1 | dead letters: 0 | conversation messages kept: 0
PASS

The first two lines come from the tool server. The first tick published the digest. The second tick for the same slot reached publish_digest, got already_published, and stopped, so one file exists after two runs. The test covers the real agent, tools and instructions. Only the clock is replaced.

Two details make this work. inject() skips the trigger's filter and goes straight to the queue, so the test is not affected by step 5. And the real schedule, 08:00 on weekdays, won't fire during a test that takes a minute. For a test that must never see a real tick, use an expression that can't match soon, such as 0 0 1 1 *.

Schedule a tick every 5 seconds and make each run take about 20, with a slow tool. Without a filter, overlap_demo.py prints this:

  0.0s  runs finished 0  ticks waiting 0  ticks skipped 0
 10.1s  runs finished 0  ticks waiting 1  ticks skipped 0
 15.1s  runs finished 0  ticks waiting 2  ticks skipped 0
…
 25.0s  runs finished 0  ticks waiting 4  ticks skipped 0
 26.8s  runs finished 1  ticks waiting 3  ticks skipped 0
…
 40.1s  runs finished 1  ticks waiting 6  ticks skipped 0
 44.9s  runs finished 1  ticks waiting 7  ticks skipped 0
after stop: ticks still queued: 7

With concurrency=1 the runs never overlap, but the backlog grows by one tick every 5 seconds, and every queued tick runs later, back to back. On a real schedule, such as every minute with three-minute runs, the queue keeps growing until it holds 1,000 events. Ticks still queued when the process stops are never run, and they are gone when the program exits.

There are three ways to handle it, similar to the choices a Kubernetes CronJob makes with concurrencyPolicy:

The skip policy is a callable filter. A filter sees each tick before it is queued, and returning False drops it:

class SkipWhileBacklogged:
    """Cron filter: drop a tick while an earlier tick is still waiting to run."""
    def __call__(self, event: TriggerEvent) -> bool:
        return self.process is None or self.process.status()["queue_size"] == 0

build_process() in the complete example sets skip.process once the process exists.

The same demo with the filter:

  0.0s  runs finished 0  ticks waiting 0  ticks skipped 0
  6.1s  runs finished 0  ticks waiting 1  ticks skipped 0
 11.1s  runs finished 0  ticks waiting 1  ticks skipped 1
 16.0s  runs finished 0  ticks waiting 1  ticks skipped 2
 21.0s  runs finished 1  ticks waiting 0  ticks skipped 3
…
 41.0s  runs finished 1  ticks waiting 1  ticks skipped 6
 41.2s  runs finished 2  ticks waiting 0  ticks skipped 6
after stop: ticks still queued: 0

At most one tick waits while a run is in progress, and the others are counted in filtered_count and cost nothing. Nothing is left queued at stop. A skipped tick is a skipped slot, so use this policy only when the next run covers the work of the skipped one.

Also bound each run: max_invocation_time fails a run that hangs, and a budget caps tool calls and cost per run and per day.

Ticks can be missed in three ways, and the trigger handles each one differently.

The machine sleeps or the process freezes. pause_demo.sh freezes a Python process for 17 seconds with SIGSTOP, as a laptop lid or a paused container would, and prints each tick of a 5-second schedule:

tick due 17:47:50  fired   0.0s late
tick due 17:47:55  fired   0.0s late
--- process frozen at 17:47:58
tick due 17:48:00  fired  15.3s late
--- process resumed at 17:48:15
tick due 17:48:20  fired   0.0s late
tick due 17:48:25  fired   0.0s late

The next tick, due two seconds after the freeze, fired once, 15 seconds late, as soon as the process continued. The three ticks due during the freeze never fired. On a real machine sleep the delay may be longer, because asyncio's timer runs on a monotonic clock that doesn't advance while the system sleeps on Linux and macOS.

The process is stopped. After start(), the next tick is computed from the current time. Nothing is caught up.

The process is suspended. process.suspend() keeps the trigger running, and ticks wait in the queue. After resume() they all run, back to back. idle_timeout is worse: the process suspends itself after that many idle seconds, nothing resumes it, and ticks pile up without running. Don't set idle_timeout on a scheduled agent. The pause and resume guide covers the lifecycle.

If a missed slot must still run, add a catch-up step at start-up: find the most recent slot, and if it is recent and has no result yet, inject it once. croniter's get_prev() gives the slot. The core of catch_up() in the complete example:

now = datetime.now(ZoneInfo(TIMEZONE))
due = croniter(schedule, now).get_prev(datetime)   # the most recent slot
if now - due > max_age or slot_file(due.isoformat()).exists():
    return None  # too old to be useful, or already done
await process.inject(TriggerEvent(trigger_id="catch-up", trigger_type="cron", payload={
    "scheduled_time": due.isoformat(), "cron_expression": schedule, "timezone": TIMEZONE}))

catch_up_demo.sh starts the agent twice on an hourly schedule:

--- start 1 at 17:53:23 UTC
digest agent running: 0 * * * * Europe/Zurich
catching up on 2026-10-19T19:00:00+02:00
publish_digest 2026-10-19T19:00:00+02:00: published
stopped: 1 runs, 0 ticks skipped, 0 dead letters
--- start 2 at 17:53:53 UTC
digest agent running: 0 * * * * Europe/Zurich
stopped: 0 runs, 0 ticks skipped, 0 dead letters
2026-10-19T19-00-00-02-00.md

The first start found the 19:00 slot without a digest and ran it. The second start found the digest and did nothing. The idempotent tool from step 2 is what makes this safe: if the catch-up and a real tick ever run the same slot, the second one stops. max_age is the equivalent of a CronJob's startingDeadlineSeconds: a digest from yesterday morning isn't worth posting at 17:00.

A process keeps a short-term conversation buffer: each run's trigger message and final reply are added to it, and the next run sees them. That is the default (conversation_history=True, up to conversation_max_messages=100). history_demo.py runs three ticks each way:

conversation_history=True: messages kept after each run: [2, 4, 6]
conversation_history=False: messages kept after each run: [0, 0, 0]

With history, every run sends the earlier digests to the model again. That costs tokens on every run, and the model may copy yesterday's answer instead of reading today's data. For a scheduled job, turn history off and keep anything the next run needs in a tool or in the process context, as the state between runs guide shows. Note that conversation_max_messages=0 means unlimited, not off. The buffer is in memory and is cleared when the process stops.

The project has three files: tickets_server.py from step 2, test_digest.py from step 4, and the process below. The process reads its schedule from SCHEDULE, so you can try it with a fast schedule first.

"""A scheduled AI agent: a support-ticket digest at 08:00 Zurich time on weekdays."""
import asyncio
import os
import signal
import sys
from datetime import datetime, timedelta
from pathlib import Path
from zoneinfo import ZoneInfo

from croniter import croniter

from promptise.config import StdioServerSpec
from promptise.runtime import (
    AgentProcess,
    ContextConfig,
    ProcessConfig,
    TriggerConfig,
    TriggerDeliveryConfig,
    TriggerEvent,
)
from tickets_server import slot_file

HERE = Path(__file__).parent
SCHEDULE = os.environ.get("SCHEDULE", "0 8 * * 1-5")  # 08:00 on weekdays
TIMEZONE = "Europe/Zurich"

INSTRUCTIONS = """\
You write the morning digest of open support tickets.
Each run starts with a [Trigger: cron] message whose payload has a scheduled_time.
1. Call list_open_tickets once.
2. Write a digest of at most four lines: high-priority tickets first, with their ids.
3. Call publish_digest once, with slot set to the payload's scheduled_time exactly.
If publish_digest says the slot is already published, stop. Reply with one short line.
"""


class SkipWhileBacklogged:
    """Cron filter: drop a tick while an earlier tick is still waiting to run."""

    def __init__(self) -> None:
        self.process: AgentProcess | None = None

    def __call__(self, event: TriggerEvent) -> bool:
        return self.process is None or self.process.status()["queue_size"] == 0


def build_process(schedule: str = SCHEDULE, *, slow_seconds: float = 0,
                  coalesce: bool = True, history: bool = False) -> AgentProcess:
    skip = SkipWhileBacklogged()
    config = ProcessConfig(
        model="openai:gpt-5-mini",
        instructions=INSTRUCTIONS,
        servers={
            "tickets": StdioServerSpec(
                command=sys.executable,
                args=[str(HERE / "tickets_server.py")],
                env={"DIGEST_DIR": str(HERE / "digests"), "SLOW_SECONDS": str(slow_seconds)},
            ),
        },
        triggers=[
            TriggerConfig(
                type="cron",
                cron_expression=schedule,
                cron_timezone=TIMEZONE,
                filter_expression=skip if coalesce else None,
            ),
        ],
        concurrency=1,  # one run at a time; later ticks wait in the queue
        context=ContextConfig(conversation_history=history),  # False: every run starts fresh
        trigger_delivery=TriggerDeliveryConfig(max_retries=2),  # safe: publishing is idempotent
        max_invocation_time=600,  # a run that hangs fails after 10 minutes
    )
    process = AgentProcess(name="ticket-digest", config=config)
    skip.process = process
    return process


async def catch_up(process: AgentProcess, schedule: str = SCHEDULE,
                   max_age: timedelta = timedelta(hours=12)) -> TriggerEvent | None:
    """Run the most recent missed slot once, if it is recent and has no digest yet."""
    now = datetime.now(ZoneInfo(TIMEZONE))
    due = croniter(schedule, now).get_prev(datetime)
    if now - due > max_age or slot_file(due.isoformat()).exists():
        return None
    event = TriggerEvent(
        trigger_id="catch-up",
        trigger_type="cron",
        payload={"scheduled_time": due.isoformat(), "cron_expression": schedule,
                 "timezone": TIMEZONE, "catch_up": True},
    )
    await process.inject(event)
    return event


async def main() -> None:
    process = build_process()
    await process.start()
    missed = await catch_up(process)
    print("digest agent running:", SCHEDULE, TIMEZONE, flush=True)
    if missed:
        print("catching up on", missed.payload["scheduled_time"], flush=True)

    stop = asyncio.Event()
    for sig in (signal.SIGINT, signal.SIGTERM):
        asyncio.get_running_loop().add_signal_handler(sig, stop.set)
    await stop.wait()

    await process.stop()
    s = process.status()
    print(f"stopped: {s['invocation_count']} runs, {s['filtered_count']} ticks skipped, "
          f"{s['dead_letter_count']} dead letters", flush=True)


if __name__ == "__main__":
    asyncio.run(main())

pip install "promptise>=1.4.3"
export OPENAI_API_KEY=sk-...
python test_digest.py                          # one slot, run twice, published once
SCHEDULE="* * * * * */20" python digest_agent.py   # a tick every 20 seconds
python digest_agent.py                         # the real schedule: 08:00 Zurich time on weekdays

Run exactly one scheduler per schedule

Two replicas of the same process each fire every tick, so the job runs twice. The cron trigger has no leader election, and Promptise doesn't coordinate schedules across machines. Run the scheduled process as a single replica, for example one container with restart: unless-stopped, and keep the idempotent tools from step 2 for the rare duplicate. If you scale the rest of your agents, keep the scheduled ones in their own process, as running multiple agents explains.

Stop cleanly

process.stop() cancels a run in progress, and ticks still in the queue are never run. Handle SIGTERM, as the complete example does, and give the container enough stop time for a run to finish, or rely on the catch-up step to run the slot again. For runs that must survive a crash mid-run, see durable agents and crash recovery.

Watch the counters

process.status() reports invocation_count, queue_size, filtered_count, dead_letter_count and pending_retries. Alert when the queue keeps growing (runs are slower than the schedule) or when dead letters appear (runs failed after their retries). Health monitoring catches runs that succeed but do nothing useful, such as an empty digest every morning.

Keep the clock right

The trigger computes ticks from the system clock. Run NTP on the host, and set cron_timezone explicitly instead of relying on the server's time zone. On Windows, install the tzdata package so IANA names such as Europe/Zurich resolve.

Deploy it as a long-running service

A scheduled agent is a service, not a script. Deploying AI agents in production walks through the first long-running agent, and deploying an AI agent with Docker packages it.

• No catch-up and no memory of past ticks. The trigger starts fresh on every start. The catch-up step in this guide covers the most recent slot only.

• No coordination across processes. Two processes with the same schedule both run it. Use one replica.

• No jitter. Every process with 0 * * * * fires at the same second, which can spike load on a shared API. Offset schedules by hand (7 * * * *).

• Overlap control is up to you. The runtime queues ticks. Skipping, as in step 5, is a filter you write.

• The trigger ID changes on every start (cron- plus random characters), so don't key anything on it. Use scheduled_time.

• The fallback without `croniter` handles only */N * * * *, * * * * * and a single minute. croniter ships with Promptise, so this only matters if it was removed.

How do I run an AI agent every day at a specific time?

Use a cron expression with the minute and hour, and set the time zone: TriggerConfig(type="cron", cron_expression="0 8 * * *", cron_timezone="Europe/Zurich") runs at 08:00 Zurich time every day, in summer and winter. Without cron_timezone, the expression is read in UTC.

Can a cron-triggered AI agent run every few seconds?

Yes. Add a sixth field for seconds: * * * * * */10 fires every 10 seconds. Each tick is a full agent run with LLM calls, so a short interval gets expensive fast and runs may overlap. Use a short interval for testing, or for polling with a filter that drops ticks when there is nothing to do.

What happens if a scheduled agent run takes longer than the interval?

The trigger keeps firing, and the extra ticks wait in the queue. With concurrency=1 they run one after another, so a slow agent falls further behind on every tick. Either drop ticks while one is already waiting (a callable filter that checks process.status()["queue_size"]), allow parallel runs with concurrency, or make the interval longer than the slowest run.

Does the agent catch up on runs it missed while it was down?

No. When the process starts, the next tick is computed from the current time, and ticks that fell in the downtime never fire. If a missed slot must still run, check the most recent slot at start-up with croniter(...).get_prev() and inject it with process.inject() when it has no result yet.

How do I test a scheduled AI agent without waiting?

Start the process and inject a TriggerEvent with trigger_type="cron" and the payload the trigger would produce, including scheduled_time. Then wait until process.status()["invocation_count"] goes up and check the result. The test runs the real agent and tools, and only the clock is replaced.

Should a scheduled agent keep conversation history between runs?

Usually not. With history on, every run resends earlier runs to the model, which costs tokens and tempts the model to repeat old answers. Set ContextConfig(conversation_history=False) and keep state the next run needs in a tool, a file or the process context.

• Compare all trigger types and the payload format: event-driven AI agents.

• Start the agent from an HTTP event instead: AI agent webhook trigger.

• Process files as they arrive: AI agent file watcher.

• Cap what each scheduled run may spend: AI agent cost control.

• Learn the runtime around the trigger: the autonomous AI agent runtime.

• Reference: cron trigger and triggers overview in the Promptise docs.