HostedAgent + Database Persistence

Run agents on Anthropic’s infrastructure while persisting session data to your database

ManagedAgent is deprecated as of PR #1550. New code should use HostedAgent for Anthropic-hosted runs (this page) or LocalAgent for local loops. Existing imports continue to work but emit a DeprecationWarning for non-Anthropic providers.

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

When context compaction is enabled, managed persistence surfaces the compaction checkpoint automatically — resume replays the compacted working history (summary + tail). See Compacted Session Resume.

Quick Start

1. Basic Usage

   Run gpt-4o-mini conversations with SQLite persistence in 5 lines:

   from praisonaiagents import Agent, 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,
       db=db(database_url="conversation.db"),
       session_id="session-1"
   )
   
   agent.start("Remember: The sky is blue")
   
2. Session Resume

   Continue conversations after restarts:

   # Later process - same session_id resumes conversation
   agent2 = Agent(
       name="Assistant",
       backend=hosted,
       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, 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,
    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,
    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, 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,
    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, 
    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 postgres

MySQL

Popular relational database:

from praisonaiagents import Agent, 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,
    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,
    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 mysql

Redis

State storage with conversation store:

from praisonaiagents import Agent, 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,
    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,
    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 redis

MongoDB

Document database for state:

from praisonaiagents import Agent, 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,
    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,
    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 mongo

ClickHouse

Analytics database with conversation store:

from praisonaiagents import Agent, 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,
    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,
    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-server

JSON Files

Zero dependencies with built-in session store:

from praisonaiagents import Agent
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,
    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"),
    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, 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")
    ),
    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")
    ),
    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:

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 Reference

Complete HostedAgent configuration options

PraisonDB Reference

Database 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, 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, 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,
        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, db
from praisonai import HostedAgent, HostedAgentConfig

hosted = HostedAgent(
    provider="anthropic",
    config=HostedAgentConfig(model="gpt-4o-mini")
)

agent = Agent(
    name="MultiBackendAgent",
    backend=hosted,
    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, 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,
    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

Hosted Agent

Run entire agent loops on Anthropic’s managed runtime

Local Agent

Run agent loops locally with any LLM

Managed CLI

Terminal commands for managing Anthropic-hosted resources

Session Management

Advanced session handling techniques