HostedAgent + Database Persistence
Run agents on Anthropic's infrastructure while persisting session data to your database
HostedAgent executes on Anthropic’s managed infrastructure while seamlessly persisting conversation history and session state to any of the registry-supported conversation and state backends. See the persistence overview for the full list.
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db
from praisonai import HostedAgent, HostedAgentConfig
hosted = HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini", system="You are helpful"),
)
agent = Agent(name="assistant", backend=hosted)
agent.start("Remember my project name is Orion.")The user chats with a hosted agent; each turn persists to your database so sessions resume after restarts.
Quick Start
- Basic Usage
Run
gpt-4o-miniconversations with SQLite persistence in 5 lines:python from praisonaiagents import Agent, MemoryConfig from praisonaiagents import db from praisonai import HostedAgent, HostedAgentConfig hosted = HostedAgent( provider="anthropic", config=HostedAgentConfig(model="gpt-4o-mini", system="You are helpful") ) agent = Agent( name="Assistant", backend=hosted, memory=MemoryConfig( db=db(database_url="conversation.db"), session_id="session-1", )) agent.start("Remember: The sky is blue") - Session Resume
Continue conversations after restarts:
python # Later process - same session_id resumes conversation agent2 = Agent( name="Assistant", backend=hosted, memory=MemoryConfig( db=db(database_url="conversation.db"), session_id="session-1", # Same ID = resume )) response = agent2.start("What color is the sky?") # Response: "The sky is blue" (remembers from previous session)
How It Works
Database Backends
SQLite
Zero external dependencies, file-based storage:
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db
from praisonai import HostedAgent, HostedAgentConfig
# Phase 1: First session (teach facts)
hosted = HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini", system="You are helpful")
)
agent = Agent(
name="Assistant",
backend=hosted,
memory=MemoryConfig(
db=db(database_url="sqlite:///my_data.db"),
session_id="learning-session",
))
agent.start("Remember: PraisonAI is an AI agent framework")
agent.start("Also remember: It supports multiple LLM providers")
# Phase 2: Direct verification
import sqlite3
conn = sqlite3.connect("my_data.db")
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM messages WHERE session_id = ?", ("learning-session",))
message_count = cursor.fetchone()[0]
print(f"Messages stored: {message_count}")
conn.close()
# Phase 3: Session resume (new instance)
hosted2 = HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini", system="You are helpful")
)
agent2 = Agent(
name="Assistant",
backend=hosted2,
memory=MemoryConfig(
db=db(database_url="sqlite:///my_data.db"),
session_id="learning-session", # Same ID resumes
))
result = agent2.start("What did I tell you about PraisonAI?")
# Result: "You told me that PraisonAI is an AI agent framework and that it supports multiple LLM providers."Prerequisites: None (built into Python)
PostgreSQL
Production-ready relational database:
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db
from praisonai import HostedAgent, HostedAgentConfig
import psycopg2
# Phase 1: First session
hosted = HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini", system="You are an expert assistant")
)
agent = Agent(
name="Expert",
backend=hosted,
memory=MemoryConfig(
db=db(database_url="postgresql://user:pass@localhost:5432/praisondb"),
session_id="expert-session",
))
agent.start("Remember: PostgreSQL is ACID compliant")
agent.start("Remember: It supports JSON columns")
# Phase 2: Direct verification
conn = psycopg2.connect("postgresql://user:pass@localhost:5432/praisondb")
cursor = conn.cursor()
cursor.execute("SELECT content FROM messages WHERE session_id = %s", ("expert-session",))
messages = cursor.fetchall()
print(f"Stored {len(messages)} messages")
conn.close()
# Phase 3: Resume session
hosted2 = HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini", system="You are an expert assistant")
)
agent2 = Agent(
name="Expert",
backend=hosted2,
memory=MemoryConfig(
db=db(database_url="postgresql://user:pass@localhost:5432/praisondb"),
session_id="expert-session",
))
result = agent2.start("What database features did I mention?")
# Result: "You mentioned that PostgreSQL is ACID compliant and supports JSON columns."Prerequisites:
pip install psycopg2-binary
# Docker: docker run -d -p 5432:5432 -e POSTGRES_PASSWORD=pass postgresMySQL
Popular relational database:
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db
from praisonai import HostedAgent, HostedAgentConfig
import mysql.connector
hosted = HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini", system="You are helpful")
)
agent = Agent(
name="Assistant",
backend=hosted,
memory=MemoryConfig(
db=db(database_url="mysql://user:pass@localhost:3306/praisondb"),
session_id="mysql-session",
))
agent.start("Remember: MySQL is owned by Oracle")
# Direct verification
conn = mysql.connector.connect(
host="localhost", user="user", password="pass", database="praisondb"
)
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM messages WHERE session_id = %s", ("mysql-session",))
count = cursor.fetchone()[0]
print(f"Messages: {count}")
conn.close()
# Resume later
agent2 = Agent(
name="Assistant",
backend=hosted,
memory=MemoryConfig(
db=db(database_url="mysql://user:pass@localhost:3306/praisondb"),
session_id="mysql-session",
))
result = agent2.start("Who owns MySQL?")
# Result: "MySQL is owned by Oracle."Prerequisites:
pip install mysql-connector-python
# Docker: docker run -d -p 3306:3306 -e MYSQL_ROOT_PASSWORD=pass mysqlRedis
State storage with conversation store:
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db
from praisonai import HostedAgent, HostedAgentConfig
import redis
hosted = HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini", system="You are helpful")
)
# Redis for state + SQLite for conversations
agent = Agent(
name="Assistant",
backend=hosted,
memory=MemoryConfig(
db=db(
database_url="sqlite:///conversations.db", # Conversation store
state_url="redis://localhost:6379" # State store
),
session_id="redis-session",
))
agent.start("Remember: Redis is in-memory")
# Direct verification
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
state_keys = r.keys("*redis-session*")
print(f"State keys: {len(state_keys)}")
# Resume session
agent2 = Agent(
name="Assistant",
backend=hosted,
memory=MemoryConfig(
db=db(
database_url="sqlite:///conversations.db",
state_url="redis://localhost:6379"
),
session_id="redis-session",
))
result = agent2.start("What type of database is Redis?")
# Result: "Redis is an in-memory database."Prerequisites:
pip install redis
# Docker: docker run -d -p 6379:6379 redisMongoDB
Document database for state:
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db
from praisonai import HostedAgent, HostedAgentConfig
import pymongo
hosted = HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini", system="You are helpful")
)
# MongoDB for state + SQLite for conversations
agent = Agent(
name="Assistant",
backend=hosted,
memory=MemoryConfig(
db=db(
database_url="sqlite:///conversations.db", # Conversation store
state_url="mongodb://localhost:27017/praisondb" # State store
),
session_id="mongo-session",
))
agent.start("Remember: MongoDB stores documents")
# Direct verification
client = pymongo.MongoClient("mongodb://localhost:27017/")
db_mongo = client["praisondb"]
state_count = db_mongo.state.count_documents({"session_id": "mongo-session"})
print(f"State documents: {state_count}")
# Resume session
agent2 = Agent(
name="Assistant",
backend=hosted,
memory=MemoryConfig(
db=db(
database_url="sqlite:///conversations.db",
state_url="mongodb://localhost:27017/praisondb"
),
session_id="mongo-session",
))
result = agent2.start("What does MongoDB store?")
# Result: "MongoDB stores documents."Prerequisites:
pip install pymongo
# Docker: docker run -d -p 27017:27017 mongoClickHouse
Analytics database with conversation store:
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db
from praisonai import HostedAgent, HostedAgentConfig
import clickhouse_connect
hosted = HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini", system="You are helpful")
)
# ClickHouse for analytics + SQLite for conversations
agent = Agent(
name="Assistant",
backend=hosted,
memory=MemoryConfig(
db=db(
database_url="sqlite:///conversations.db", # Conversation store
analytics_url="clickhouse://localhost:8123/default" # Analytics store
),
session_id="clickhouse-session",
))
agent.start("Remember: ClickHouse is columnar")
# Direct verification
client = clickhouse_connect.get_client(host="localhost", port=8123)
result = client.query("SELECT count() FROM analytics WHERE session_id = 'clickhouse-session'")
print(f"Analytics rows: {result.result_rows[0][0]}")
# Resume session
agent2 = Agent(
name="Assistant",
backend=hosted,
memory=MemoryConfig(
db=db(
database_url="sqlite:///conversations.db",
analytics_url="clickhouse://localhost:8123/default"
),
session_id="clickhouse-session",
))
result = agent2.start("What type of database is ClickHouse?")
# Result: "ClickHouse is a columnar database."Prerequisites:
pip install clickhouse-connect
# Docker: docker run -d -p 8123:8123 clickhouse/clickhouse-serverJSON Files
Zero dependencies with built-in session store:
from praisonaiagents import Agent, MemoryConfig
from praisonai import HostedAgent, HostedAgentConfig
from praisonaiagents.session.store import DefaultSessionStore
import json
import os
hosted = HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini", system="You are helpful")
)
# JSON file-based session store (no external deps)
store = DefaultSessionStore(directory="./sessions")
agent = Agent(
name="Assistant",
backend=hosted,
session_store=store,
memory=MemoryConfig(
session_id="json-session",
))
agent.start("Remember: JSON is human-readable")
# Direct verification
session_file = "./sessions/json-session.json"
if os.path.exists(session_file):
with open(session_file, 'r') as f:
data = json.load(f)
print(f"Messages stored: {len(data.get('messages', []))}")
# Resume session
agent2 = Agent(
name="Assistant",
backend=hosted,
session_store=DefaultSessionStore(directory="./sessions"),
memory=MemoryConfig(
session_id="json-session",
))
result = agent2.start("What format is JSON?")
# Result: "JSON is human-readable."Prerequisites: None (built into Python)
Session Metadata Persistence
HostedAgent automatically persists both chat messages and session metadata to your database, ensuring complete session recovery after process restarts.
What Survives Process Restarts
When using database persistence, these session components automatically survive crashes, restarts, and deployments:
| Component | Purpose | Restored After Restart |
|---|---|---|
| Chat Messages | Full conversation history | ✅ Yes |
| Token Counts | total_input_tokens, total_output_tokens | ✅ Yes |
| Session History | Per-turn audit trail | ✅ Yes |
| Agent Identity | agent_id for session continuity | ✅ Yes |
| Compute Instance | Sandbox/runner references | ✅ Yes |
Complete Session Resume Example
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db
from praisonai import HostedAgent, HostedAgentConfig
# First process: run a turn, metadata persisted automatically
agent = Agent(
name="Assistant",
backend=HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini")
),
memory=MemoryConfig(
db=db(database_url="conversation.db"),
session_id="session-1",
))
response = agent.start("Plan a 3-step research workflow on quantum computing")
# Check usage tokens (automatically tracked)
print(f"Tokens used: {agent.total_tokens}")
# Process exits. New process starts with same session_id.
agent2 = Agent(
name="Assistant",
backend=HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini")
),
memory=MemoryConfig(
db=db(database_url="conversation.db"),
session_id="session-1",
), # Same ID → resumes with full metadata
)
# All metadata restored: token totals, session history, agent_id, compute instance
response = agent2.start("Continue from step 2.")
print(f"Total tokens across restarts: {agent2.total_tokens}") # Cumulative count restoredMetadata Fields Preserved
The following metadata fields are automatically persisted and restored:
| Field | Type | Description |
|---|---|---|
agent_id | str | Stable identity for session continuity |
total_input_tokens | int | Cumulative input tokens for cost tracking |
total_output_tokens | int | Cumulative output tokens for cost tracking |
session_history | List[dict] | Per-turn audit trail with actions and results |
compute_instance | str | Sandbox/runner ID for compute reattachment |
These fields enable cost tracking, usage analytics, and compute resource management across process boundaries.
Configuration Options
HostedAgent API ReferenceComplete HostedAgent configuration options
PraisonDB ReferenceDatabase adapter configuration options
| Component | Purpose | Key Parameters |
|---|---|---|
HostedAgent | Anthropic execution backend | provider, config, api_key, timeout |
HostedAgentConfig | Agent definition | model, system, tools, packages |
PraisonDB | Database adapter | database_url, state_url, analytics_url |
DbSessionAdapter | Session bridge | Auto-configured based on database URL |
Clearing & Deleting Sessions
The DbSessionAdapter now properly purges persisted messages from the database, not just the in-memory cache, ensuring that cleared sessions stay clear even after restarts.
Clear vs Delete Sessions
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db
from praisonai.integrations.db_session_adapter import DbSessionAdapter
store = DbSessionAdapter(db(database_url="conversation.db"))
# Clear the conversation — also wipes persisted rows in the DB
store.clear_session("session-1")
assert store.get_chat_history("session-1") == []
# A brand-new instance after a restart starts empty — no stale history
store2 = DbSessionAdapter(db(database_url="conversation.db"))
assert store2.get_chat_history("session-1") == []
# Delete session completely — removes all data and metadata
store.delete_session("session-1")Privacy Guarantee: Cleared messages do not come back after a restart or when creating a new adapter instance. Both clear_session() and delete_session() now purge persisted messages from the underlying conversation store.
| Method | Purpose | Behavior |
|---|---|---|
clear_session(session_id) | Empty the conversation but keep it re-creatable | Purges DB messages, resets cache to empty |
delete_session(session_id) | Remove the session entirely | Purges DB messages and removes all metadata |
Common Patterns
Session Resume Pattern
The most common pattern for persistent managed agents:
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db
from praisonai import HostedAgent, HostedAgentConfig
def create_agent(session_id: str):
hosted = HostedAgent(
provider="anthropic",
config=HostedAgentConfig(
model="gpt-4o-mini",
system="You are a helpful assistant with perfect memory"
)
)
return Agent(
name="PersistentAgent",
backend=hosted,
memory=MemoryConfig(
db=db(database_url="postgresql://localhost/agentdb"),
session_id=session_id,
))
# First conversation
agent1 = create_agent("user-123")
agent1.start("My name is Alice and I live in Paris")
# Later conversation (different process/server restart)
agent2 = create_agent("user-123") # Same session_id
response = agent2.start("What's my name and where do I live?")
# Response: "Your name is Alice and you live in Paris."Multi-Backend Pattern
Use different backends for different data types:
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db
from praisonai import HostedAgent, HostedAgentConfig
hosted = HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini")
)
agent = Agent(
name="MultiBackendAgent",
backend=hosted,
memory=MemoryConfig(
db=db(
database_url="postgresql://localhost/conversations", # Conversations
state_url="redis://localhost:6379", # Fast state
analytics_url="clickhouse://localhost:8123/default" # Analytics
),
session_id="multi-backend-session",
))
# All data types are automatically stored in appropriate backends
agent.start("Analyze this data and remember the insights")Session ID Management
Best practices for session identification:
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db
from praisonai import HostedAgent, HostedAgentConfig
import hashlib
from datetime import datetime
def generate_session_id(user_id: str, conversation_type: str) -> str:
"""Generate deterministic session IDs"""
# Per-user, per-type sessions
base = f"{user_id}-{conversation_type}"
return hashlib.md5(base.encode()).hexdigest()[:16]
def get_daily_session_id(user_id: str) -> str:
"""Daily session rotation"""
today = datetime.now().strftime("%Y-%m-%d")
return f"{user_id}-{today}"
# Usage examples
user_session = generate_session_id("user-456", "support")
daily_session = get_daily_session_id("user-456")
hosted = HostedAgent(
provider="anthropic",
config=HostedAgentConfig(model="gpt-4o-mini")
)
agent = Agent(
name="SupportAgent",
backend=hosted,
memory=MemoryConfig(
db=db(database_url="sqlite:///support.db"),
session_id=user_session, # Consistent across requests
))Best Practices
Session Management
- Use meaningful session IDs (user-based, not random)
- Implement session rotation for long conversations
- Store session metadata for debugging
- Concurrent metadata writes are safe with
DefaultSessionStore(locked read-modify-write); for custom stores, implementupdate_session_metadataor your own equivalent to avoid stale-copy overwrites
Database Selection
- SQLite: Development, single-user apps, file-based persistence
- PostgreSQL: Production apps, complex queries, ACID compliance
- MySQL: Existing MySQL infrastructure, compatibility requirements
- Redis: High-speed state, session caching, temporary data
- MongoDB: Document-based state, flexible schemas
- ClickHouse: Analytics, large-scale logging, data warehousing
- JSON Files: Prototyping, zero dependencies, simple use cases
Performance Optimization
- Use connection pooling for database connections
- Implement message compaction for long sessions
- Cache frequently accessed session data
- Use async database operations when possible
- Monitor database performance metrics
Error Handling
- Implement retry logic for transient database failures
- Handle session corruption gracefully
- Log database errors for debugging
- Provide fallback behavior when persistence fails
- Test database connection before agent creation