Resource Lifecycle

Automatically clean up agents, shared memory, and connections using with or async with

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 point	Method	Description
with team:	__enter__ / __exit__	Sync — calls close() on exit
async with team:	__aenter__ / __aexit__	Async — prefers aclose() when available
Manual	team.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

Prefer context managers over manual close()

with and async with guarantee cleanup even when exceptions occur.

Cleanup is best-effort

close() logs warnings on failure but does not raise — cleanup failures will not mask the original error.

Do not reuse a team after exiting its block

Create a new PraisonAIAgents instance for each batch or request.

Create one team per request in servers

Avoid sharing a global workflow across HTTP requests — isolate resources per call.

---

Related

Advanced Memory System

Shared memory stores that benefit from automatic cleanup

MongoDB Memory

MongoDB-backed memory with connection lifecycle support