> ## 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. # Background Tasks > Run agent tasks and recipes asynchronously in the background Run agent tasks and recipes in the background without blocking your main thread. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent agent = Agent( name="AsyncAssistant", instructions="Research in the background while the user keeps working.", background=True, ) agent.start("Summarise today's news.") ``` The user submits long-running work; the background runner executes it concurrently while the main thread continues. ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} graph LR Main[ā–¶ļø Main thread] -->|submit| Runner[āš™ļø BackgroundRunner] Runner --> Task[šŸ¤– Agent task] Task --> Result[āœ… Result] Main -->|continue| Other[Other work] classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff classDef tool fill:#189AB4,stroke:#7C90A0,color:#fff classDef ok fill:#10B981,stroke:#7C90A0,color:#fff classDef input fill:#6366F1,stroke:#7C90A0,color:#fff class Task agent class Runner tool class Result ok class Main,Other input ``` ## Quick Start ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} import asyncio from praisonaiagents import Agent from praisonaiagents.background import BackgroundRunner, BackgroundConfig async def main(): runner = BackgroundRunner(config=BackgroundConfig(max_concurrent_tasks=3)) agent = Agent( name="AsyncAssistant", instructions="You are a research assistant.", background=runner, ) task = await agent.background.submit_agent( agent=agent, prompt="Research AI trends in 2025", name="research_task", ) await task.wait(timeout=60.0) print(task.result) asyncio.run(main()) ``` ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonai import recipe task = recipe.run_background( "my-recipe", input={"query": "What is AI?"}, config={"max_tokens": 1000}, session_id="session_123", timeout_sec=300, ) print(f"Task ID: {task.task_id}") task.wait() ``` Inside a running event loop (a FastAPI handler, a Jupyter cell, or any `async def`), use `arun_background` ā€” the sync `run_background` raises `RuntimeError` there instead of deadlocking: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonai import recipe async def handler(): task = await recipe.arun_background( "my-recipe", input={"query": "What is AI?"}, config={"max_tokens": 1000}, ) result = await task.wait() return result ``` Do not call `recipe.run_background()` from within a running event loop ā€” it raises `RuntimeError` pointing you to `arun_background`. Use the sync form only from plain synchronous code. ## Features * **Async Execution**: Run tasks without blocking * **Concurrency Control**: Limit concurrent tasks * **Progress Tracking**: Monitor task status * **Timeout Support**: Set execution time limits * **Cancellation**: Cancel running tasks ## Configuration ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents.background import BackgroundConfig config = BackgroundConfig( max_concurrent_tasks=5, # Max parallel tasks default_timeout=300.0, # 5 minute default timeout auto_cleanup=True # Auto-remove completed tasks ) ``` ## Task Status ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents.background import TaskStatus # Check status if task.status == TaskStatus.COMPLETED: print(f"Result: {task.result}") elif task.status == TaskStatus.FAILED: print(f"Error: {task.error}") elif task.status == TaskStatus.RUNNING: print("Still running...") ``` ## In-session `/tasks` command Inspect and cancel background tasks from inside a `praisonai code` session or a bot chat ā€” no need to `Ctrl-C` and run `praisonai background list`. | Command | Purpose | | -------------------- | -------------------------------------------------- | | `/tasks` | List background tasks (id, name, status, progress) | | `/tasks ` | Show detail for one task (incl. result / error) | | `/tasks cancel ` | Cancel a running background task | Type `/tasks` after kicking off a `background=True` run to check on it without leaving the conversation: ```text theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} > /tasks šŸ§© Background tasks: ā€¢ aa1b2cā€¦ research ā€” running (35%) ā€¢ ff9d3eā€¦ summary ā€” completed (100%) Use /tasks for detail or /tasks cancel . > /tasks aa1b2c šŸ§© Task aa1b2c ā€” research Status: running | Progress: 35% > /tasks cancel aa1b2c āœ… Cancelled task aa1b2c. ``` The REPL, bots, and CLI all read the same shared runner, so a task appears wherever you look for it. ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} sequenceDiagram participant User participant Surface as REPL / Bot / CLI participant Runner as Shared BackgroundRunner participant Task as BackgroundTask User->>Surface: /tasks Surface->>Runner: get_background_runner() Runner-->>Surface: task list (owner-scoped in bots) Surface-->>User: compact table User->>Surface: /tasks cancel Surface->>Runner: cancel_task_sync(id) Runner->>Task: stop underlying future Task-->>Runner: CANCELLED Runner-->>Surface: āœ… cancelled Surface-->>User: confirmation ``` In bot chats, `/tasks` is **per-user**: a caller can only see and cancel tasks whose `metadata["user_id"]` matches their bot user id. Tasks submitted from the CLI or REPL without a `user_id` are **not** exposed to bot users ā€” fail-closed by design. The same command works in Telegram, Slack, and Discord ā€” see [Bot Chat Commands](/docs/features/bot-commands#tasks). ### Common Pattern: check on a background run from anywhere Ask the agent for a long research task with `background=True`. The agent replies with a task id and keeps chatting. Type `/tasks` ā€” the research shows as `running` at 60%. Type `/tasks ` for the current progress detail. Type `/tasks` there. Because bot `/tasks` is per-user scoped, you see only the tasks you submitted from Telegram ā€” the REPL ones stay hidden. `/tasks cancel ` ā€” the task actually stops (its underlying future is cancelled, not just the record marked). ### Shared runner accessor Every inspection surface resolves the **same** process-wide runner through `get_background_runner()`. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents.background import get_background_runner # Same runner across REPL /tasks, bot /tasks, praisonai background list, # and recipe.arun_background ā€” created lazily on first call. runner = get_background_runner() ``` `/tasks` sees the same tasks regardless of which surface submitted them. `BackgroundRunner` is still available directly for advanced users who need a private runner. ## CLI Usage ```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # Submit a recipe as background task praisonai background submit --recipe my-recipe # List tasks praisonai background list # Check status praisonai background status # Cancel task praisonai background cancel # Clear completed praisonai background clear ``` ## Safe Defaults | Setting | Default | Description | | ------------------- | ------- | ------------------------------------------ | | `timeout_sec` | 300 | Maximum execution time (5 minutes) | | `max_concurrent` | 5 | Maximum concurrent tasks | | `cleanup_delay_sec` | 3600 | Time before completed tasks are cleaned up | *** ## Low-level API Reference ### BackgroundRunner Direct Usage ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} import asyncio from praisonaiagents.background import BackgroundRunner, BackgroundConfig async def main(): # Create runner with config config = BackgroundConfig(max_concurrent_tasks=3) runner = BackgroundRunner(config=config) # Define a task async def my_task(name: str) -> str: await asyncio.sleep(2) return f"Task {name} completed" # Submit task task = await runner.submit(my_task, args=("example",), name="my_task") print(f"Submitted: {task.id[:8]}") # Wait for completion await task.wait(timeout=10.0) print(f"Result: {task.result}") asyncio.run(main()) ``` ### Submitting Tasks ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # Submit async function task = await runner.submit( func=my_async_function, args=(arg1, arg2), kwargs={"key": "value"}, name="descriptive_name", timeout=60.0 ) # Submit sync function (runs in thread pool) task = await runner.submit( func=my_sync_function, args=(arg1,), name="sync_task" ) ``` ### Task Management ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # List all tasks for task in runner.tasks: print(f"{task.name}: {task.status.value}") # Get running tasks running = runner.running_tasks # Get pending tasks pending = runner.pending_tasks # Clear completed tasks runner.clear_completed() ``` ### Synchronous Job Manager For simpler use cases, use `BackgroundJobManager` for synchronous job management: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents.background.job_manager import BackgroundJobManager, JobStatus # Create manager with auto-background threshold manager = BackgroundJobManager(auto_background_threshold=5.0) # Start a job job_id = manager.start_job(lambda: expensive_computation()) # Check status status = manager.get_status(job_id) if status == JobStatus.COMPLETED: result = manager.get_result(job_id) elif status == JobStatus.FAILED: error = manager.get_error(job_id) # List all jobs for job_id, info in manager.list_jobs().items(): print(f"{job_id}: {info.status}") # Cancel a running job manager.cancel_job(job_id) ``` Jobs already survive a process restart when you use `get_job_manager()` ā€” it attaches a `SqliteBackgroundJobStore` and reconciles on boot. You only pass `store=` when you construct `BackgroundJobManager` yourself. See [Durability](#durability-survive-a-restart) below. ## Durability ā€” survive a restart Durability is **on by default**. The shared manager from `get_job_manager()` persists every state transition to `/background_jobs.db` and runs `reconcile_on_start()` once at first construction, so a crash mid-job no longer drops in-flight work or the promised deliver-back. In-memory is the opt-out: set `PRAISONAI_BACKGROUND_JOB_STORE=0` for the shared manager, or omit `store=` on a custom `BackgroundJobManager`. ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} sequenceDiagram participant App participant Manager as BackgroundJobManager participant Store as BackgroundJobStore App->>Manager: start_job(func, origin=...) Manager->>Store: upsert(PENDING) Manager->>Store: upsert(RUNNING) Note over Manager: šŸ’„ process crashes mid-job App->>Manager: restart + reconcile_on_start(redeliver) Manager->>Store: list_unreconciled() Store-->>Manager: orphaned + undelivered jobs Manager->>Store: upsert(LOST) Manager->>App: redeliver(job_info) for undelivered ``` ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents.background.job_manager import get_job_manager manager = get_job_manager() # already backed by SqliteBackgroundJobStore # already ran reconcile_on_start() once ``` Building your own `BackgroundJobManager`? Pass `store=SqliteBackgroundJobStore()` (or any `BackgroundJobStore`) and reconcile once at boot: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents.background import SqliteBackgroundJobStore from praisonaiagents.background.job_manager import BackgroundJobManager manager = BackgroundJobManager(store=SqliteBackgroundJobStore()) counts = manager.reconcile_on_start(redeliver=on_job_complete) # {"lost": 0, "redelivered": 0, "rehydrated": 0} on a clean start ``` ### New API surface | Symbol | Kind | Purpose | | -------------------------------- | ------------------------------------- | ------------------------------------------------------------------------------------------------ | | `get_job_manager()` | function | Shared manager ā€” durable by default via `SqliteBackgroundJobStore` | | `SqliteBackgroundJobStore` | class | `from praisonaiagents.background import SqliteBackgroundJobStore` ā€” built-in stdlib SQLite store | | `BackgroundJobStore` | Protocol | Implement this to plug in a non-SQLite durable store | | `store=` | `BackgroundJobManager.__init__` kwarg | Enable persistence on a custom manager (keyword-only) | | `PRAISONAI_BACKGROUND_JOB_STORE` | env var | Set to `0` / `false` to force `get_job_manager()` back to pure in-memory | | `reconcile_on_start(redeliver)` | method | Reconcile orphans + replay undelivered on startup | | `JobStatus.LOST` | enum value | Terminal state for `RUNNING`/`PENDING` jobs interrupted by a crash | | `JobInfo.delivered` | field (`bool`, default `False`) | Whether the deliver-back has fired | `reconcile_on_start(redeliver)` passes the persisted `JobInfo` to your `redeliver` callback ā€” make it idempotent, since a raised exception means "retry on the next restart" (the job is left undelivered, never lost). `LOST` jobs are age-evictable by `cleanup_completed(max_age=...)`, so reconciled orphans don't leak across restarts. ## Architecture The sync wrappers and `ScheduleLoop` bridge the gap between disconnected modules: ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} graph TB subgraph Scheduler["scheduler/ module"] S1["schedule_add()"] S2["ConfigYamlScheduleStore (default)
FileScheduleStore"] S3["ScheduleRunner"] S4["ScheduleLoop āœØ"] S1 --> S2 S3 --> S2 S4 -->|"polls"| S3 end subgraph Background["background/ module"] B1["BackgroundRunner"] B2["BackgroundTask"] B3["submit_sync() āœØ"] B4["submit_agent_sync() āœØ"] B1 --> B2 B3 --> B1 B4 --> B1 end subgraph YourApp["Your Application"] C1["Sync Code"] C2["on_trigger callback"] end S4 -->|"fires"| C2 C2 -->|"can use"| B3 C1 -->|"direct use"| B3 C1 -->|"agent tasks"| B4 ``` * **`submit_sync()` / `submit_agent_sync()`** ā€” let sync code submit background tasks without asyncio boilerplate * **`ScheduleLoop`** ā€” polls for due jobs on a daemon thread and fires your callback ## Sync Wrappers For sync code (scripts, bot handlers, `Agent.start()` callbacks), use the sync-friendly methods that handle asyncio automatically: ### submit\_sync() Submit any callable from synchronous code. A daemon event loop thread is created lazily on first call. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents.background.runner import BackgroundRunner import time runner = BackgroundRunner() def heavy_computation(data): time.sleep(10) return {"result": sum(data)} # Non-blocking ā€” returns immediately task = runner.submit_sync( func=heavy_computation, args=([1, 2, 3, 4, 5],), name="compute" ) print(task.status) # "running" # Check later while not task.is_completed: time.sleep(1) print(f"Result: {task.result}") # {'result': 15} ``` ### submit\_agent\_sync() Submit an Agent task from synchronous code. Resolves the agent's callable (`start` ā†’ `chat` ā†’ `run`) automatically. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent from praisonaiagents.background.runner import BackgroundRunner agent = Agent(name="researcher", instructions="Research AI trends") runner = BackgroundRunner() task = runner.submit_agent_sync( agent=agent, prompt="What are the top AI trends in 2026?", name="research-task" ) # task.result will contain the agent's response when done ``` ### cancel\_task\_sync() Cancel a task from sync code or an unrelated event loop. It hops threads via `run_coroutine_threadsafe`, so it's safe to call from within a running event loop and actually stops the underlying future. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents.background import get_background_runner runner = get_background_runner() cancelled = runner.cancel_task_sync("task-id-here") # True ā†’ cancelled; False ā†’ not found or already completed ``` | Parameter | Type | Required | Description | | ---------------- | ---------------- | -------- | ------------------------------------------------------------------------------------------------------------------------------------------- | | `func` / `agent` | Callable / Agent | Yes | Function or Agent to execute | | `args` | tuple | No | Positional arguments (submit\_sync only) | | `prompt` | str | Yes | Agent prompt (submit\_agent\_sync only) | | `name` | str | No | Human-readable task name | | `timeout` | float | No | Timeout in seconds | | `on_complete` | Callable | No | Fires exactly once when the task reaches any terminal state ā€” `COMPLETED`, `FAILED`, `CANCELLED`. Callback errors are swallowed and logged. | **Terminal-branch guarantee (PR [#3883](https://github.com/MervinPraison/PraisonAI/pull/3883)):** `on_complete(task)` fires on **every** terminal branch ā€” success, generic exception, `asyncio.TimeoutError`, and `asyncio.CancelledError`. Before #3883 it did not fire on timeout or cancel, so callers writing subscribers had to poll `.status` to detect those. That workaround is no longer needed. Inspect `task.status`, `task.error`, and `task.result` inside your callback to distinguish outcomes. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents.background import BackgroundRunner, TaskStatus def on_done(task): if task.status == TaskStatus.COMPLETED: print("done:", task.result) elif task.status == TaskStatus.CANCELLED: print("cancelled") elif task.status == TaskStatus.FAILED: print("failed:", task.error) runner = BackgroundRunner() runner.submit_sync(coro, name="my-task", timeout=30, on_complete=on_done) ``` Cancellation still re-raises `asyncio.CancelledError` for cooperative shutdown ā€” but only **after** `on_complete` fires. *** ## ScheduleLoop `ScheduleLoop` bridges scheduled jobs to actual execution. It runs a daemon thread that polls `get_due_jobs()` and fires your callback. To skip ticks cheaply before the model turn runs, see [`pre_run` in Schedule Tools](/docs/docs/tools/schedule-tools#pre-run-condition-gate). ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents.scheduler import ScheduleLoop def handle_job(job): print(f"šŸ”” Firing: {job.name} ā€” {job.message}") loop = ScheduleLoop( on_trigger=handle_job, tick_seconds=30, # check every 30 seconds ) loop.start() # daemon thread ā€” won't block # loop.stop() # clean shutdown when needed ``` ### Combined Example: Scheduler + Background ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent from praisonaiagents.tools import schedule_add, schedule_list, schedule_remove from praisonaiagents.scheduler import ScheduleLoop from praisonaiagents.background.runner import BackgroundRunner agent = Agent( name="assistant", instructions="You can set reminders and schedules.", tools=[schedule_add, schedule_list, schedule_remove], ) runner = BackgroundRunner() def on_schedule_fire(job): task = runner.submit_agent_sync(agent, job.message, name=f"schedule-{job.name}") print(f"ā†’ Background task {task.id} started for '{job.name}'") loop = ScheduleLoop(on_trigger=on_schedule_fire, tick_seconds=30) loop.start() # Agent creates schedules that now actually fire! agent.start("Remind me to check email every morning at 7am") ``` | Parameter | Type | Default | Description | | -------------- | --------------------- | --------------------------- | ------------------------------------------------- | | `on_trigger` | Callable | *required* | Called with each due `ScheduleJob` | | `store` | ScheduleStoreProtocol | `ConfigYamlScheduleStore()` | Schedule store to poll. Auto-created when omitted | | `tick_seconds` | float | `30.0` | Poll interval in seconds | | Method | Description | | ------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `loop.start(on_due=None, store=None)` | Start polling (no-op if already running). Optional `on_due` callback replaces the internal claim+fire path ā€” lets an external caller drive *when* firing happens. Optional `store` overrides the configured store before starting. Zero-arg call is unchanged. | | `loop.fire_due()` | Claim + fire one tick's worth of due jobs **without** starting the daemon thread. Use from external providers (webhook, cron, systemd). Serialized under a per-instance lock. | | `loop.stop(timeout=5.0)` | Signal stop and wait for thread exit | | `loop.is_running` | Whether the daemon thread is alive | **`ScheduleLoop` is the default `SchedulerProviderProtocol`** ā€” the in-process poll thread, also exported as `InProcessScheduleProvider`. For event-driven / serverless firing (webhook, systemd timer, cron, K8s CronJob), see [Scheduler Providers](/docs/features/scheduler-providers). **Error handling:** If `on_trigger()` raises an exception, it's logged but not propagated ā€” the loop continues with remaining jobs and future ticks. *** ## Zero Performance Impact The background module uses lazy loading ā€” no overhead when not used: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # Only loads when accessed from praisonaiagents.background import BackgroundRunner # Sync loop thread only created on first submit_sync() call import praisonaiagents.background.runner as br assert br._bg_loop is None # Not created until needed ``` *** ## Best Practices Set `BackgroundConfig(max_concurrent_tasks=...)` to match CPU and API rate limits ā€” unbounded parallelism can exhaust tokens or file handles. Background work can hang on tool calls; pass `timeout=` so your main loop can cancel or retry instead of blocking forever. Fire scheduled jobs into `BackgroundRunner.submit_agent_sync` so reminders run without blocking the scheduler thread. When callers are external services, prefer the [Async Jobs](/docs/features/async-jobs) server instead of embedding `BackgroundRunner` in app code. *** ## Related HTTP API for submitting and polling long-running agent jobs. Let agents create reminders that trigger background runs. Inspect and cancel background tasks from Telegram, Slack, Discord ā€” per-user scoped.