Async DB Hooks

Async database lifecycle hooks for non-blocking persistence in async agents

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

from praisonaiagents import Agent, PraisonAIDB

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    db=PraisonAIDB("sqlite:///agent_data.db"),
    async_mode=True,
)

# In async code: await agent.achat("Remember my preferences")

Breaking Change in PR #1829: The async_* prefixed methods have been removed from async stores. The orchestrator now uses isinstance(store, AsyncConversationStore) for dispatch instead of runtime method introspection.

# Before (removed)
session = await store.async_get_session(session_id)

# After (current)
session = await store.get_session(session_id)

The user chats asynchronously; DB hooks persist messages without blocking the event loop.

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	Dispatches via isinstance(store, AsyncConversationStore)
aon_user_message	User message logging	Dispatches via isinstance(store, AsyncConversationStore)
aon_agent_message	Agent response logging	Dispatches via isinstance(store, AsyncConversationStore)
aon_tool_call	Tool execution logging	Dispatches via isinstance(store, AsyncConversationStore)
aon_agent_end	Agent session end	Dispatches via isinstance(store, AsyncConversationStore)

---

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

Prefer async with for exception-safety

Manual aclose() is now safe and idempotent — calling it twice does nothing harmful. Prefer async with for exception-safety and readability, so cleanup runs even when an error is raised mid-block:

# ✅ Good: cleanup runs even on error
async with PraisonAIDB(store) as db:
    await db.aon_user_message("session", "Hello")

# ✅ Also fine: aclose() is idempotent and safe to call
db = PraisonAIDB(store)
await db.aon_user_message("session", "Hello")
await db.aclose()

Reusing a PraisonAIDB after close

close() and aclose() reset the internal stores, so re-entering a with db: ... block after close cleanly re-initializes instead of dispatching to closed handles:

db = PraisonAIDB("sqlite:///agent.db")
with db:
    db.on_user_message("session", "Hello")   # first use
# ...later in the same process...
with db:
    db.on_user_message("session", "Later")   # cleanly re-initializes

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