Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.praison.ai/llms.txt

Use this file to discover all available pages before exploring further.

Async DB hooks enable non-blocking database operations in async agents through automatic async/sync detection and asyncio.to_thread fallback.

​
Quick Start

1

Async Context Manager

Use async DB operations with the async context manager for automatic cleanup.
from praisonaiagents import Agent, PraisonAIDB

async def main():
    async with PraisonAIDB("sqlite:///agent_data.db") as db:
        # Store user message
        await db.aon_user_message(
            session_id="chat_123",
            content="Hello, how can you help me?",
            metadata={"user_id": "user_456"}
        )
        
        # Store agent message  
        await db.aon_agent_message(
            session_id="chat_123", 
            content="I can help with various tasks.",
            metadata={"agent_name": "assistant"}
        )
2

Wire to Async Agent

Connect async DB hooks to an async agent for seamless persistence.
from praisonaiagents import Agent

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    db=PraisonAIDB("postgresql://user:pass@host/db"),
    async_mode=True
)

async def chat():
    async with agent.db:
        response = await agent.run_async("What's the weather like?")
        print(response)

​
How It Works

MethodPurposeAsync Detection
aon_agent_startAgent session startChecks async_set_agent_state
aon_user_messageUser message loggingChecks async_add_message
aon_agent_messageAgent response loggingChecks async_add_message
aon_tool_callTool execution loggingChecks async_add_tool_call
aon_agent_endAgent session endChecks async_set_agent_state

​
Configuration Options

All async hooks support these signatures from the DB adapter: 
async def aon_agent_start(
    self, 
    session_id: str, 
    name: str, 
    agent_id: str = "", 
    metadata: Optional[Dict[str, Any]] = None
) -> None

async def aon_user_message(
    self, 
    session_id: str, 
    content: str, 
    metadata: Optional[Dict[str, Any]] = None
) -> None

async def aon_agent_message(
    self, 
    session_id: str, 
    content: str, 
    metadata: Optional[Dict[str, Any]] = None
) -> None

async def aon_tool_call(
    self, 
    session_id: str, 
    tool_name: str, 
    arguments: Dict[str, Any], 
    result: Any = None, 
    metadata: Optional[Dict[str, Any]] = None
) -> None

async def aon_agent_end(
    self, 
    session_id: str, 
    name: str, 
    agent_id: str = "", 
    metadata: Optional[Dict[str, Any]] = None
) -> None

async def aclose(self) -> None

​
Common Patterns

​
Complete Async Lifecycle

async def agent_lifecycle():
    async with PraisonAIDB("sqlite:///lifecycle.db") as db:
        session_id = "session_789"
        
        # Start agent session
        await db.aon_agent_start(
            session_id=session_id,
            name="DataAgent", 
            agent_id="agent_123"
        )
        
        # User interaction
        await db.aon_user_message(
            session_id=session_id,
            content="Analyze this data",
            metadata={"timestamp": "2024-01-01T10:00:00Z"}
        )
        
        # Tool execution
        await db.aon_tool_call(
            session_id=session_id,
            tool_name="data_analyzer",
            arguments={"file": "data.csv"},
            result={"rows": 1000, "columns": 5}
        )
        
        # Agent response
        await db.aon_agent_message(
            session_id=session_id,
            content="Analysis complete: 1000 rows, 5 columns found."
        )
        
        # End session
        await db.aon_agent_end(
            session_id=session_id,
            name="DataAgent",
            agent_id="agent_123"
        )

​
Sync Store Compatibility

# Works with existing sync stores
class MySQLStore:
    def add_message(self, content, metadata):
        # Sync implementation
        pass
    
    # No async_add_message - will use asyncio.to_thread

async with PraisonAIDB(MySQLStore()) as db:
    # Still works - automatically wrapped
    await db.aon_user_message("session_1", "Hello")

​
Native Async Store

# Implement async methods for true async performance
class AsyncPostgreStore:
    async def async_add_message(self, content, metadata):
        async with self.pool.acquire() as conn:
            await conn.execute("INSERT INTO messages...", content)
    
    def add_message(self, content, metadata):
        # Sync fallback if needed
        pass

# Will use native async methods
async with PraisonAIDB(AsyncPostgreStore()) as db:
    await db.aon_user_message("session_1", "Hello")  # True async

​
Best Practices

The async context manager ensures proper connection cleanup and transaction handling:
# ✅ Good: Automatic cleanup
async with PraisonAIDB(store) as db:
    await db.aon_user_message("session", "Hello")

# ❌ Bad: Manual cleanup required  
db = PraisonAIDB(store)
await db.aon_user_message("session", "Hello")
await db.aclose()  # Must remember to call
For high-throughput async applications, implement native async methods:
class OptimizedStore:
    # ✅ Native async - no thread overhead
    async def async_add_message(self, content, metadata):
        await self.async_conn.execute(query, content)
    
    # ✅ Sync fallback for compatibility
    def add_message(self, content, metadata):
        self.sync_conn.execute(query, content)
All hooks accept optional metadata dictionaries:
metadata = {
    "user_id": "user_123",
    "session_type": "chat",
    "timestamp": datetime.utcnow().isoformat(),
    "environment": "production"
}

await db.aon_user_message(
    session_id="session_456", 
    content="User input",
    metadata=metadata
)
Both sync and async context managers are supported:
# Sync context (for mixed sync/async code)
with PraisonAIDB(store) as db:
    # Use sync methods
    db.on_user_message("session", "Hello")

# Async context (for async agents)  
async with PraisonAIDB(store) as db:
    # Use async methods
    await db.aon_user_message("session", "Hello")

Persistence Overview

Complete persistence system documentation

Agent Architecture

Learn about async agent patterns