Async Agent Scheduler

Async-native agent scheduler that replaces daemon threads with proper async execution and cooperative cancellation.

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

async def main():
    agent = Agent(name="NewsChecker", instructions="Summarise the latest AI news.")
    scheduler = AsyncAgentScheduler(agent=agent, task="Top 3 AI stories")
    await scheduler.start("hourly", run_immediately=True)

asyncio.run(main())

The user schedules recurring agent work; the async scheduler runs jobs with cooperative cancellation.

Quick Start

Simple Usage

Create and start an async scheduler with basic configuration.

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

async def main():
    agent = Agent(
        name="NewsChecker", 
        instructions="Summarise the latest AI news."
    )
    
    scheduler = AsyncAgentScheduler(
        agent=agent, 
        task="Summarise top 3 AI stories"
    )
    
    await scheduler.start("hourly", max_retries=3, run_immediately=True)
    
    # Let it run for 2 hours
    await asyncio.sleep(3600 * 2)
    await scheduler.stop()

asyncio.run(main())

With Callbacks and Configuration

Add success/failure callbacks and custom configuration.

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

def success_callback(result):
    print(f"✅ Success: {result}")

async def async_failure_callback(exception):
    print(f"❌ Async failure: {exception}")

async def main():
    agent = Agent(
        name="DataProcessor",
        instructions="Process and analyze data efficiently."
    )
    
    scheduler = AsyncAgentScheduler(
        agent=agent,
        task="Process latest data batch",
        config={"timeout": 120},
        on_success=success_callback,
        on_failure=async_failure_callback
    )
    
    await scheduler.start("*/30m", max_retries=5)
    
    # Monitor stats
    stats = await scheduler.get_stats_async()
    print(f"Stats: {stats}")
    
    await scheduler.stop()

asyncio.run(main())

With Timeout & Budget Limit

Control execution time and budget with built-in enforcement.

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

How It Works

Multi-process deployments: When more than one BotOS process shares the same schedule store, each due job is claimed atomically before running and fires at most once across all processes — provided the store supports claim_due (the default FileScheduleStore does). See BotOS → Multi-Process / HA Deployments.

Phase	Description
Initialization	Creates async primitives lazily on first use
Scheduling	Runs agent at specified intervals with exponential backoff
Execution	Uses thread pool for sync agents, direct await for async agents
Cancellation	Cooperative cancellation via `asyncio.Event`

Multi-process safety (at-most-once)

Without an atomic claim, two workers polling the same store could each fire the same due job — causing double-posts, double-charges, or duplicated side effects. Since PraisonAI PR #2628, FileScheduleStore claims each due job under an OS advisory file lock (fcntl on POSIX, msvcrt on Windows) and hands it back to only one worker per tick. Losers receive an empty list and skip silently. Under the lock the store re-reads on-disk state, pre-advances last_run_at, and takes a short lease atomically. One-shot jobs are removed on claim. An unfinished claim releases automatically once its lease expires — this is the crash-safety mechanism.

Concept	What it is	Default
`owner_id`	Unique per-worker identifier	`f"{hostname}:{pid}:{uuid}"` (auto)
`lease_seconds`	How long a claim is held before it can be re-claimed	`300.0`
`supports_atomic_claim()`	Runner feature-detect flag; when `False`, the wrapper falls back to non-atomic `get_due_jobs`	auto-detected from the store

from praisonaiagents.scheduler import ScheduleRunner, FileScheduleStore

store = FileScheduleStore()
runner = ScheduleRunner(store)

if runner.supports_atomic_claim():
    jobs = runner.claim_due_jobs(owner_id="worker-1", lease_seconds=300.0)
    for job in jobs:
        # ... run the job ...
        runner.complete_run(job.id, owner_id="worker-1")
else:
    jobs = runner.get_due_jobs()

The wrapper’s ScheduledAgentExecutor.tick() does this automatically: it claims under the lock when the store supports it (default owner_id = f"{hostname}:{pid}:{uuid}"), releases the lease after each run, and falls back to get_due_jobs otherwise.

fcntl / msvcrt locks are honored per host. Sharing one FileScheduleStore path across hosts only works on a filesystem that honors the lock — network filesystems like NFS do not reliably do so. Use a shared database-backed store for cross-host at-most-once instead.

Configuration Options

AsyncAgentScheduler API Reference

Complete parameter documentation and examples

Constructor Parameters

Parameter	Type	Default	Description
`agent`	`Any`	Required	Agent instance to schedule
`task`	`str`	Required	Task description to execute
`config`	`Dict[str, Any]`	`{}`	Optional configuration
`on_success`	`Callable`	`None`	Success callback (sync or async)
`on_failure`	`Callable`	`None`	Failure callback (sync or async)
`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 Parameters

Parameter	Type	Default	Description
`schedule_expr`	`str`	Required	Schedule interval expression
`max_retries`	`int`	`3`	Total attempts (1 initial + retries)
`run_immediately`	`bool`	`False`	Execute immediately before scheduling

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.

Common Patterns

Pattern 1: Event Loop Safety

The scheduler is safe to construct outside an event loop:

# Safe to create in sync code
scheduler = AsyncAgentScheduler(agent, task)

async def run_later():
    # Async primitives created here
    await scheduler.start("hourly")

Pattern 2: FastAPI Integration

from fastapi import FastAPI
from contextlib import asynccontextmanager

scheduler = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global scheduler
    scheduler = AsyncAgentScheduler(agent, "Background task")
    await scheduler.start("*/10m")
    yield
    await scheduler.stop()

app = FastAPI(lifespan=lifespan)

Pattern 3: Cancellation Handling

async def graceful_shutdown():
    try:
        await scheduler.start("hourly")
    except asyncio.CancelledError:
        print("Scheduler cancelled")
    finally:
        await scheduler.stop()

Pattern 4: 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.

Best Practices

Event Loop Safety

Always create async primitives lazily. The scheduler binds asyncio.Event and asyncio.Lock to the caller’s loop on first async entry, preventing “different loop” errors.

# ✅ Good: Constructor safe in sync code
scheduler = AsyncAgentScheduler(agent, task)

async def later():
    # ✅ Good: Async primitives created here
    await scheduler.start("hourly")

Callback Best Practices

Use both sync and async callbacks safely. The safe_call utility handles both types automatically.

def sync_callback(result):
    print(f"Sync: {result}")

async def async_callback(result):
    await log_to_database(result)

# ✅ Both work seamlessly
scheduler = AsyncAgentScheduler(
    agent=agent,
    task=task,
    on_success=async_callback,
    on_failure=sync_callback
)

Retry Strategy

Use exponential backoff with jitter. Both sync and async schedulers share the same backoff_delay algorithm for consistency.

# Delay formula: min(max(30, 2 ** attempt), 300) * jitter
# - Floor: ~27s, Cap: 300s
# - Same behavior as sync AgentScheduler
await scheduler.start("hourly", max_retries=5)

Graceful Shutdown

Always await stop() for proper cleanup and statistics reporting.

try:
    await scheduler.start("hourly")
    await asyncio.sleep(3600)
finally:
    stats = await scheduler.get_stats()
    await scheduler.stop()
    print(f"Final stats: {stats}")

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.

Multi-process lease tuning

Pick a lease_seconds long enough to cover your slowest job, but short enough to recover quickly after a crash — an unfinished claim only re-runs after its lease expires.

# Long-running jobs: hold the claim for the full worst-case run time
runner.claim_due_jobs(owner_id="worker-1", lease_seconds=600.0)

# Fast jobs: shorter lease recovers sooner from a crashed worker
runner.claim_due_jobs(owner_id="worker-1", lease_seconds=60.0)

Don’t share a FileScheduleStore path across hosts unless the filesystem honors the lock (NFS does not). Leave owner_id on its auto default (f"{hostname}:{pid}:{uuid}") unless you need a stable value for observability.

Import paths (PR #1552):

Pending deprecation (still works, emits PendingDeprecationWarning — will move to praisonai.scheduler.async_agent_scheduler in a future release): from praisonai.scheduler import AsyncAgentScheduler
Canonical: from praisonai.scheduler import AgentScheduler (sync scheduler)
Deprecated (still works, emits DeprecationWarning): from praisonai.agent_scheduler import AgentScheduler

The canonical sync AgentScheduler exposes from_yaml, start_from_yaml_config, and from_recipe, which the deprecated top-level shim does not advertise but now re-exports correctly.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.

Skipping Ticks with a Pre-Run Gate

A pre-run gate runs a cheap shell command before each tick and skips the model turn when the check says there’s nothing to do — zero tokens spent on quiet ticks. See Scheduler Pre-Run Gate for configuration options and examples.

Sync Agent Scheduler

Thread-based scheduler for sync environments

Pre-Run Gate

Skip ticks when a cheap check says nothing to do

Shared Scheduler Utilities

Common primitives used by both schedulers

​Quick Start

​How It Works

​Multi-process safety (at-most-once)

​Configuration Options

AsyncAgentScheduler API Reference

​Constructor Parameters

​Start Parameters

​Stats Response Format

​Common Patterns

​Pattern 1: Event Loop Safety

​Pattern 2: FastAPI Integration

​Pattern 3: Cancellation Handling

​Pattern 4: Budget-aware Scheduling

​Best Practices

​Skipping Ticks with a Pre-Run Gate

​Related

Sync Agent Scheduler

Pre-Run Gate

Shared Scheduler Utilities

Quick Start

How It Works

Multi-process safety (at-most-once)

Configuration Options

Constructor Parameters

Start Parameters

Stats Response Format

Common Patterns

Pattern 1: Event Loop Safety

Pattern 2: FastAPI Integration

Pattern 3: Cancellation Handling

Pattern 4: Budget-aware Scheduling

Best Practices

Skipping Ticks with a Pre-Run Gate

Related