Skip to main content
Use with or async with on PraisonAIAgents to close agents, memory stores, and connections when a workflow finishes. 
from praisonaiagents import Agent, Task, PraisonAIAgents

researcher = Agent(name="Researcher", instructions="Research topics thoroughly")
task = Task(description="Research quantum computing", agent=researcher)

with PraisonAIAgents(agents=[researcher], tasks=[task]) as workflow:
    result = workflow.start()

print(result)
The user runs agents inside a context manager; leaving the block closes agents, memory stores, and connections automatically.

​
Quick Start

1

Simple Usage

from praisonaiagents import Agent, Task, PraisonAIAgents

researcher = Agent(name="Researcher", instructions="Research topics thoroughly")
task = Task(description="Research quantum computing", agent=researcher)

with PraisonAIAgents(agents=[researcher], tasks=[task]) as workflow:
    result = workflow.start()

print(result)
2

With Configuration

Async usage with shared memory inside a FastAPI endpoint:
import asyncio
from fastapi import FastAPI
from praisonaiagents import Agent, Task, PraisonAIAgents, ChromaMemory

app = FastAPI()

@app.post("/research")
async def research_topic(topic: str):
    researcher = Agent(name="Researcher", instructions="Research topics")
    task = Task(description=f"Research {topic}", agent=researcher)

    async with PraisonAIAgents(
        agents=[researcher],
        tasks=[task],
        shared_memory=ChromaMemory(),
    ) as workflow:
        result = await workflow.astart()

    return {"research": result}

​
How It Works

Entry pointMethodDescription
with team:__enter__ / __exit__Sync — calls close() on exit
async with team:__aenter__ / __aexit__Async — prefers aclose() when available
Manualteam.close()Explicit cleanup for long-running workers

​
Common Patterns

​
Shared memory cleanup

from praisonaiagents import ChromaMemory

with PraisonAIAgents(
    agents=[researcher],
    tasks=[task],
    shared_memory=ChromaMemory(),
) as workflow:
    result = workflow.start()
# ChromaDB connection closed here

​
Long-running worker shutdown

workflow = PraisonAIAgents(agents=[researcher])
try:
    workflow.add_task(Task(description="Research AI", agent=researcher))
    workflow.start()
finally:
    workflow.close()

​
Per-request isolation in servers

@app.post("/analyse")
async def analyse(data: str):
    agent = Agent(name="Analyzer", instructions="Analyse data")
    async with PraisonAIAgents(agents=[agent]) as workflow:
        return await workflow.astart()

​
Best Practices

with and async with guarantee cleanup even when exceptions occur.
close() logs warnings on failure but does not raise — cleanup failures will not mask the original error.
Create a new PraisonAIAgents instance for each batch or request.
Avoid sharing a global workflow across HTTP requests — isolate resources per call.

Advanced Memory System

Shared memory stores that benefit from automatic cleanup

MongoDB Memory

MongoDB-backed memory with connection lifecycle support