Async Agent Scheduler

Run agents on a schedule with async-native execution, cancellation, and retries

Run an agent on a recurring schedule with async-native execution, cooperative cancellation, and built-in retries.

from praisonaiagents import Agent

agent = Agent(
    name="NewsChecker",
    instructions="Summarise today's AI news in three bullet points.",
)
# Schedule via AsyncAgentScheduler (see Quick Start)

The user sets a recurring schedule; the scheduler runs the agent asynchronously without blocking the event loop.

Quick Start

1. Simple Usage

   import asyncio
   from praisonaiagents import Agent
   from praisonai.scheduler import AsyncAgentScheduler
   
   async def main():
       agent = Agent(
           name="NewsChecker",
           instructions="Summarise today's AI news in 3 bullet points.",
       )
   
       scheduler = AsyncAgentScheduler(agent, task="Check the latest AI news")
       await scheduler.start("hourly", max_retries=3, run_immediately=True)
   
       # ... run your app ...
   
       await scheduler.stop()
       print(await scheduler.get_stats())
   
   asyncio.run(main())

2. With Callbacks

   import asyncio
   from praisonaiagents import Agent
   from praisonai.scheduler import AsyncAgentScheduler
   
   def on_success(result):
       print(f"Agent completed successfully: {result}")
   
   def on_failure(error):
       print(f"Agent failed: {error}")
   
   async def main():
       agent = Agent(
           name="DataProcessor",
           instructions="Process incoming data efficiently",
       )
   
       scheduler = AsyncAgentScheduler(
           agent, 
           task="Process latest batch of data",
           on_success=on_success,
           on_failure=on_failure
       )
   
       # Run every 30 minutes
       await scheduler.start("*/30m", max_retries=3, run_immediately=True)
   
       # Keep running
       try:
           await asyncio.sleep(3600)  # Run for 1 hour
       finally:
           await scheduler.stop()
           stats = await scheduler.get_stats_async()
           print(f"Completed {stats['successful_executions']} successful executions")
   
   asyncio.run(main())

3. With Timeout & Budget Limit

   import asyncio
   from praisonaiagents import Agent
   from praisonai.scheduler import AsyncAgentScheduler
   
   async def main():
       agent = Agent(
           name="CostAwareAgent",
           instructions="Summarise the latest tech headlines.",
       )
   
       scheduler = AsyncAgentScheduler(
           agent,
           task="Summarise tech news",
           timeout=30,        # stop a single run after 30s
           max_cost=1.00,     # auto-shutdown once $1.00 spent
       )
   
       await scheduler.start("hourly", run_immediately=True)
       await asyncio.sleep(3600 * 4)
       stats = await scheduler.get_stats_async()
       print(f"Spent ${stats['total_cost_usd']}, remaining ${stats['remaining_budget']}")
       await scheduler.stop()
   
   asyncio.run(main())

4. Deliver Results to a Chat Channel

   import asyncio
   from praisonaiagents import Agent
   from praisonai.scheduler import AsyncAgentScheduler
   
   async def main():
       agent = Agent(name="Briefer", instructions="Summarise today's AI news in 5 bullets.")
       scheduler = AsyncAgentScheduler(agent, task="Morning brief", deliver="telegram:123456")
       await scheduler.start("hourly")
   
   asyncio.run(main())

   AsyncAgentScheduler accepts the same deliver= parameter as the sync scheduler and dispatches delivery through asyncio.to_thread — the shared helper uses the sync bridge, so delivery never blocks the event loop and never raises. See Scheduler → Deliver Results for the full token grammar and the praisonai[bot] optional dependency.

How It Works

The AsyncAgentScheduler uses async-native execution with cooperative cancellation, replacing the old thread-based scheduler.

Schedule Expression Reference

Legacy forms above are unchanged — every one keeps its existing behaviour.

Timezones

A naive at: timestamp (no Z, no +HH:MM offset) and every cron: expression resolve their timezone in this order:

1. The tz field on the schedule (per-schedule).
2. The default_timezone argument on the scheduler / agent (per-process default set in code).
3. The PRAISONAI_SCHEDULE_TIMEZONE environment variable (process-wide default).
4. UTC (fallback when none of the above is set).

from praisonaiagents.tools import schedule_add

# Naive — interpreted in the schedule tz (America/New_York here)
schedule_add(
    name="report-ny",
    schedule="at:2026-03-01T09:00:00",
    message="Send morning report",
    tz="America/New_York",
)

# Offset in the string wins — 09:00 India time everywhere
schedule_add(
    name="report-ist",
    schedule="at:2026-03-01T09:00:00+05:30",
    message="Send morning report",
)

See Schedule Tools → Timezones for the full ladder and the PRAISONAI_SCHEDULE_TIMEZONE example.

Natural language

Plain English clock times and day names parse into the same one-shot and cron schedules — no prefixes required.

Natural-language expressions. A bare clock time (at 9am) is treated as a one-shot at the next matching local time in the schedule’s timezone. Any recurring form — every …, daily …, or one that names a day of the week — is translated to a cron expression and follows the same wall-clock semantics as cron:. The croniter engine that evaluates these expressions ships with praisonaiagents by default. If it is somehow missing (a stripped install), parse_schedule() raises ValueError at creation time with an install hint — no more silent “accepted but never fires”.

Day-of-week names accept full or abbreviated forms (monday/mon, tuesday/tue/tues, thursday/thu/thurs, …), plus weekday(s) → 1-5 and weekend(s) → 0,6. A bare clock time is timezone-aware: at 9am denotes 09:00 local time in the schedule’s timezone, not 09:00 UTC.

from praisonaiagents.scheduler import parse_schedule

parse_schedule("at 9am")                    # one-shot: next 09:00 local
parse_schedule("at 9:30pm")                 # one-shot: next 21:30 local
parse_schedule("every day at 9am")          # cron: "0 9 * * *"
parse_schedule("daily at 9am")              # cron: "0 9 * * *"
parse_schedule("weekdays at 9am")           # cron: "0 9 * * 1-5"
parse_schedule("weekends at 10:30am")       # cron: "30 10 * * 0,6"
parse_schedule("every monday 9am")          # cron: "0 9 * * 1"
parse_schedule("every mon,wed,fri at 9am")  # cron: "0 9 * * 1,3,5"

# Timezone-aware one-shot: 09:00 New York, not 09:00 UTC
schedule = parse_schedule("at 9am", tz="America/New_York")

Add the same forms from the CLI:

praisonai schedule add "standup" -s "weekdays at 9am" -m "post the standup summary"
praisonai schedule add "reminder" -s "at 5pm"          -m "wrap up the day" --once

Timezones

A naive at: timestamp and the clock forms (at 9am, at 17:00) resolve their zone in order: the schedule tz → the instance default → PRAISONAI_SCHEDULE_TIMEZONE → the machine’s local zone. An at: string that already carries an offset (+05:30, Z) is used verbatim. cron: follows the same order but falls back to UTC, not local.

• Stored value is aware. As of PraisonAI #4732, parse_schedule stamps a naive at: with the resolved zone at parse time, so the stored ISO string names an unambiguous instant and fires at the same wall-clock time on any runner.
• DST is per target date. The offset attached is the one in force on the target date. In Europe/London, at:2026-07-01T09:00:00 stamps +01:00 and at:2026-12-01T09:00:00 stamps +00:00, whichever day you parse it.
• Bad zone names fail loudly. An unknown IANA zone in PRAISONAI_SCHEDULE_TIMEZONE (or tz=) now raises ValueError at parse time instead of being accepted and never firing.

from praisonaiagents.scheduler.parser import parse_schedule

parse_schedule("at:2026-07-01T09:00:00", tz="Europe/London").at  # -> "...T09:00:00+01:00"
parse_schedule("at:2026-12-01T09:00:00", tz="Europe/London").at  # -> "...T09:00:00+00:00"

See Schedule Tools → Timezones for the full precedence ladder and the legacy-job migration story.

Wall-clock cron

cron: schedules honour the real time-of-day, not the interval since process start.

• Time-of-day accuracy — cron:0 9 fires at 09:00 wall-clock every day. A restart at 09:30 does not* re-phase the schedule to 09:30.
• Downtime catch-up (exactly once) — if the process was down when a slot was due (e.g. a scale-to-zero window), the schedule catches up once on resume, then re-anchors to the next future occurrence. Never double-fires the same slot.
• State file — the daemon persists last_run_at to ~/.praisonai/schedulers/<name>.json so restarts pick up the schedule exactly where they left off.
• Bundled engine — the croniter engine that powers wall-clock cron ships with praisonaiagents by default, so cron: schedules work out of the box with no extra install step.

Plain interval expressions (hourly, */30m, raw seconds) are unchanged — they keep the fixed-interval sleep behaviour.

Introduced in PraisonAI PR #3527.

Configuration Options

AsyncAgentScheduler Constructor

start() Method Options

start() returns a bool: True when the background task is running, False when it never starts — already running, a parse error, or an immediate run that tripped the budget brake (see the max_cost note above).

Reading Stats

graph LR
    A[Async Caller] --> B[get_stats() / get_stats_async()]
    B --> C[_stats_lock]
    C --> D[Atomic Snapshot]

    E[Sync Caller] --> F[get_stats() / get_stats_sync()]
    F --> I[_delivery_lock]
    I --> J[Delivery pair snapshotted;<br/>other counters best-effort]

    classDef async fill:#10B981,stroke:#7C90A0,color:#fff
    classDef sync fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef atomic fill:#6366F1,stroke:#7C90A0,color:#fff

    class A,B async
    class E,F sync
    class C,D,I,J atomic

Statistics can be read in both sync and async contexts with different guarantees:

Stats Response Format

Both counters are explicit: SUPPRESSED and NOT_CONFIGURED runs contribute zero to each, so delivered_deliveries + undelivered_deliveries ≤ successful_executions. See the per-outcome counter table for which cell each run lands in.

Since PR #1857 the shared stats builder landed on _BaseAgentScheduler; since PR #5322 AsyncAgentScheduler.get_stats_sync() inherits it, so the sync and async surfaces return the same dict shape (including runtime_seconds and cost_per_execution) and the delivery-counter pair is read under _delivery_lock. Before #5322 the async get_stats_sync() overrode the base and omitted those two keys.

Async Blueprints

AsyncAgentScheduler.from_blueprint(...) builds a scheduler from a named blueprint template, dispatching the blueprint prompt via astart() so scheduled runs never block the event loop.

Blueprint resolution lives in the shared _base_scheduler.build_from_blueprint, so the async and sync surfaces stay in lock-step. from_blueprint also populates _yaml_schedule_config, so start_from_yaml_config() starts the scheduler on the blueprint’s resolved interval.

1. Build from a blueprint

   import asyncio
   from praisonai.scheduler import AsyncAgentScheduler
   
   async def main():
       # Async twin of AgentScheduler.from_blueprint — the blueprint prompt is
       # dispatched via astart(), so scheduled runs never block the event loop.
       scheduler = AsyncAgentScheduler.from_blueprint(
           "morning-brief",
           slots={"topic": "AI"},
       )
       await scheduler.start_from_yaml_config()
   
       await asyncio.sleep(3600)
       await scheduler.stop()
   
   asyncio.run(main())

2. Override interval and delivery

   import asyncio
   from praisonai.scheduler import AsyncAgentScheduler
   
   async def main():
       scheduler = AsyncAgentScheduler.from_blueprint(
           "morning-brief",
           slots={"topic": "AI"},
           deliver="telegram:123456",   # overrides the blueprint default
           interval_override="*/30m",   # overrides the resolved schedule
       )
       await scheduler.start_from_yaml_config()
   
       await asyncio.sleep(3600)
       await scheduler.stop()
   
   asyncio.run(main())

from_blueprint() Parameters

from_blueprint raises ValueError if the blueprint is not found or a required slot is missing.

Common Patterns

Running in FastAPI Application

from contextlib import asynccontextmanager
from fastapi import FastAPI
from praisonaiagents import Agent
from praisonai.scheduler import AsyncAgentScheduler

scheduler = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    global scheduler
    agent = Agent(name="BackgroundWorker", instructions="Process background tasks")
    scheduler = AsyncAgentScheduler(agent, task="Process pending tasks")
    await scheduler.start("*/5m", max_retries=2)

    yield

    # Shutdown
    if scheduler:
        await scheduler.stop()

app = FastAPI(lifespan=lifespan)

Error Handling with Logging

import asyncio
import logging
from praisonaiagents import Agent
from praisonai.scheduler import AsyncAgentScheduler

logging.basicConfig(level=logging.INFO)

def handle_failure(error):
    # Fires on EITHER an execution error OR a delivery failure. When it's a
    # delivery failure the reason string is "scheduled result could not be
    # delivered" (see PR #4476) — match on it to distinguish the two.
    logging.error(f"Agent execution failed: {error}")
    # Send alert, write to database, etc.

async def main():
    agent = Agent(name="MonitoringAgent", instructions="Monitor system health")
    scheduler = AsyncAgentScheduler(
        agent,
        task="Check system status",
        # on_failure now fires on EITHER an execution error OR a delivery failure
        # (reason is "scheduled result could not be delivered" for the latter).
        on_failure=handle_failure
    )

    await scheduler.start("*/10m")

    try:
        await asyncio.sleep(float('inf'))
    except KeyboardInterrupt:
        await scheduler.stop()

asyncio.run(main())

Graceful Shutdown on SIGINT

import asyncio
import signal
from praisonaiagents import Agent
from praisonai.scheduler import AsyncAgentScheduler

scheduler = None

def signal_handler():
    if scheduler:
        asyncio.create_task(scheduler.stop())

async def main():
    global scheduler

    # Setup signal handling
    for sig in [signal.SIGINT, signal.SIGTERM]:
        signal.signal(sig, lambda s, f: signal_handler())

    agent = Agent(name="LongRunningAgent", instructions="Process data continuously")
    scheduler = AsyncAgentScheduler(agent, task="Process data batch")

    await scheduler.start("*/15m", run_immediately=True)

    try:
        # Keep running until signal
        await asyncio.sleep(float('inf'))
    except KeyboardInterrupt:
        print("Received interrupt, shutting down gracefully...")
    finally:
        if scheduler:
            await scheduler.stop()
            stats = await scheduler.get_stats_async()
            print(f"Final stats: {stats}")

asyncio.run(main())

Budget-aware Scheduling

import asyncio
from praisonaiagents import Agent
from praisonai.scheduler import AsyncAgentScheduler

async def main():
    agent = Agent(name="ReportAgent", instructions="Generate hourly report")

    # Hard-stop after $5 total spend, 60s per run
    scheduler = AsyncAgentScheduler(
        agent,
        task="Generate report",
        timeout=60,
        max_cost=5.00,
    )

    await scheduler.start("hourly")
    while scheduler.is_running:
        await asyncio.sleep(60)
        stats = await scheduler.get_stats_async()
        if stats["remaining_budget"] is not None and stats["remaining_budget"] < 0.50:
            print(f"⚠️  Budget nearly exhausted: ${stats['remaining_budget']:.4f} left")
    print("Scheduler stopped (budget reached or manual stop)")

asyncio.run(main())

Manual one-shot runs

execute_once() runs the agent once outside the schedule loop and honours max_cost like the scheduled loop does.

scheduler = AsyncAgentScheduler(agent, task, max_cost=1.00)
scheduler._total_cost = 1.00    # simulate an exhausted budget
assert await scheduler.execute_once() is None   # no LLM call, no cost added

When the running total has already reached the cap, the call logs a warning and returns None without hitting the LLM. Before PraisonAI PR #5766, manual execute_once() could bill one extra call past the cap.

Daemon state persistence

When AsyncAgentScheduler runs as a daemon (started via praisonai schedule start <name> ...), it now updates ~/.praisonai/schedulers/<name>.json after every execution with the current executions count and cost. The file I/O is offloaded with asyncio.to_thread() so the event loop is never blocked. Previously these writes were a TODO and async daemons silently reported stale numbers. See PR #1857.

Best Practices

# Good
try:
    await scheduler.start("hourly")
    await asyncio.sleep(3600)
finally:
    await scheduler.stop()

# Bad - may interrupt agent mid-execution
await scheduler.start("hourly")
await asyncio.sleep(3600)
# Exit without stopping

# Test immediately, then schedule
await scheduler.start("hourly", run_immediately=True)

# Good for smoke tests
await scheduler.start("daily", run_immediately=True)

# Good - lightweight logging
def on_success(result):
    logger.info(f"Agent completed: {result}")

# Bad - heavy database operations
def on_success(result):
    database.save_large_dataset(result)  # Blocks scheduler

# Better - offload heavy work
def on_success(result):
    asyncio.create_task(save_to_database(result))

# New async approach
from praisonai.scheduler import AsyncAgentScheduler
scheduler = AsyncAgentScheduler(agent, task)
await scheduler.start("hourly")

# Legacy thread-based (avoid for new code)
from praisonai.scheduler import AgentScheduler
scheduler = AgentScheduler(agent, task)
scheduler.start("hourly")

# Good — async app
from praisonai.scheduler import AsyncAgentScheduler
scheduler = AsyncAgentScheduler.from_blueprint("morning-brief", slots={"topic": "AI"})

# Avoid inside a running event loop
from praisonai.scheduler import AgentScheduler
scheduler = AgentScheduler.from_blueprint("morning-brief", slots={"topic": "AI"})

# Production with explicit budget
scheduler = AsyncAgentScheduler(agent, task, max_cost=50.00)

# Disable budget (only when external limits enforce it)
scheduler = AsyncAgentScheduler(agent, task, max_cost=None)

# Dry-run / safety-drill: schedule wired up but nothing will execute
scheduler = AsyncAgentScheduler(agent, task, max_cost=0)
await scheduler.start("hourly")
assert scheduler.is_running is False   # trips on the first tick

scheduler = AsyncAgentScheduler(
    agent,
    task,
    timeout=30,        # individual run cannot exceed 30s
    max_cost=1.00,
)

from praisonai.scheduler import AsyncAgentScheduler

# Fires at 09:00 wall-clock every day — croniter ships with praisonaiagents
scheduler = AsyncAgentScheduler(agent, task="Morning brief")
await scheduler.start("cron:0 9 * * *")

Skipping Ticks with a Pre-Run Gate

A pre-run gate lets you run a cheap shell check before each scheduled tick — the agent only fires when the check says there’s something to do. This cuts token spend on quiet ticks (e.g. polling for new emails every 5 minutes but only summarising when mail actually arrives).

See Scheduler Pre-Run Gate for the full configuration reference and examples.

Delivering Results to a Chat

Pass deliver= to push each successful result to Telegram, Discord, Slack, or WhatsApp — no gateway required.

scheduler = AsyncAgentScheduler(agent, task="Morning brief", deliver="telegram:123456")
await scheduler.start("hourly")

See Scheduler Delivery for the token grammar (Python / YAML / CLI) and reliability guarantees.

Related