Async Agents

Run agents and tasks asynchronously for better throughput — parallel execution with asyncio

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

1. 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())
   
2. 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.

Knowledge search and memory writes also offload to threads. On the async path, knowledge.search(...) and task-callback memory writes (store_in_memory) each run via asyncio.to_thread, so an asyncio.gather(...) fan-out doesn’t stall on one slow embedding or DB write. The built-in in-memory adapter is thread-safe under these parallel writes. See Async Safety.

---

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.

---

Failed turns leave chat_history clean

A failed await agent.achat(...) or await agent.astart(...) rolls back chat_history to its pre-turn state, so the user message from the failed turn never lingers without a reply.

Behaviour change in PraisonAI PR #4739 — parity with the sync chat() path. Both async generic exception handlers in _achat_impl now call self._rollback_chat_history_to(chat_history_length). If you had a manual .chat_history.pop() workaround around await agent.achat(...), you can remove it.

import asyncio
from praisonaiagents import Agent

async def main():
    agent = Agent(name="worker", llm="gpt-4o-mini")
    before = len(agent.chat_history)
    try:
        await agent.achat("...")  # provider transient / non-tool exception
    except Exception:
        pass
    assert len(agent.chat_history) == before  # unchanged, safe to retry

asyncio.run(main())

“Clean” means “unchanged from just before the call” — not “empty”, and not “user message present with no reply”.

Retry on a transient without polluting the next turn:

for attempt in range(3):
    try:
        reply = await agent.achat(user_prompt)
        break
    except Exception:
        continue  # chat_history is guaranteed clean by the SDK

See Reliability Guarantees for the full set of default correctness behaviours.

---

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.

---

Related

Workflows

Build parallel and sequential multi-agent workflows

Async Crew Kickoff

Run YAML-defined crews asynchronously

Async Safety

Offloaded knowledge search, memory writes, and the thread-safe in-memory adapter

Cost Tracking

Async cost telemetry is now at parity with sync (#4887)

Spawn & Announce

Reliable async sub-agent spawn — no hangs, safe sync/async mixing