Skip to main content
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

MethodPurposeAsync Detection
aon_agent_startAgent session startDispatches via isinstance(store, AsyncConversationStore)
aon_user_messageUser message loggingDispatches via isinstance(store, AsyncConversationStore)
aon_agent_messageAgent response loggingDispatches via isinstance(store, AsyncConversationStore)
aon_tool_callTool execution loggingDispatches via isinstance(store, AsyncConversationStore)
aon_agent_endAgent session endDispatches 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

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()
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
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.now(timezone.utc).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