What this does

AI agents often need to run on schedules—nightly data refreshes, periodic monitoring, or event-driven triggers based on queue depth. This guide covers cron-based scheduling, webhook trigger setup, and queue-depth threshold automation using APScheduler or systemd timers.

Steps

1. Choose a scheduling approach

Approach	Best for
APScheduler (Python)	In-process scheduling, flexible intervals
system cron	Simple recurring tasks, system-level reliability
Celery Beat	Distributed queues, high-throughput pipelines

APScheduler is recommended for most agent codebases.

2. Install APScheduler

pip install apscheduler

3. Define agent tasks

from apscheduler.schedulers.blocking import BlockingScheduler

def run_data_refresh_agent():
    from agents.data_refresh import DataRefreshAgent
    agent = DataRefreshAgent()
    agent.execute()

def run_monitoring_agent():
    from agents.monitor import MonitorAgent
    agent = MonitorAgent()
    agent.health_check()

4. Configure schedules

scheduler = BlockingScheduler()

scheduler.add_job(run_data_refresh_agent, "cron", hour="*/6", id="data_refresh")

scheduler.add_job(run_monitoring_agent, "cron", day_of_week="mon", hour=8, id="monitor")

from apscheduler.triggers.interval import IntervalTrigger

def check_queue_and_trigger():
    from queue_manager import get_queue_depth
    depth = get_queue_depth()
    if depth > 50:
        run_data_refresh_agent()

scheduler.add_job(check_queue_and_trigger, IntervalTrigger(minutes=5))

scheduler.start()

5. Add a webhook trigger endpoint For event-driven execution, expose a lightweight FastAPI endpoint:

from fastapi import FastAPI
app = FastAPI()

@app.post("/trigger/{agent_name}")
def trigger(agent_name: str):
    if agent_name == "data_refresh":
        run_data_refresh_agent()
        return {"status": "triggered"}
    return {"error": "unknown agent"}, 404

Record the local run evidence. Save the exact command, runtime or package version, model name if applicable, and observed output so the result can be reproduced later.

Verification

Test the scheduler in dry-run mode:

python -m scheduler.run --dry-run

Expected output:

[data_refresh] next run: 2025-07-15 06:00:00 (cron: */6 hours)
[monitor] next run: 2025-07-21 08:00:00 (cron: mon 08:00)
[queue_check] next run: in 5 minutes (interval)

Test the webhook:

curl -X POST http://localhost:8000/trigger/data_refresh

Expected: {"status": "triggered"}

Common failures

Missed runs on system sleep: Use a persistent scheduler backend (Redis, SQLAlchemy) so missed jobs are picked up on restart.
Webhook authentication: Always add API key or JWT validation to trigger endpoints to prevent unauthorized agent execution.
Overlapping runs: Set max_instances=1 on jobs that should not run concurrently, preventing pile-up when execution time exceeds the interval.
Timezone issues: Always specify timezone="Africa/Lagos" or similar for scheduled jobs to avoid drift.

Related guides

How to build an AI content generation pipeline — Scheduling the generation pipeline nightly ensures fresh content without manual triggers.
How to build a local AI product with Nigerian naira pricing — Scheduled billing reconciliation agents can sync Paystack transactions daily.