Async Agent Scheduler

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

Quick Start

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

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

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.

Schedule Expression Reference

Expression	Interval	Description
`"daily"`	86400s	Every 24 hours
`"hourly"`	3600s	Every hour
`"*/30m"`	1800s	Every 30 minutes
`"*/1h"`	3600s	Every 1 hour
`"*/5s"`	5s	Every 5 seconds
`"60"`	60s	Custom seconds (plain digits)

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

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
`get_stats()`	sync	Best-effort, NOT atomic — counters may be observed mid-update	Quick sync checks, tests, scripts
`get_stats_sync()`	sync	Alias for `get_stats()` (same best-effort behaviour, named for clarity)	When you want the sync intent obvious in code
`get_stats_async()`	async	Atomic snapshot under `_stats_lock`	Inside async code paths where consistency matters

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)

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

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

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.

Scheduler CLI

Command-line interface for scheduling agents

Background Tasks

Running agents as background processes

​Quick Start

​How It Works

​Schedule Expression Reference

​Configuration Options

​AsyncAgentScheduler Constructor

​start() Method Options

​Reading Stats

​Stats Response Format

​Common Patterns

​Running in FastAPI Application

​Error Handling with Logging

​Graceful Shutdown on SIGINT

​Best Practices

​Related

Scheduler CLI

Background Tasks

Quick Start

How It Works

Schedule Expression Reference

Configuration Options

AsyncAgentScheduler Constructor

start() Method Options

Reading Stats

Stats Response Format

Common Patterns

Running in FastAPI Application

Error Handling with Logging

Graceful Shutdown on SIGINT

Best Practices

Related