Async Crew Kickoff

Run PraisonAI crews with native async execution — no thread offload required

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.

Each agenerate_crew_and_kickoff() run opens an async_scoped_bridge() around the adapter call, so concurrent tenants each get their own per-run bridge and a stuck coroutine in one cannot park the shared loop for the others.

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

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.

What’s actually async

Workflow mode

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

Configuration Options

strict_validation and tool_timeout_workers are recognised loose keys — **kwargs forwards them through cli_config for per-run multi-tenant isolation.

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

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

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

import praisonai

# Native async for both praisonai and crewai (crewai ≥ 0.28)
result = await praisonai.arun(agent_file="agents.yaml", framework="crewai")

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

Related