> ## Documentation Index
> Fetch the complete documentation index at: https://docs.praison.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Async db hooks
from datetime import datetime, timezone
***
title: "Async DB Hooks"
sidebarTitle: "Async DB Hooks"
description: "Async database lifecycle hooks for non-blocking persistence in async agents"
icon: "database"
----------------
Async DB hooks enable non-blocking database operations in async agents through automatic async/sync detection and `asyncio.to_thread` fallback.
```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 input fill:#8B0000,stroke:#7C90A0,color:#fff
classDef process fill:#189AB4,stroke:#7C90A0,color:#fff
classDef output fill:#10B981,stroke:#7C90A0,color:#fff
class A input
class B,C,E process
class D,F,G output
```
## 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, 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
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
```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: Check for async_add_message()
alt Native async support
Adapter->>Store: await async_add_message()
Store-->>Adapter: Result
else Sync fallback
Adapter->>Store: asyncio.to_thread(add_message)
Store-->>Adapter: Result
end
Adapter-->>Agent: Success
```
| Method | Purpose | Async Detection |
| ------------------- | ---------------------- | ------------------------------ |
| `aon_agent_start` | Agent session start | Checks `async_set_agent_state` |
| `aon_user_message` | User message logging | Checks `async_add_message` |
| `aon_agent_message` | Agent response logging | Checks `async_add_message` |
| `aon_tool_call` | Tool execution logging | Checks `async_add_tool_call` |
| `aon_agent_end` | Agent session end | Checks `async_set_agent_state` |
***
## Configuration Options
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,
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
```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
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
```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
The async context manager ensures proper connection cleanup and transaction handling:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# ✅ Good: Automatic cleanup
async with PraisonAIDB(store) as db:
await db.aon_user_message("session", "Hello")
# ❌ Bad: Manual cleanup required
db = PraisonAIDB(store)
await db.aon_user_message("session", "Hello")
await db.aclose() # Must remember to call
```
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