Async DB Hooks

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, MemoryConfig, PraisonAIDB

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    memory=MemoryConfig(
        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

Async Context Manager

Use async DB operations with the async context manager for automatic cleanup.

from praisonaiagents import Agent, MemoryConfig, 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"}
        )

Wire to Async Agent

Connect async DB hooks to an async agent for seamless persistence.

from praisonaiagents import Agent, MemoryConfig

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    memory=MemoryConfig(
        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)

Sync hook against an async-mode store

Calling a sync hook against a store opened with mode="async" now fails clearly instead of crashing deep inside the store.

Calling the sync on_agent_start against a store opened with mode="async" from inside a running event loop now raises a clear RuntimeError telling you to use the async surface (previously it crashed with the opaque TypeError: 'coroutine' object is not iterable). Called outside a loop, sync now resolves the coroutine transparently.

State-store lifecycle key

The wrapper writes both start and end under agent:{session_id}, so the end transition updates the same record the start wrote instead of a disconnected one. Start previously used agent:{session_id}:{agent_id or name} and end wrote a separate record. Consumers who query the state store directly must update their key format to agent:{session_id}.

Configuration Options

Signature change in PR #3857: aon_agent_start(agent_name, session_id, user_id, metadata) -> List and aon_agent_end(session_id, metadata). user_id is now persisted (previously silently dropped) and aon_agent_start returns the resumed message list. Update any custom AsyncDbAdapter implementations to match.

All async hooks support these signatures from the DB adapter:

async def aon_agent_start(
    self,
    agent_name: str,
    session_id: str,
    user_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
) -> List   # previous messages for resume; empty list on new session

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,
    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 — agent_name is first, user_id is persisted
        messages = await db.aon_agent_start(
            "DataAgent",           # agent_name (now first)
            session_id,
            user_id="user_456",    # new; previously silently dropped
            metadata={"env": "prod"},
        )
        
        # 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 — no longer takes name / agent_id
        await db.aon_agent_end(session_id, metadata={"outcome": "ok"})

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

Persistence Overview

Complete persistence system documentation

Agent Architecture

Learn about async agent patterns

Getting Started

Install

Core Concepts

Guides

Features

Models

Databases

Observability

Memory

Telemetry & Analytics

Knowledge

RAG

Persistence

Tools

Other Features

Developers

Configuration

Best Practices

Getting Started (No Code)

Quick Start

How It Works

Sync hook against an async-mode store

State-store lifecycle key

Configuration Options

Common Patterns

Complete Async Lifecycle

Sync Store Compatibility

Native Async Store

Best Practices

Persistence Overview

Agent Architecture

​Quick Start

​How It Works

​Sync hook against an async-mode store

​State-store lifecycle key

​Configuration Options

​Common Patterns

​Complete Async Lifecycle

​Sync Store Compatibility

​Native Async Store

​Best Practices

​Related

Persistence Overview

Agent Architecture

Quick Start

How It Works

Sync hook against an async-mode store

State-store lifecycle key

Configuration Options

Common Patterns

Complete Async Lifecycle

Sync Store Compatibility

Native Async Store

Best Practices

Related