Skip to main content
Run PraisonAI crews with native async execution from FastAPI, Jupyter, Discord bots, and other event loop contexts.

​
Quick Start

1

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}
2

Jupyter Notebook

Works directly in async cells without blocking:
import praisonai

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

​
How It Works

​
What’s actually async

AdapterAsync pathNotes
praisonai (praisonaiagents)Native — AgentTeam.astart()True cooperative async
crewaiThread offload (default fallback)Until CrewAI exposes async
autogen / ag2Thread 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

OptionTypeDefaultDescription
agent_filestrRequiredPath to the agent YAML file
frameworkstrNoneFramework to use (auto-detected if None)
toolslistNoneAdditional tools to make available
agent_yamlstrNoneDirect YAML content as string
cli_configdictNoneCLI configuration overrides

​
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

​
Best Practices

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