Async Agents

Async agents run non-blocking AI tasks with await, letting you process multiple requests in parallel or embed agents inside async web servers.

import asyncio
from praisonaiagents import Agent

async def main():
    agent = Agent(name="helper", instructions="Reply concisely.")
    print(await agent.astart("Hello"))

asyncio.run(main())

The user sends concurrent requests; each astart() yields on I/O so the server stays responsive under load.

Quick Start

Simple Usage

Use astart() inside an async function:

import asyncio
from praisonaiagents import Agent

agent = Agent(
    name="AsyncAgent",
    instructions="You are a helpful assistant."
)

async def main():
    result = await agent.astart("Summarize the benefits of async programming")
    print(result)

asyncio.run(main())

With Configuration

Run multiple agents in parallel with asyncio.gather:

import asyncio
from praisonaiagents import Agent, Task, PraisonAIAgents

research_agent = Agent(name="Researcher", instructions="Research topics deeply")
writer_agent = Agent(name="Writer", instructions="Write clear summaries")

task1 = Task(
    description="Research async patterns in Python",
    agent=research_agent,
    async_execution=True,
    expected_output="Research notes"
)
task2 = Task(
    description="Write a summary of async best practices",
    agent=writer_agent,
    async_execution=True,
    expected_output="Clear summary"
)

async def main():
    agents = PraisonAIAgents(
        agents=[research_agent, writer_agent],
        tasks=[task1, task2],
        process="sequential"
    )
    result = await agents.astart()
    print(result)

asyncio.run(main())

How It Works

Sync vs Async Methods

Sync	Async	Use When
`agent.start()`	`await agent.astart()`	Running an agent
`agent.chat()`	`await agent.achat()`	Conversational turn
`agents.start()`	`await agents.astart()`	Multi-agent orchestration

Async performance: await agent.achat() no longer blocks the event loop on message persistence. The file-locked disk write is offloaded via asyncio.to_thread, so concurrent async turns stay responsive. The sync path is unchanged.

Configuration Options

Set async_execution=True on individual tasks to mark them for async execution:

from praisonaiagents import Task

task = Task(
    description="Process this document",
    agent=my_agent,
    async_execution=True,
    expected_output="Processed result",
    callback=my_async_callback,
)

Option	Type	Default	Description
`async_execution`	`bool`	`False`	Mark task for async execution
`callback`	`Callable`	`None`	Sync or async callback on task completion

Common Patterns

Parallel Requests

import asyncio
from praisonaiagents import Agent

agent = Agent(
    name="Analyst",
    instructions="Analyze topics concisely."
)

async def main():
    topics = ["Python", "JavaScript", "Rust"]
    tasks = [agent.astart(f"In one sentence, describe {t}") for t in topics]
    results = await asyncio.gather(*tasks)
    for topic, result in zip(topics, results):
        print(f"{topic}: {result}")

asyncio.run(main())

Async Callback

import asyncio
from praisonaiagents import Agent, Task, PraisonAIAgents

async def on_complete(output):
    print(f"Task done: {output.raw[:100]}")

agent = Agent(name="Worker", instructions="Complete assigned tasks")

task = Task(
    description="Summarize the history of the internet",
    agent=agent,
    async_execution=True,
    callback=on_complete,
    expected_output="Brief summary"
)

async def main():
    agents = PraisonAIAgents(agents=[agent], tasks=[task])
    await agents.astart()

asyncio.run(main())

Failure semantics. PraisonAIAgents.arun_all_tasks waits for every sibling async task to complete before re-raising the first exception it saw. Predictable behaviour: no orphaned background tasks continue to mutate self.tasks after the workflow has surfaced a failure.

Failure cascade in `asequential()`

asequential() now skips downstream tasks whose upstream dependency permanently failed, matching what sequential() already did. Before this change the async entry point astart() could run a task against context data its upstream had never produced — the task ran with empty context and emitted garbage. Now the failure short-circuits and cascades a skip down the dependency chain.

import asyncio
from praisonaiagents import Agent, Task, PraisonAIAgents

fetcher = Agent(name="Fetcher", instructions="Fetch the raw data.")
transformer = Agent(name="Transformer", instructions="Transform the fetched data.")
reporter = Agent(name="Reporter", instructions="Report on the transformed data.")

task_a = Task(name="A", description="Fetch data", agent=fetcher, expected_output="Raw data")
task_b = Task(name="B", description="Transform data", agent=transformer, context=[task_a], expected_output="Clean data")
task_c = Task(name="C", description="Report data", agent=reporter, context=[task_b], expected_output="Report")

async def main():
    workflow = PraisonAIAgents(
        agents=[fetcher, transformer, reporter],
        tasks=[task_a, task_b, task_c],
        process="sequential",
    )
    await workflow.astart()

asyncio.run(main())

If task A fails permanently, B and C are marked as skipped (their upstream failed) instead of running against empty context.

Behaviour verified against merged source in PraisonAI PR #3960. The async path (asequential() / astart()) now mirrors the sync sequential() dependency-cascade check.

Inside a Web Framework (FastAPI)

from fastapi import FastAPI
from praisonaiagents import Agent

app = FastAPI()
agent = Agent(name="APIAgent", instructions="Answer user questions helpfully.")

@app.get("/ask")
async def ask(question: str):
    result = await agent.astart(question)
    return {"answer": result}

The synchronous praisonai.run(...) API is also safe to call from an async context — as of the 2026-07-30 release it dispatches through the shared async bridge and works from FastAPI handlers, notebooks, and async def tests without a RuntimeError. Prefer await agent.astart(...) when you can, but sync mode is a valid fallback when a library only exposes a sync surface.

Lifecycle Hooks in Async Workflows

on_task_start and on_task_complete fire from astart() / arun_task() exactly as they do from start():

import asyncio
from praisonaiagents import Agent, Task, PraisonAIAgents, MultiAgentHooksConfig

worker = Agent(name="Worker", instructions="Do the work.")
task = Task(description="Process batch #42", agent=worker)

async def on_complete(task, task_output):
    # native awaitable — no run_in_executor needed
    print(f"Notifying downstream: {task.name}")

team = PraisonAIAgents(
    agents=[worker],
    tasks=[task],
    hooks=MultiAgentHooksConfig(on_task_complete=on_complete),
)

asyncio.run(team.astart())

async def callbacks are awaited; sync callbacks are offloaded to the default executor. See Multi-Agent Hooks for the full contract.

Best Practices

Always use asyncio.run() as the entry point

Call asyncio.run(main()) once at the program entry point. Avoid calling it inside already-running event loops — use await there instead.

Limit concurrency to avoid rate limits

Use asyncio.Semaphore(n) to cap simultaneous LLM calls: async with asyncio.Semaphore(5): result = await agent.astart(...). Start with n=5 and tune based on your API limits.

Handle errors per-task with gather(return_exceptions=True)

Pass return_exceptions=True to asyncio.gather() so one failed task doesn’t cancel the others. Check each result for isinstance(result, Exception) before using it.

Use async callbacks for async downstream work

Callbacks can be async (async def callback(output): ...). The runtime dispatches them safely even when called from within a running event loop.

Workflows

Build parallel and sequential multi-agent workflows

Async Crew Kickoff

Run YAML-defined crews asynchronously

Getting Started

Install

Core Concepts

Guides

Features

Models

Databases

Observability

Memory

Telemetry & Analytics

Knowledge

RAG

Persistence

Tools

Other Features

Developers

Configuration

Best Practices

Getting Started (No Code)

Quick Start

How It Works

Sync vs Async Methods

Configuration Options

Common Patterns

Parallel Requests

Async Callback

Failure cascade in `asequential()`

Inside a Web Framework (FastAPI)

Lifecycle Hooks in Async Workflows

Best Practices

Workflows

Async Crew Kickoff

​Quick Start

​How It Works

​Sync vs Async Methods

​Configuration Options

​Common Patterns

​Parallel Requests

​Async Callback

​Failure cascade in asequential()

​Inside a Web Framework (FastAPI)

​Lifecycle Hooks in Async Workflows

​Best Practices

​Related

Workflows

Async Crew Kickoff

Quick Start

How It Works

Sync vs Async Methods

Configuration Options

Common Patterns

Parallel Requests

Async Callback

Failure cascade in `asequential()`

Inside a Web Framework (FastAPI)

Lifecycle Hooks in Async Workflows

Best Practices

Related