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.

python

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.

Diagram

Quick Start

Simple Usage

python

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)

With Configuration

Async usage with shared memory inside a FastAPI endpoint:

python

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

Diagram

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

python

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

python

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

Per-request isolation in servers

python

@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()

HTTP invoke API — no manual cleanup needed

python

# In an n8n / HTTP invoke deployment you do NOT need to wrap invokes
# in with / try/finally — the wrapper closes each per-request clone.
# Just register the template once and serve.

from praisonaiagents import Agent
from praisonai.api.agent_invoke import register_agent

register_agent("assistant", Agent(name="assistant", instructions="…"))
# praisonai serve agents.yaml — every POST /invoke closes its clone.

See n8n API → Per-request cleanup.