> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# 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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
subgraph "Async DB Hook Flow"
A[📋 Async Agent] --> B[🔍 DB Adapter]
B --> C{Native async?}
C -->|Yes| D[⚡ async_* method]
C -->|No| E[🔄 asyncio.to_thread]
E --> F[📝 sync method]
D --> G[✅ Result]
F --> G
end
classDef agent fill:#8B0000,color:#fff
classDef tool fill:#189AB4,color:#fff
class A agent
class B,C,D,E,F,G tool
```
## Quick Start
Use async DB operations with the async context manager for automatic cleanup.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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"}
)
```
Connect async DB hooks to an async agent for seamless persistence.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant Agent as Async Agent
participant Adapter as DB Adapter
participant Store as Storage Backend
Agent->>Adapter: aon_user_message()
Adapter->>Adapter: isinstance(store, AsyncConversationStore)
alt Native async support
Adapter->>Store: await store.add_message()
Store-->>Adapter: Result
else Sync fallback
Adapter->>Store: asyncio.to_thread(store.add_message)
Store-->>Adapter: Result
end
Adapter-->>Agent: Success
```
| 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](https://github.com/MervinPraison/PraisonAI/pull/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](https://github.com/MervinPraison/PraisonAI/pull/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](https://github.com/MervinPraison/PraisonAI/pull/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](/docs/features/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](https://github.com/MervinPraison/PraisonAI/pull/4821), fixes [#4820](https://github.com/MervinPraison/PraisonAI/issues/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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
Q{Am I inside
a running loop?}
Q -->|No| Sync[✅ Use sync hook
on_agent_start / on_message]
Q -->|Yes, and I'm reading| Async[⚡ Use async hook
aon_agent_start / aon_message]
Q -->|Yes, and I'm writing| SyncOK[✅ Sync write is safe
fire-and-forget on bridge]
classDef question fill:#F59E0B,stroke:#7C90A0,color:#fff
classDef good fill:#10B981,stroke:#7C90A0,color:#fff
classDef async fill:#189AB4,stroke:#7C90A0,color:#fff
class Q question
class Sync,SyncOK good
class Async async
```
**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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# ✅ 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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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
Complete persistence system documentation
Learn about async agent patterns