> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

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

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent

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

<Note>
  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.
</Note>

<Note>
  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](/docs/features/async-scheduler#multi-process-safety-at-most-once).
</Note>

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

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
    subgraph "Async Agent Scheduler"
        A[📋 Agent] --> B[⏰ Scheduler]
        B --> C[🔄 Executor]
        C --> D[✅ Success]
        C --> E[❌ Failure]
        D --> F[📊 Stats]
        E --> F
    end
    
    classDef agent fill:#8B0000,color:#fff
    classDef tool fill:#189AB4,color:#fff
    
    class A agent
    class B,C,D,E,F tool
```

## Quick Start

<Steps>
  <Step title="Simple Usage">
    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    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())
    ```
  </Step>

  <Step title="With Callbacks">
    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    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())
    ```
  </Step>

  <Step title="With Timeout & Budget Limit">
    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    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())
    ```
  </Step>
</Steps>

***

## How It Works

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
    participant User
    participant Scheduler
    participant Agent
    participant Executor
    
    User->>Scheduler: start(schedule_expr)
    Scheduler->>Scheduler: Parse interval
    
    loop Every Interval
        Scheduler->>Executor: execute(task)
        Executor->>Agent: start(task) or astart(task)
        Agent-->>Executor: Result
        alt Success
            Executor-->>Scheduler: Success callback
        else Failure
            Executor-->>Scheduler: Retry with backoff
        end
    end
    
    User->>Scheduler: stop()
    Scheduler->>Scheduler: Cancel gracefully (30s timeout)
```

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

<Note>
  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](https://github.com/MervinPraison/PraisonAI/pull/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](/docs/cli/scheduler) for the corresponding fix to `AgentScheduler`.
</Note>

***

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

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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_sync()]
    F --> G[Direct Read]
    G --> H[May Tear]
    
    classDef async fill:#10B981,stroke:#7C90A0,color:#fff
    classDef sync fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef atomic fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef tear fill:#8B0000,stroke:#7C90A0,color:#fff
    
    class A,B async
    class E,F sync
    class C,D atomic
    class H tear
```

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](https://github.com/MervinPraison/PraisonAI/pull/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

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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())
```

<Note>
  **Real cost tracking ([PR #2171](https://github.com/MervinPraison/PraisonAI/pull/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.
</Note>

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
    Start[Execution Start] --> Check{total_cost >= max_cost?}
    Check -->|Yes| Stop[stop_event.set] --> Halt[Scheduler Halts]
    Check -->|No| Run[Run Agent]
    Run --> Extract[Extract usage from result]
    Extract --> HasUsage{usage present?}
    HasUsage -->|Yes| Price[ModelPricing.calculate_cost in + out tokens]
    HasUsage -->|No| Zero[run_cost = $0.00]
    Price --> AddCost[total_cost += run_cost]
    Zero --> AddCost
    AddCost --> NextLoop[Wait for next interval]

    classDef start fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef check fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef stop fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef run fill:#10B981,stroke:#7C90A0,color:#fff
    classDef calc fill:#189AB4,stroke:#7C90A0,color:#fff

    class Start start
    class Check,HasUsage check
    class Stop,Halt stop
    class Run,NextLoop run
    class Extract,Price,Zero,AddCost calc
```

***

### 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](https://github.com/MervinPraison/PraisonAI/pull/1857).

***

## Best Practices

<AccordionGroup>
  <Accordion title="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.

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    # 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
    ```
  </Accordion>

  <Accordion title="Use run_immediately=True for testing">
    Enable `run_immediately=True` to verify your agent works correctly before waiting for the first scheduled interval.

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    # Test immediately, then schedule
    await scheduler.start("hourly", run_immediately=True)

    # Good for smoke tests
    await scheduler.start("daily", run_immediately=True)
    ```
  </Accordion>

  <Accordion title="Keep callbacks lightweight">
    Success and failure callbacks are called synchronously. Heavy operations should be offloaded to avoid blocking the scheduler.

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    # 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))
    ```
  </Accordion>

  <Accordion title="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.

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    # 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")
    ```
  </Accordion>

  <Accordion title="Budget Control">
    The default `max_cost=1.00` caps unattended cost runaway. Since [PR #2171](https://github.com/MervinPraison/PraisonAI/pull/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.

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    # 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](/docs/cli/cost-tracking#custom-pricing) snippet so the brake stays meaningful.
  </Accordion>

  <Accordion title="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.

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    scheduler = AsyncAgentScheduler(
        agent,
        task,
        timeout=30,        # individual run cannot exceed 30s
        max_cost=1.00,
    )
    ```

    `timeout=None` (the default) imposes no limit.
  </Accordion>
</AccordionGroup>

***

<Note>
  **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`.
</Note>

<Warning>
  **Jupyter/Event Loop Compatibility:** Starting with [PR #1448](https://github.com/MervinPraison/PraisonAI/pull/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.
</Warning>

***

## 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](/docs/features/scheduler-pre-run-gate) for the full configuration reference and examples.

***

## Related

<Note>
  `RunPolicy` adds run-scoped guardrails (tool scoping, prompt scanning, durable audit) to `ScheduledAgentExecutor`. `AsyncAgentScheduler` is independent of `RunPolicy` — see [Scheduled Run Policy](/docs/features/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.
</Note>

<CardGroup cols={2}>
  <Card title="Scheduler CLI" icon="terminal" href="/docs/cli/scheduler">
    Command-line interface for scheduling agents
  </Card>

  <Card title="Pre-Run Gate" icon="filter" href="/docs/features/scheduler-pre-run-gate">
    Skip ticks when a cheap check says nothing to do
  </Card>

  <Card title="Background Tasks" icon="play" href="/docs/features/background-tasks">
    Running agents as background processes
  </Card>

  <Card title="Run Policy" icon="shield-halved" href="/docs/features/scheduled-run-policy">
    Scope tools, scan prompts, and audit output for unattended runs
  </Card>
</CardGroup>