Async Crew Kickoff

Run a PraisonAI crew from inside a running event loop — FastAPI, Jupyter, Discord bots — without asyncio.run().

Quick Start

FastAPI Route

The headline use case — running crews inside FastAPI routes without blocking the event loop:

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 in Jupyter cells where asyncio.run() would fail:

import praisonai

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

How It Works

Function	Method	When Used
Sync	`praisonai.run()`	Plain scripts, CLI usage
Async	`praisonai.arun()`	FastAPI, Jupyter, event loop contexts

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

When to use which

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

Prefer async entry point over wrapping sync

Use praisonai.arun() instead of wrapping the sync version:

import asyncio
import praisonai

# ✅ Good - proper async entry point
result = await praisonai.arun(agent_file="agents.yaml")

# ❌ Avoid - wrapping sync call (this won't work inside event loop)
# result = asyncio.run(praisonai.run(agent_file="agents.yaml"))

Use asyncio.gather for concurrent crews

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

import asyncio
import praisonai

# ✅ Concurrent execution
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 auto-detection works in async mode

All supported frameworks work transparently with async execution:

import praisonai

# Works with any framework - auto-detected or explicit
result = await praisonai.arun(agent_file="agents.yaml", framework="crewai")

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

Async Bridge

Lower-level async utilities and bridging functions

Framework Adapter Plugins

Custom framework adapters with async support

Documentation Index

​Quick Start

​How It Works

​Configuration Options

​When to use which

​Common Patterns

​FastAPI Background Task

​Concurrent Crew Execution

​Best Practices

​Related