Background Tasks

Run agent tasks and recipes asynchronously in the background

Run agent tasks and recipes in the background without blocking your main thread.

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.

Quick Start

1. Agent with background runner

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

2. Recipe in the background

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

3. Async callers (FastAPI / Jupyter)

   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:

   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

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

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

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.

Type /tasks after kicking off a background=True run to check on it without leaving the conversation:

> /tasks
🧩 Background tasks:
• aa1b2c… research — running (35%)
• ff9d3e… summary  — completed (100%)

Use /tasks <id> for detail or /tasks cancel <id>.

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

The same command works in Telegram, Slack, and Discord — see Bot Chat Commands.

/tasks now also lists background subagents (chat-spawned jobs) owned by the caller — see Background Subagents and Bot Commands → /tasks.

Common Pattern: check on a background run from anywhere

1. Kick off a background task in praisonai code

   Ask the agent for a long research task with background=True. The agent replies with a task id and keeps chatting.

2. Peek a few minutes later

   Type /tasks — the research shows as running at 60%. Type /tasks <id> for the current progress detail.

3. Switch to your Telegram bot

   Type /tasks there. Because bot /tasks is per-user scoped, you see only the tasks you submitted from Telegram — the REPL ones stay hidden.

4. Cancel when no longer needed

   /tasks cancel <id> — 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().

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

# Submit a recipe as background task
praisonai background submit --recipe my-recipe

# List tasks
praisonai background list

# Check status
praisonai background status <task_id>

# Cancel task
praisonai background cancel <task_id>

# Clear completed
praisonai background clear

Safe Defaults

Low-level API Reference

BackgroundRunner Direct Usage

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

# 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

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

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)

get_result raises:

Duplicate job ids

A job_id that is still pending or running cannot be reused — start_job raises ValueError. Ids of finished jobs (completed, failed, cancelled, lost) can be reused.

manager.start_job(build_report, job_id="daily-report")
manager.start_job(build_report, job_id="daily-report")  # ValueError — still running

manager.get_result("daily-report")  # wait for the first run to finish
manager.start_job(build_report, job_id="daily-report")  # ok

Auto-generated ids are 12 hex characters.

Durability — survive a restart

Durability is on by default. The shared manager from get_job_manager() persists every state transition to <runs_dir>/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.

1. Use the shared manager (durability is already on)

   from praisonaiagents.background.job_manager import get_job_manager
   
   manager = get_job_manager()  # already backed by SqliteBackgroundJobStore
                                # already ran reconcile_on_start() once

2. Advanced: custom store on your own manager

   Building your own BackgroundJobManager? Pass store=SqliteBackgroundJobStore() (or any BackgroundJobStore) and reconcile once at boot:

   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

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

Architecture

The sync wrappers and ScheduleLoop bridge the gap between disconnected modules:

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

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.

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.

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

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.

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

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

Zero Performance Impact

The background module uses lazy loading — no overhead when not used:

# 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

Related