Async db hooks

from datetime import datetime, timezone

---

title: “Async DB Hooks” sidebarTitle: “Async DB Hooks” description: “Async database lifecycle hooks for non-blocking persistence in async agents” icon: “database”

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

Method	Purpose	Async Detection
aon_agent_start	Agent session start	Checks async_set_agent_state
aon_user_message	User message logging	Checks async_add_message
aon_agent_message	Agent response logging	Checks async_add_message
aon_tool_call	Tool execution logging	Checks async_add_tool_call
aon_agent_end	Agent session end	Checks 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

Always use async with for cleanup

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

Implement async_* methods for performance

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)

Handle metadata consistently

All hooks accept optional metadata dictionaries:

metadata = {
    "user_id": "user_123",
    "session_type": "chat",
    "timestamp": datetime.now(timezone.utc).isoformat(),
    "environment": "production"
}

await db.aon_user_message(
    session_id="session_456", 
    content="User input",
    metadata=metadata
)

Use sync context manager for mixed environments

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

---

Related

Persistence Overview

Complete persistence system documentation

Agent Architecture

Learn about async agent patterns