HostedAgent + Database Persistence

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

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

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

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)

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

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

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

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

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

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

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

​Quick Start

​How It Works

​Database Backends

​Session Metadata Persistence

​What Survives Process Restarts

​Complete Session Resume Example

​Metadata Fields Preserved

​Configuration Options

HostedAgent API Reference

PraisonDB Reference

​Clearing & Deleting Sessions

​Clear vs Delete Sessions

​Common Patterns

​Best Practices

​Related

Hosted Agent

Local Agent

Managed CLI

Session Management

Quick Start

How It Works

Database Backends

Session Metadata Persistence

What Survives Process Restarts

Complete Session Resume Example

Metadata Fields Preserved

Configuration Options

Clearing & Deleting Sessions

Clear vs Delete Sessions

Common Patterns

Best Practices

Related