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.

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

Diagram

HostedAgent + Database Persistence

📝 User Input

🧠 Anthropic Infrastructure

💾 Your Database

✅ Session Resume

Quick Start

  1. Basic Usage

    Run gpt-4o-mini conversations 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")
  2. 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

Diagram
ManagedAgentDatabaseAnthropicHostedAgentUserManagedAgentDatabaseAnthropicHostedAgentUserInstance destroyed“Remember: X”Create session & executeResponse + usageStore messages & metadataSession persistedResume sessionLoad chat historyPrevious messagesContinue sessionContext-aware response“I remember X”

Database Backends

SQLite

Zero external dependencies, file-based storage:

python
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:

python
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:

bash
pip install psycopg2-binary
# Docker: docker run -d -p 5432:5432 -e POSTGRES_PASSWORD=pass postgres

MySQL

Popular relational database:

python
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:

bash
pip install mysql-connector-python
# Docker: docker run -d -p 3306:3306 -e MYSQL_ROOT_PASSWORD=pass mysql

Redis

State storage with conversation store:

python
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:

bash
pip install redis
# Docker: docker run -d -p 6379:6379 redis

MongoDB

Document database for state:

python
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:

bash
pip install pymongo
# Docker: docker run -d -p 27017:27017 mongo

ClickHouse

Analytics database with conversation store:

python
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:

bash
pip install clickhouse-connect
# Docker: docker run -d -p 8123:8123 clickhouse/clickhouse-server

JSON Files

Zero dependencies with built-in session store:

python
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.

Diagram

Metadata Persistence Flow

🤖 Agent Turn

💾 Save Messages

📊 Save Metadata

💥 Process Restart

🔄 New Process

📂 Load Messages

📈 Load Metadata

✅ Full Session Restored

What Survives Process Restarts

When using database persistence, these session components automatically survive crashes, restarts, and deployments:

ComponentPurposeRestored After Restart
Chat MessagesFull conversation history✅ Yes
Token Countstotal_input_tokens, total_output_tokens✅ Yes
Session HistoryPer-turn audit trail✅ Yes
Agent Identityagent_id for session continuity✅ Yes
Compute InstanceSandbox/runner references✅ Yes

Complete Session Resume Example

python
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 restored

Metadata Fields Preserved

The following metadata fields are automatically persisted and restored:

FieldTypeDescription
agent_idstrStable identity for session continuity
total_input_tokensintCumulative input tokens for cost tracking
total_output_tokensintCumulative output tokens for cost tracking
session_historyList[dict]Per-turn audit trail with actions and results
compute_instancestrSandbox/runner ID for compute reattachment

These fields enable cost tracking, usage analytics, and compute resource management across process boundaries.


Configuration Options

HostedAgent API Reference

Complete HostedAgent configuration options

PraisonDB Reference

Database adapter configuration options

ComponentPurposeKey Parameters
HostedAgentAnthropic execution backendprovider, config, api_key, timeout
HostedAgentConfigAgent definitionmodel, system, tools, packages
PraisonDBDatabase adapterdatabase_url, state_url, analytics_url
DbSessionAdapterSession bridgeAuto-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.

Diagram

Session Clearing Flow

🗑️ Clear Session

💾 Purge DB Rows

🔄 Empty Cache

✅ Clean Reload

Clear vs Delete Sessions

python
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.

MethodPurposeBehavior
clear_session(session_id)Empty the conversation but keep it re-creatablePurges DB messages, resets cache to empty
delete_session(session_id)Remove the session entirelyPurges DB messages and removes all metadata

Common Patterns

Session Resume Pattern

The most common pattern for persistent managed agents:

python
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:

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

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:

python
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, implement update_session_metadata or 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

Related