Async Crew Kickoff

from praisonaiagents import Agent
import asyncio

agent = Agent(name="async-agent", instructions="Process tasks asynchronously.")

async def main():
    result = await agent.astart("Run this task asynchronously.")
    print(result)

asyncio.run(main())

Run PraisonAI crews with native async execution from FastAPI, Jupyter, Discord bots, and other event loop contexts. The user awaits agent.astart() from an async app; the crew kickoff runs natively on the event loop without thread offload.

Quick Start

FastAPI Route

Native async execution — no worker threads, true cooperative multitasking:

from fastapi import FastAPI
import praisonai

app = FastAPI()

@app.post("/run")
async def run_crew():
    result = await praisonai.arun(agent_file="agents.yaml")
    return {"result": result}

Jupyter Notebook

Works directly in async cells without blocking:

import praisonai

result = await praisonai.arun(agent_file="agents.yaml")
print(result)

How It Works

Non-blocking startup

Config loading, adapter setup, and workflow preparation run in a worker thread via asyncio.to_thread, so a slow disk read or heavy adapter import never stalls the event loop. The _aload_config and _aprepare_for_run helpers wrap the blocking open() / yaml.safe_load / adapter setup calls, keeping the loop free to serve other requests while startup work runs off-thread.

Sync/Async Parity: As of PR #1870, sync and async kickoff paths share the same prep logic (AutoGen version selection, AgentOps init, cli_backend validation), so behavior is identical between generate_crew_and_kickoff() and agenerate_crew_and_kickoff().

Extended in PR #2738: Config validation, merge, and dump logic live in a single _build_yaml_workflow builder shared by both paths, so sync and async behavior can no longer drift apart.

PR #3252 (July 2026) extended parity to the process: workflow path — observability, cli_backend validation, and tool_timeout diagnostics now fire on workflow YAML for both sync and async, matching the sequential / hierarchical treatment.As of PR #3963, an explicit tool_timeout on process: workflow now raises on both paths rather than logging a warning — pick sequential or hierarchical if you need per-tool timeouts. See tool_timeout on workflow YAML.

What’s actually async

Adapter	Async path	Notes
`praisonai` (praisonaiagents)	Native — `AgentTeam.astart()`	True cooperative async
`crewai`	Thread offload (default fallback)	Until CrewAI exposes async
`autogen` / `ag2`	Thread offload (default fallback)	Adapter-specific implementation

Workflow mode

YAML files with process: workflow also run natively async via YAMLWorkflowParser + workflow.astart() — no extra configuration needed.

Configuration Options

Option	Type	Default	Description
`agent_file`	`str`	Required	Path to the agent YAML file
`framework`	`str`	`None`	Framework to use (auto-detected if None)
`tools`	`list`	`None`	Additional tools to make available
`agent_yaml`	`str`	`None`	Direct YAML content as string
`cli_config`	`dict`	`None`	CLI configuration overrides (explicit keys win over `**kwargs`)
`model` / `llm`	`str`	`None`	Selects the LLM — same as the CLI’s `--llm` flag
`session`	`str`	`None`	Resume a session — same as the CLI’s `--resume` flag
`**kwargs`	any	—	Any other keyword argument is forwarded through `cli_config` to the generator, mirroring the CLI’s pass-through

See Advanced CLI options from Python for the full alias table and precedence rules.

Common Patterns

FastAPI Background Task

import asyncio
import praisonai
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

async def run_crew_background():
    result = await praisonai.arun(agent_file="agents.yaml")
    # Store result in database, send notification, etc.
    print(f"Background crew completed: {result}")

@app.post("/start-crew")
async def start_crew(background_tasks: BackgroundTasks):
    background_tasks.add_task(run_crew_background)
    return {"message": "Crew started"}

Concurrent Crew Execution

import asyncio
import praisonai

async def run_multiple_crews():
    # Run all crews concurrently with true async
    results = await asyncio.gather(
        praisonai.arun(agent_file="research.yaml"),
        praisonai.arun(agent_file="analysis.yaml"),
        praisonai.arun(agent_file="summary.yaml")
    )
    
    return results

CLI options from an async handler

arun() (and run()) accept any extra keyword argument the CLI accepts. model=/llm= and session= get friendly Python aliases; everything else is forwarded through the same cli_config bridge the CLI already uses.

import praisonai

async def handler():
    # model= is aliased to --llm; session= is aliased to --resume
    result = await praisonai.arun(
        "agents.yaml",
        model="gpt-4o",
        session="chat-42",
    )
    return result

An explicit cli_config= still wins over loose kwargs — pass either, not both, for the same key.

Best Practices

Cancellation and timeouts propagate

Under native async, asyncio.CancelledError and asyncio.wait_for now actually cancel SDK work instead of being trapped behind a worker thread:

import asyncio
import praisonai

async def cancelable_crew():
    try:
        # This will properly cancel if timeout is reached
        result = await asyncio.wait_for(
            praisonai.arun(agent_file="agents.yaml"),
            timeout=30.0
        )
        return result
    except asyncio.TimeoutError:
        print("Crew execution timed out and was cancelled")
        return None

Use asyncio.gather for concurrent crews

When running multiple crews, use asyncio.gather for parallel execution:

import asyncio
import praisonai

# ✅ Concurrent execution with native async
crew1_task = praisonai.arun(agent_file="crew1.yaml")
crew2_task = praisonai.arun(agent_file="crew2.yaml")
results = await asyncio.gather(crew1_task, crew2_task)

# ❌ Sequential execution (slower)
result1 = await praisonai.arun(agent_file="crew1.yaml")
result2 = await praisonai.arun(agent_file="crew2.yaml")

Framework detection works transparently

All supported frameworks work with async execution — praisonai-native uses true async, others fall back to thread offload automatically:

import praisonai

# Native async for praisonai, thread offload for crewai
result = await praisonai.arun(agent_file="agents.yaml", framework="praisonai")

Handle errors gracefully in async contexts

Wrap async crew execution in try-catch blocks:

import asyncio
import logging
import praisonai

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def safe_crew_run():
    try:
        result = await praisonai.arun(agent_file="agents.yaml")
        return {"success": True, "result": result}
    except Exception as e:
        logger.error(f"Crew execution failed: {e}")
        return {"success": False, "error": str(e)}

YAML Template Variables

Use placeholders safely alongside JSON literals

Framework Adapter Plugins

Custom framework adapters with async support

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)

Async Crew Kickoff

Quick Start

How It Works

Non-blocking startup

What’s actually async

Workflow mode

Configuration Options

Common Patterns

FastAPI Background Task

Concurrent Crew Execution

CLI options from an async handler

Best Practices

YAML Template Variables

Framework Adapter Plugins

​Quick Start

​How It Works

​Non-blocking startup

​What’s actually async

​Workflow mode

​Configuration Options

​Common Patterns

​FastAPI Background Task

​Concurrent Crew Execution

​CLI options from an async handler

​Best Practices

​Related

YAML Template Variables

Framework Adapter Plugins

Quick Start

How It Works

Non-blocking startup

What’s actually async

Workflow mode

Configuration Options

Common Patterns

FastAPI Background Task

Concurrent Crew Execution

CLI options from an async handler

Best Practices

Related