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.

python

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.

Diagram

Quick Start

Simple Usage

Use astart() inside an async function:

python

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:

python

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

Diagram

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
`team.start()` / `team.run()`	`await team.astart()` / `await team.arun()`	Required when any `Task` sets `async_execution=True` — only the async entry point fans them out in parallel

Configuration Options

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

python

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

Async execution requires the async entry point

async_execution=True marks a task as parallelisable, but only await team.astart() / await team.arun() actually runs it in parallel. The sync path runs it sequentially and warns.

Diagram

python

import asyncio
from praisonaiagents import Agent, Task, PraisonAIAgents

r1 = Agent(name="R1", instructions="Research AI")
r2 = Agent(name="R2", instructions="Research ML")

team = PraisonAIAgents(
    agents=[r1, r2],
    tasks=[
        Task(description="Research AI", agent=r1, async_execution=True),
        Task(description="Research ML", agent=r2, async_execution=True),
    ],
    process="workflow",
)

async def main():
    await team.astart()   # parallel fan-out — sync .start() would run sequentially and warn

asyncio.run(main())

process="hierarchical" does not advertise parallel fan-out, so the warning never fires there.

Common Patterns

Parallel Requests

python

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

python

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

python

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.

Diagram

Inside a Web Framework (FastAPI)

python

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

python

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.

python

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

Diagram

Retry on a transient without polluting the next turn:

python

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.

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

Async Agents

Quick Start

How It Works

Sync vs Async Methods

Configuration Options

Async execution requires the async entry point

Common Patterns

Parallel Requests

Async Callback

Failure cascade in asequential()

Inside a Web Framework (FastAPI)

Lifecycle Hooks in Async Workflows

Failed turns leave chat_history clean

Best Practices

Related

Failure cascade in `asequential()`

Failed turns leave `chat_history` clean