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)

Since PraisonAI PR #1566, exceptions inside a scheduled run are caught and reported via on_failure without killing the scheduler loop. Use await scheduler.get_stats_async() for metrics from async code.

Agent-scheduler runs inherit the same at-most-once claim behavior when the underlying store supports it — a due job fires on only one worker per tick even with several processes polling the same store. See Multi-process safety.

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.

The scheduler’s async primitives (_stop_event, _cancel_event, _stats_lock) are now created lazily inside _ensure_async_primitives() and bound to the loop that start() runs on. Tests that call stop() without first calling start() must invoke scheduler._ensure_async_primitives() explicitly — see tests/unit/scheduler/test_async_agent_scheduler.py in PR #1583 for the canonical pattern.

Shared dispatch helper (PR #2147): AsyncPraisonAgentExecutor.execute() now delegates to the shared adispatch_agent helper (praisonai.scheduler._dispatch). No behaviour change — the dispatch ladder (astart → to_thread(start) → AttributeError) is unchanged. See also the sync scheduler note for the corresponding fix to AgentScheduler.

---

Schedule Expression Reference

Expression	Interval	Description
"hourly"	3600s	Every hour
"daily"	86400s	Every 24 hours
"weekly"	604800s	Every 7 days
"*/30m"	1800s	Every 30 minutes
"*/1h"	3600s	Every 1 hour
"*/5s"	5s	Every 5 seconds
"60"	60s	Custom seconds (plain digits)
"cron:0 7 * * *"	best-effort	Best-effort integer collapse for the polling loop
"at:2026-03-01T09:00:00"	one-shot	ISO 8601 timestamp

Since PraisonAI PR #2916, the wrapper delegates the shared grammar (hourly / daily / weekly / */N{m,h,s} / raw seconds) to praisonaiagents.scheduler.parser.parse_schedule, so weekly is now accepted by every wrapper scheduler (AgentScheduler, AsyncAgentScheduler, DeploymentScheduler, and the praisonai schedule CLI). The wrapper-specific cron: collapse and the bare */N (unit-less) fast-path stay local to the wrapper.

---

Configuration Options

AsyncAgentScheduler Constructor

Parameter	Type	Default	Description
agent	Any	Required	Agent instance to schedule
task	str	Required	Task description to execute
config	Optional[Dict[str, Any]]	None	Optional configuration dictionary
on_success	Optional[Callable[[Any], None]]	None	Callback function on successful execution
on_failure	Optional[Callable[[Exception], None]]	None	Callback function on failed execution
timeout	Optional[int]	None	NEW — Maximum execution time per run in seconds. None means no limit. Enforced with asyncio.wait_for().
max_cost	Optional[float]	1.00	NEW — Maximum total cost in USD. Scheduler auto-stops when reached. Set to None to disable. Default $1.00 is a safety guard.

start() Method Options

Parameter	Type	Default	Description
schedule_expr	str	Required	Schedule expression (e.g., “hourly”, ”*/1h”, “3600”)
max_retries	int	3	Maximum retry attempts on failure
run_immediately	bool	False	If True, run agent immediately before starting schedule

Reading Stats

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

Method	Sync/Async	Atomicity	When to Use
scheduler.get_stats()	sync	Best-effort, no lock — counters may be observed mid-update during concurrent execution.	Backward compatibility. Quick stats access in sync contexts.
await scheduler.get_stats_async()	async	Atomic snapshot under _stats_lock — all counters read together.	Async contexts where consistent stats are needed.
scheduler.get_stats_sync()	sync	Best-effort, no lock — same as get_stats() for clarity.	Explicit sync contexts (tests, scripts, REPL).

Stats Response Format

Field	Type	Description
is_running	bool	Whether scheduler is currently running
total_executions	int	Total number of execution attempts
successful_executions	int	Number of successful executions
failed_executions	int	Number of failed executions
success_rate	float	Success percentage (0-100)
total_cost_usd	float	Running total cost in USD (4 dp)
remaining_budget	float | None	max_cost - total_cost_usd, or None if max_cost is disabled
runtime_seconds	float	NEW (PR #1857) — Seconds since start() was called. 0 when scheduler hasn’t started.
cost_per_execution	float	NEW (PR #1857) — total_cost_usd / total_executions, rounded to 4 dp. 0 when no executions yet.

Since PR #1857, AsyncAgentScheduler shares its stats builder with the sync AgentScheduler via the internal _BaseAgentScheduler mixin — the schema is now identical across sync and async.

---

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):
    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=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())

Real cost tracking (PR #2171): Before #2171, both schedulers added a fixed $0.0001 to total_cost per run, so the default max_cost=1.00 only tripped after ~10,000 runs regardless of model. The scheduler now pulls usage (input/output tokens) and model off the agent response and prices it through praisonai.cli.features.cost_tracker.ModelPricing. Responses with no usage metadata contribute $0 — the brake errs on the side of running rather than tripping on missing data. Negative token counts are clamped to 0 so they can never bypass the brake.

---

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

Always await scheduler.stop() before exiting

The stop() method waits up to 30 seconds for the current execution to complete before canceling. This prevents data corruption and ensures clean shutdown.

# 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

Use run_immediately=True for testing

Enable run_immediately=True to verify your agent works correctly before waiting for the first scheduled interval.

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

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

Keep callbacks lightweight

Success and failure callbacks are called synchronously. Heavy operations should be offloaded to avoid blocking the scheduler.

# 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))

Prefer AsyncAgentScheduler over legacy thread-based scheduler

For new code, use AsyncAgentScheduler instead of the legacy AgentScheduler. The async version provides better cancellation, no daemon threads, and fits naturally into async applications.

# 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")

Budget Control

The default max_cost=1.00 caps unattended cost runaway. Since PR #2171, total_cost_usd is the real per-token spend computed from each response’s usage field — pick a value that matches your approved budget, not a multiple chosen to compensate for an undercount.

# 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)

When the budget triggers, the scheduler logs a warning and calls stop_event.set() internally — stats["is_running"] flips to False.

If your model is missing from DEFAULT_PRICING, the run is priced at $0 and total_cost_usd will under-report — register it via the custom pricing snippet so the brake stays meaningful.

Timeout Configuration

Set timeout to bound the worst-case wall-clock time per run. Internally implemented with asyncio.wait_for(), which raises asyncio.TimeoutError and triggers the standard retry path.

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

timeout=None (the default) imposes no limit.

---

Import paths (updated in PR #1723):

• Canonical (recommended): from praisonai.scheduler import AsyncAgentScheduler
• Explicit module path: from praisonai.scheduler.async_agent_scheduler import AsyncAgentScheduler
• Deprecated (still works, emits DeprecationWarning): from praisonai.async_agent_scheduler import AsyncAgentScheduler

The sync AgentScheduler is unchanged: from praisonai.scheduler import AgentScheduler.

Jupyter/Event Loop Compatibility: Starting with PR #1448, PraisonAI no longer calls nest_asyncio.apply() or asyncio.set_event_loop() on your behalf when ACP/LSP is enabled. If you embed PraisonAI inside a Jupyter kernel or another running event loop, either call nest_asyncio.apply() yourself at the top of your notebook, or run PraisonAI from a separate process.

---

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

RunPolicy adds run-scoped guardrails (tool scoping, prompt scanning, durable audit) to ScheduledAgentExecutor. AsyncAgentScheduler is independent of RunPolicy — see Scheduled Run Policy for how to add guardrails when using ScheduledAgentExecutor or EnhancedScheduledAgentExecutor.

AsyncAgentScheduler lives in the praisonai wrapper (from praisonai.scheduler import AsyncAgentScheduler). ScheduledAgentExecutor and JobResult live in the bot tier — use from praisonai_bot.scheduler import ScheduledAgentExecutor, JobResult when using them directly.

Scheduler CLI

Command-line interface for scheduling agents

Pre-Run Gate

Skip ticks when a cheap check says nothing to do

Scheduler Delivery

Push scheduled results to Telegram/Discord/Slack/WhatsApp

Background Tasks

Running agents as background processes

Run Policy

Scope tools, scan prompts, and audit output for unattended runs