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

1. 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"}
           )
   
2. 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)

Shared dispatch for both entry points

Both persistence entry points flow through the same async-store dispatch.

As of PR #4394, the async-store dispatch (isinstance(store, AsyncConversationStore) → await store.method(...) vs. asyncio.to_thread(store.method, ...)) lives on PraisonAIDB, in its internal _call_store / _dispatch_async helpers.

PersistenceOrchestrator.on_message / aon_message / on_agent_end / aon_agent_end delegate to those helpers instead of keeping a second copy, so the async/sync-store behaviour is identical no matter which entry point you pick — MemoryConfig(db=PraisonAIDB(...)) or the orchestrator-based wrap_agent_with_persistence / PersistentAgent / create_persistent_session.

_call_store splits by op kind: writes stay fire-and-forget on the shared bridge (idempotent, uuid-keyed); reads (_READ_OPS = {"get", "get_session", "get_messages", "list_sessions"}) route through run_sync_or_offload, returning the real value on a sync path and raising loudly inside a running loop. As of PR #4861, _call_store and flush_pending_writes are instance methods on PraisonAIDB (previously staticmethods), and each instance tracks its own fire-and-forget writes in self._bg_writes — not a process-global set. _call_store, _dispatch_async, and _READ_OPS are internal plumbing on PraisonAIDB, not public API. They are named here only to describe the shared dispatch mechanism.

Because bg-write tracking is per-instance (PR #4861), a tenant’s close() / aclose() flushes only its own in-flight writes — one tenant’s stuck backend can no longer consume another tenant’s 5-second flush budget or discard its writes. See Thread Safety.

Sync reads from inside a running loop

Sync reads reached from inside a running event loop now fail loudly instead of silently dropping the read.

Sync reads from inside a running loop now fail loudly. Calling a sync hook whose store op is a read — get, get_session, get_messages, list_sessions, or any sync completion hook (on_run_end, on_trace_end, on_span_end) that does a read-modify-write — from inside a running event loop raises RuntimeError steering you to the async surface. In previous releases it silently returned None: a resumed session looked empty, and the completion hooks merged into {} and overwrote the persisted record with a partial dict. Outside a loop, sync reads resolve the coroutine transparently. Writes are unaffected — they remain uuid-keyed fire-and-forget on the bridge.

Fixed in PR #4821, fixes #4820.

Which hook to call from where

Pick sync or async by whether you sit inside a running loop and whether you read or write.

Troubleshooting. Symptom in older versions: a resumed session came back empty even though messages had been persisted; a completed run/trace/span record lost run_id, started_at, input_content, or its spans/events after a sync completion hook fired from async code. Root cause: sync reads reached from a running loop were silently dropped. Fixed in PR #4821 — the same call now raises RuntimeError pointing you at the aon_* hook.

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

---

Related

Persistence Overview

Complete persistence system documentation

Agent Architecture

Learn about async agent patterns