Mervin Praison

Universal database persistence for ManagedAgent with automatic session resume across 7 backends — SQLite, PostgreSQL, MySQL, Redis, MongoDB, ClickHouse, and JSON files. Uses the simplified DB import for maximum developer ergonomics.

%%{init: {"theme": "base", "themeVariables": {"background": "transparent", "lineColor": "#000000"}}}%%
graph TD
    A["from praisonai import DB"] --> B["DB(database_url=...)"]
    B --> C{Backend Auto-Detect}
    C -->|"sqlite:// or .db"| D[SQLite]
    C -->|"postgresql://"| E[PostgreSQL]
    C -->|"mysql://"| F[MySQL]
    C -->|"state_url=redis://"| G[Redis + SQLite]
    C -->|"mongodb://"| H[MongoDB + SQLite]
    C -->|ClickHouse client| I[ClickHouse + SQLite]
    C -->|DefaultSessionStore| J[JSON Files]

    D --> K[ManagedAgent]
    E --> K
    F --> K
    G --> K
    H --> K
    I --> K
    J --> K
    K --> L["Agent(backend=managed)"]
    L --> M["agent.run('prompt')"]
    M --> N[Auto-Persist Messages + State]

    classDef hook fill:#189AB4,color:#fff
    classDef agent fill:#8B0000,color:#fff
    classDef decision fill:#444,color:#fff

    class A,B hook
    class K,L,M agent
    class C decision
    class D,E,F,G,H,I,J hook

Simplified API (DB Alias)

PraisonAIDB now has a short alias DB importable directly from praisonai. No more nested imports.

# Before (still works)
from praisonai.db import PraisonAIDB
db = PraisonAIDB(database_url="postgresql://localhost/mydb")

# After (recommended)
from praisonai import DB
db = DB(database_url="postgresql://localhost/mydb")

Architecture: 5-Phase Session Lifecycle

%%{init: {"theme": "base", "themeVariables": {"background": "transparent", "lineColor": "#000000"}}}%%
sequenceDiagram
    participant User
    participant Agent
    participant ManagedAgent
    participant DB

    rect rgb(139, 0, 0, 0.1)
    Note over User,DB: Phase 1 — Create and Teach
    User->>Agent: agent.run("Remember: dog=Biscuit")
    Agent->>ManagedAgent: execute prompt
    ManagedAgent->>DB: persist messages + state
    DB-->>ManagedAgent: saved
    ManagedAgent-->>Agent: response
    Agent-->>User: "Got it! Biscuit the golden retriever."
    end

    rect rgb(68, 68, 68, 0.1)
    Note over User,DB: Phase 2 — Verify DB
    User->>DB: SELECT COUNT(*) FROM messages
    DB-->>User: rows confirmed
    end

    rect rgb(139, 0, 0, 0.1)
    Note over User,DB: Phase 3 — Destroy Instance
    User->>ManagedAgent: save_ids()
    ManagedAgent-->>User: {session_id, agent_id}
    User->>Agent: del agent, managed, db
    end

    rect rgb(24, 154, 180, 0.1)
    Note over User,DB: Phase 4 — Resume (no config needed)
    User->>DB: DB(database_url=same_path)
    User->>ManagedAgent: resume_session(session_id)
    ManagedAgent->>DB: restore config + history
    DB-->>ManagedAgent: full state restored
    User->>Agent: agent.run("What is my dog's name?")
    Agent-->>User: "Biscuit, golden retriever!"
    end

    rect rgb(0, 100, 0, 0.1)
    Note over User,DB: Phase 5 — Validate
    User->>User: assert "biscuit" in result
    end

Supported Backends

1. SQLite (Zero Dependencies)

from praisonai import ManagedAgent, LocalManagedConfig, DB
from praisonaiagents import Agent

# Create with SQLite persistence
db = DB(database_url="/tmp/managed_sqlite.db")
managed = ManagedAgent(
    provider="local", db=db,
    config=LocalManagedConfig(
        model="gpt-4o-mini", name="SQLite Memory Agent",
        system="Remember all facts the user tells you.",
    ),
)
agent = Agent(name="User", backend=managed)
agent.run("Remember: My dog is Biscuit, a golden retriever.")

# Save and destroy
saved_ids = managed.save_ids()
del agent, managed, db

# Resume — no config needed
db2 = DB(database_url="/tmp/managed_sqlite.db")
managed2 = ManagedAgent(provider="local", db=db2)
managed2.resume_session(saved_ids["session_id"])
agent2 = Agent(name="User", backend=managed2)
result = agent2.run("What is my dog's name?")
# Agent remembers: Biscuit, golden retriever

2. PostgreSQL

from praisonai import ManagedAgent, LocalManagedConfig, DB
from praisonaiagents import Agent

PG_URL = "postgresql://postgres:postgres@localhost:5432/postgres"
db = DB(database_url=PG_URL)
managed = ManagedAgent(
    provider="local", db=db,
    config=LocalManagedConfig(
        model="gpt-4o-mini", name="PG Memory Agent",
        system="Remember all facts the user tells you.",
    ),
)
agent = Agent(name="User", backend=managed)
agent.run("Remember: My favourite language is Rust, 3 years experience.")

# Resume later
saved_ids = managed.save_ids()
del agent, managed, db

db2 = DB(database_url=PG_URL)
managed2 = ManagedAgent(provider="local", db=db2)
managed2.resume_session(saved_ids["session_id"])
agent2 = Agent(name="User", backend=managed2)
result = agent2.run("What is my favourite language?")

3. MySQL

from praisonai import ManagedAgent, LocalManagedConfig, DB
from praisonaiagents import Agent

MYSQL_URL = "mysql://root:password@localhost:3307/praisonai"
db = DB(database_url=MYSQL_URL)
managed = ManagedAgent(
    provider="local", db=db,
    config=LocalManagedConfig(
        model="gpt-4o-mini", name="MySQL Memory Agent",
        system="Remember all facts the user tells you.",
    ),
)
agent = Agent(name="User", backend=managed)
agent.run("Remember: I live in Tokyo and love ramen from Ichiran.")

# Resume
saved_ids = managed.save_ids()
del agent, managed, db

db2 = DB(database_url=MYSQL_URL)
managed2 = ManagedAgent(provider="local", db=db2)
managed2.resume_session(saved_ids["session_id"])
agent2 = Agent(name="User", backend=managed2)
result = agent2.run("Where do I live?")

4. Redis (State Store + SQLite Conversations)

from praisonai import ManagedAgent, LocalManagedConfig, DB
from praisonaiagents import Agent

REDIS_URL = "redis://:myredissecret@localhost:6379/0"
db = DB(database_url="/tmp/redis_conv.db", state_url=REDIS_URL)
managed = ManagedAgent(
    provider="local", db=db,
    config=LocalManagedConfig(
        model="gpt-4o-mini", name="Redis Memory Agent",
        system="Remember all facts the user tells you.",
    ),
)
agent = Agent(name="User", backend=managed)
agent.run("Remember: My cat Luna is a Siamese who loves tuna treats.")

# Resume
saved_ids = managed.save_ids()
del agent, managed, db

db2 = DB(database_url="/tmp/redis_conv.db", state_url=REDIS_URL)
managed2 = ManagedAgent(provider="local", db=db2)
managed2.resume_session(saved_ids["session_id"])
agent2 = Agent(name="User", backend=managed2)
result = agent2.run("What is my cat's name?")

5. MongoDB (State Verification + SQLite Conversations)

from praisonai import ManagedAgent, LocalManagedConfig, DB
from praisonaiagents import Agent
import pymongo

MONGO_URL = "mongodb://localhost:27017"
db = DB(database_url="/tmp/mongo_conv.db")
managed = ManagedAgent(
    provider="local", db=db,
    config=LocalManagedConfig(
        model="gpt-4o-mini", name="MongoDB Memory Agent",
        system="Remember all facts the user tells you.",
    ),
)
agent = Agent(name="User", backend=managed)
agent.run("Remember: My favourite book is Dune by Frank Herbert, 1965.")

# Verify in MongoDB
client = pymongo.MongoClient(MONGO_URL)
# ... verify state documents

# Resume
saved_ids = managed.save_ids()
del agent, managed, db

db2 = DB(database_url="/tmp/mongo_conv.db")
managed2 = ManagedAgent(provider="local", db=db2)
managed2.resume_session(saved_ids["session_id"])
agent2 = Agent(name="User", backend=managed2)
result = agent2.run("What is my favourite book?")

6. ClickHouse (Analytics + SQLite Conversations)

from praisonai import ManagedAgent, LocalManagedConfig, DB
from praisonaiagents import Agent
import clickhouse_connect

db = DB(database_url="/tmp/ch_conv.db")
managed = ManagedAgent(
    provider="local", db=db,
    config=LocalManagedConfig(
        model="gpt-4o-mini", name="ClickHouse Memory Agent",
        system="Remember all facts the user tells you.",
    ),
)
agent = Agent(name="User", backend=managed)
agent.run("Remember: My favourite movie is Interstellar by Nolan, 2014.")

# Log analytics to ClickHouse
client = clickhouse_connect.get_client(host="localhost", port=8123)
client.insert("managed_agent_sessions", [[
    managed.session_id, managed.agent_id,
    managed.total_input_tokens, managed.total_output_tokens,
    "{}", # state_json
]])

# Resume
saved_ids = managed.save_ids()
del agent, managed, db

db2 = DB(database_url="/tmp/ch_conv.db")
managed2 = ManagedAgent(provider="local", db=db2)
managed2.resume_session(saved_ids["session_id"])
agent2 = Agent(name="User", backend=managed2)
result = agent2.run("What is my favourite movie?")

7. JSON Files (Zero Dependencies)

from praisonai import ManagedAgent, LocalManagedConfig
from praisonaiagents import Agent
from praisonaiagents.session.store import DefaultSessionStore

store = DefaultSessionStore(session_dir="/tmp/json_sessions")
managed = ManagedAgent(
    provider="local", session_store=store,
    config=LocalManagedConfig(
        model="gpt-4o-mini", name="JSON Memory Agent",
        system="Remember all facts the user tells you.",
    ),
)
agent = Agent(name="User", backend=managed)
agent.run("Remember: I'm learning piano, Chopin is my favourite composer.")

# Resume
saved_ids = managed.save_ids()
del agent, managed, store

store2 = DefaultSessionStore(session_dir="/tmp/json_sessions")
managed2 = ManagedAgent(provider="local", session_store=store2)
managed2.resume_session(saved_ids["session_id"])
agent2 = Agent(name="User", backend=managed2)
result = agent2.run("What instrument am I learning?")

CLI Parity

All persistence features are also available via CLI with the --db flag:

# Single prompt with SQLite persistence
praisonai managed run --db /tmp/data.db "Say hello"

# PostgreSQL persistence
praisonai managed run --db postgresql://localhost/mydb "Remember my name is Alice"

# Multi-turn conversation with persistence
praisonai managed multi --db /tmp/data.db

Import Paths (All Supported)

Component Architecture

%%{init: {"theme": "base", "themeVariables": {"background": "transparent", "lineColor": "#000000"}}}%%
graph LR
    subgraph "Python API"
        A["from praisonai import DB"]
        B["DB(database_url=...)"]
        C["ManagedAgent(db=db)"]
        D["Agent(backend=managed)"]
    end

    subgraph "CLI"
        E["praisonai managed run --db URL"]
        F["praisonai managed multi --db URL"]
    end

    subgraph "Storage Layer"
        G[(SQLite)]
        H[(PostgreSQL)]
        I[(MySQL)]
        J[(Redis)]
        K[(MongoDB)]
        L[(ClickHouse)]
        M[JSON Files]
    end

    A --> B
    B --> C
    C --> D
    E --> C
    F --> C
    C --> G
    C --> H
    C --> I
    C --> J
    C --> K
    C --> L
    C --> M

    classDef hook fill:#189AB4,color:#fff
    classDef agent fill:#8B0000,color:#fff
    classDef storage fill:#444,color:#fff

    class A,B hook
    class C,D,E,F agent
    class G,H,I,J,K,L,M storage

All examples located in examples/managed-agents/persistence/. GitHub Issue #73

PraisonAI supports every major LLM provider through a unified Agent interface. Switch providers by changing a single model parameter — no code changes needed. Supports OpenAI, Anthropic Claude, Google Gemini, Ollama, DeepSeek, Groq, Mistral, Together AI, Cohere, and 100+ models via LiteLLM.

Architecture

%%{init: {"theme": "base", "themeVariables": {"background": "transparent", "lineColor": "#000000"}}}%%
graph TD
    A[Agent] --> B[LLM Router]
    B --> C[OpenAI]
    B --> D[Anthropic]
    B --> E[Gemini]
    B --> F[Ollama]
    B --> G[DeepSeek]
    B --> H[Groq]
    B --> I[Mistral]
    B --> J[Together AI]
    B --> K[Cohere]
    B --> L[100+ via LiteLLM]

    classDef agent fill:#8B0000,color:#fff
    classDef router fill:#189AB4,color:#fff
    classDef provider fill:#444,color:#fff

    class A agent
    class B router
    class C,D,E,F,G,H,I,J,K,L provider

Supported Providers

1. OpenAI

Default provider. Works out of the box with OPENAI_API_KEY.

from praisonaiagents import Agent

# GPT-4o (default)
agent = Agent(
    name="OpenAI Agent",
    instructions="You are a helpful assistant.",
    model="gpt-4o"
)
result = agent.start("Explain quantum computing in simple terms")
print(result)

# GPT-4o Mini (faster, cheaper)
agent_mini = Agent(
    name="Fast Agent",
    model="gpt-4o-mini",
    instructions="Be concise."
)
result = agent_mini.start("What is Python?")

# O1 reasoning model
agent_o1 = Agent(
    name="Reasoning Agent",
    model="o1-mini",
    instructions="Think step by step."
)
result = agent_o1.start("Solve: If 3x + 7 = 22, what is x?")

2. Anthropic Claude

Supports Claude 3.5 Sonnet, Claude 3 Opus, and Claude 3 Haiku.

from praisonaiagents import Agent

# Claude 3.5 Sonnet
agent = Agent(
    name="Claude Agent",
    model="claude-3-5-sonnet-20241022",
    instructions="You are a coding expert."
)
result = agent.start("Write a Python quicksort implementation")

# Claude 3 Opus (most capable)
agent_opus = Agent(
    name="Opus Agent",
    model="claude-3-opus-20240229",
    instructions="You are an expert analyst."
)

# Claude 3 Haiku (fastest)
agent_haiku = Agent(
    name="Haiku Agent",
    model="claude-3-haiku-20240307",
    instructions="Be brief and precise."
)

3. Google Gemini

Supports Gemini 2.0 Flash, 1.5 Pro, and 1.5 Flash with massive context windows up to 2M tokens.

from praisonaiagents import Agent

# Gemini 2.0 Flash
agent = Agent(
    name="Gemini Agent",
    model="gemini/gemini-2.0-flash",
    instructions="You are a research assistant."
)
result = agent.start("Summarize the latest AI trends")

# Gemini 1.5 Pro (2M context window)
agent_pro = Agent(
    name="Long Context Agent",
    model="gemini/gemini-1.5-pro",
    instructions="You can process very long documents."
)

# Gemini 1.5 Flash (fast, 1M context)
agent_flash = Agent(
    name="Flash Agent",
    model="gemini/gemini-1.5-flash",
    instructions="Fast and efficient."
)

4. Ollama (Local Models)

Run models locally with zero API costs. Requires Ollama installed on your machine.

from praisonaiagents import Agent

# Llama 3 via Ollama
agent = Agent(
    name="Local Agent",
    model="ollama/llama3",
    instructions="You are a helpful local assistant."
)
result = agent.start("What is machine learning?")

# Mistral via Ollama
agent_mistral = Agent(
    name="Mistral Local",
    model="ollama/mistral",
    instructions="Be helpful and concise."
)

# Custom Ollama endpoint
agent_custom = Agent(
    name="Custom Ollama",
    model="ollama/llama3",
    llm_config={"base_url": "http://192.168.1.100:11434"}
)

5. DeepSeek

Cost-effective reasoning model with 128K context window.

from praisonaiagents import Agent

agent = Agent(
    name="DeepSeek Agent",
    model="deepseek/deepseek-chat",
    instructions="You are an expert coder."
)
result = agent.start("Implement a binary search tree in Python")

6. Groq

Ultra-fast inference with LPU hardware acceleration.

from praisonaiagents import Agent

# Llama 3 70B on Groq
agent = Agent(
    name="Groq Agent",
    model="groq/llama3-70b-8192",
    instructions="You are a fast assistant."
)
result = agent.start("Explain REST APIs")

# Mixtral on Groq
agent_mixtral = Agent(
    name="Mixtral Agent",
    model="groq/mixtral-8x7b-32768",
    instructions="Be thorough."
)

7. Mistral AI

European AI provider with strong multilingual support.

from praisonaiagents import Agent

agent = Agent(
    name="Mistral Agent",
    model="mistral/mistral-large-latest",
    instructions="You are a multilingual assistant."
)
result = agent.start("Explain transformers in AI")

# Mistral Small (faster)
agent_small = Agent(
    name="Mistral Small",
    model="mistral/mistral-small-latest",
    instructions="Quick and helpful."
)

8. Together AI

Access to open-source models with serverless inference.

from praisonaiagents import Agent

agent = Agent(
    name="Together Agent",
    model="together_ai/meta-llama/Llama-3-70b-chat-hf",
    instructions="You are a research assistant."
)
result = agent.start("Compare supervised and unsupervised learning")

9. Cohere

Enterprise-grade models optimized for RAG and search.

from praisonaiagents import Agent

agent = Agent(
    name="Cohere Agent",
    model="cohere/command-r-plus",
    instructions="You are an enterprise assistant."
)
result = agent.start("What are the best practices for RAG?")

10. xAI Grok

from praisonaiagents import Agent

agent = Agent(
    name="Grok Agent",
    model="xai/grok-beta",
    instructions="You are witty and helpful."
)
result = agent.start("What is the meaning of life?")

Multi-Provider Workflow

Use different providers for different agents in the same workflow.

from praisonaiagents import Agent, AgentTeam

# Research agent with large context
researcher = Agent(
    name="Researcher",
    model="gemini/gemini-1.5-pro",
    instructions="Research topics thoroughly."
)

# Writer with creative capability
writer = Agent(
    name="Writer",
    model="claude-3-5-sonnet-20241022",
    instructions="Write engaging content based on research."
)

# Fast reviewer
reviewer = Agent(
    name="Reviewer",
    model="groq/llama3-70b-8192",
    instructions="Review and provide feedback quickly."
)

team = AgentTeam(agents=[researcher, writer, reviewer])
result = team.start("Write a blog post about quantum computing")

Agent with Tools

Tools work with all providers.

from praisonaiagents import Agent

def get_weather(city: str) -> str:
    """Get weather for a city."""
    return f"Weather in {city}: Sunny, 72F"

def search_web(query: str) -> str:
    """Search the web."""
    return f"Results for: {query} - Found 10 relevant articles"

# Works with any provider
agent = Agent(
    name="Tool Agent",
    model="gpt-4o",  # or any other provider
    instructions="Use tools to help answer questions.",
    tools=[get_weather, search_web]
)
result = agent.start("What is the weather in Tokyo?")

Environment Setup

Set API keys as environment variables or in a .env file.

# .env file
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GEMINI_API_KEY=AI...
DEEPSEEK_API_KEY=sk-...
GROQ_API_KEY=gsk_...
MISTRAL_API_KEY=...
TOGETHERAI_API_KEY=...
COHERE_API_KEY=...
XAI_API_KEY=xai-...

Model Comparison

Installation

# Core (OpenAI only)
pip install praisonaiagents

# With LiteLLM for all providers
pip install praisonaiagents[llm]

# For Ollama (local)
# Install Ollama from https://ollama.ai
ollama pull llama3

PraisonAI supports persistent agent state across 7 database backends — from zero-dependency SQLite to production-scale PostgreSQL, Redis, MongoDB, MySQL, ClickHouse, and JSON file storage. Every backend supports full CRUD, session resume, and direct DB verification.

Architecture

%%{init: {"theme": "base", "themeVariables": {"background": "transparent", "lineColor": "#000000"}}}%%
graph TD
    A[Agent] --> B{Store Type}
    B --> C[ConversationStore]
    B --> D[StateStore]
    B --> E[DefaultSessionStore]
    C --> F[SQLite]
    C --> G[PostgreSQL]
    C --> H[MySQL]
    D --> I[Redis]
    D --> J[MongoDB]
    E --> K[JSON File]
    A --> L[ClickHouse]

    classDef agent fill:#8B0000,color:#fff
    classDef store fill:#189AB4,color:#fff
    classDef db fill:#444,color:#fff

    class A agent
    class B,C,D,E store
    class F,G,H,I,J,K,L db

Supported Backends

1. SQLite ConversationStore

Zero external dependencies — uses Python built-in sqlite3. Perfect for local development.

from praisonai.persistence.conversation.sqlite import SQLiteConversationStore
from praisonai.persistence.conversation.base import ConversationSession, ConversationMessage

store = SQLiteConversationStore(path="praisonai.db")

# Create session
session = ConversationSession(
    session_id="demo-001",
    agent_id="my_agent",
    metadata={"agent_version": 3},
)
store.create_session(session)

# Add messages
msg = ConversationMessage(session_id="demo-001", role="user", content="Hello!")
store.add_message("demo-001", msg)

# Retrieve
stored = store.get_messages("demo-001")
print(f"Messages: {len(stored)}")

# Session resume after restart
store.close()
store2 = SQLiteConversationStore(path="praisonai.db")
resumed = store2.get_session("demo-001")
print(f"Resumed: {resumed.agent_id}")  # my_agent

2. PostgreSQL ConversationStore

Production-grade persistence with ACID compliance. Requires psycopg2-binary.

from praisonai.persistence.conversation.postgres import PostgresConversationStore
from praisonai.persistence.conversation.base import ConversationSession, ConversationMessage

store = PostgresConversationStore(
    url="postgresql://postgres:postgres@localhost:5432/postgres",
    table_prefix="myapp_",
)

session = ConversationSession(
    session_id="pg-demo",
    agent_id="pg_agent",
    metadata={"model": "gpt-4o"},
)
store.create_session(session)
store.add_message("pg-demo", ConversationMessage(
    session_id="pg-demo", role="user", content="What is PostgreSQL?"
))

# Direct SQL verification
import psycopg2
conn = psycopg2.connect("postgresql://postgres:postgres@localhost:5432/postgres")
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM myapp_messages WHERE session_id = %s", ("pg-demo",))
print(f"Messages in DB: {cur.fetchone()[0]}")

# Resume
store.close()
store2 = PostgresConversationStore(url="postgresql://postgres:postgres@localhost:5432/postgres", table_prefix="myapp_")
msgs = store2.get_messages("pg-demo")
print(f"Resumed messages: {len(msgs)}")

3. MySQL ConversationStore

Popular for web applications. Requires mysql-connector-python.

from praisonai.persistence.conversation.mysql import MySQLConversationStore
from praisonai.persistence.conversation.base import ConversationSession, ConversationMessage

store = MySQLConversationStore(
    host="localhost", port=3306, user="root",
    password="secret", database="praisonai", table_prefix="myapp_",
)

session = ConversationSession(session_id="mysql-demo", agent_id="mysql_agent")
store.create_session(session)
store.add_message("mysql-demo", ConversationMessage(
    session_id="mysql-demo", role="user", content="Hello MySQL!"
))

# Direct verification
import mysql.connector
conn = mysql.connector.connect(host="localhost", port=3306, user="root", password="secret", database="praisonai")
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM myapp_messages WHERE session_id = %s", ("mysql-demo",))
print(f"Messages: {cur.fetchone()[0]}")

4. Redis StateStore

Fast key-value store for agent metadata and state. Supports JSON, hash operations, and TTL.

from praisonai.persistence.state.redis import RedisStateStore

store = RedisStateStore(host="localhost", port=6379, password="secret", prefix="myapp:")

# Key-value
store.set("agent_id", "redis_agent_001")
print(store.get("agent_id"))  # redis_agent_001

# JSON state (full agent metadata)
state = {
    "agent_id": "redis_agent",
    "total_input_tokens": 1500,
    "total_output_tokens": 600,
    "compute_instance_id": "docker_001",
    "session_history": [{"id": "s1", "status": "completed"}],
}
store.set_json("managed_state", state)

# Resume after restart
store.close()
store2 = RedisStateStore(host="localhost", port=6379, password="secret", prefix="myapp:")
recovered = store2.get_json("managed_state")
print(f"Tokens: in={recovered['total_input_tokens']}")  # 1500

# Hash operations
store2.hset("agent_meta", "version", "5")
print(store2.hgetall("agent_meta"))

5. MongoDB StateStore

Document-oriented state store with flexible schema. Requires pymongo.

from praisonai.persistence.state.mongodb import MongoDBStateStore

store = MongoDBStateStore(
    url="mongodb://localhost:27017",
    database="praisonai",
    collection="agent_state",
)

store.set("agent_id", "mongo_agent_001")
store.set("managed_state", {
    "agent_version": 4,
    "total_input_tokens": 800,
    "compute_instance_id": "e2b_001",
})

# Direct MongoDB verification
import pymongo
client = pymongo.MongoClient("mongodb://localhost:27017")
doc = client["praisonai"]["agent_state"].find_one({"_id": "managed_state"})
print(f"Agent version: {doc['value']['agent_version']}")  # 4

# Resume
store.close()
store2 = MongoDBStateStore(url="mongodb://localhost:27017", database="praisonai", collection="agent_state")
recovered = store2.get("managed_state")
print(f"Recovered: {recovered['compute_instance_id']}")  # e2b_001

6. ClickHouse

Column-oriented database for analytics and aggregations. Uses clickhouse-connect.

import json
import clickhouse_connect

client = clickhouse_connect.get_client(host="localhost", port=8123, username="default", password="")

# Create table
client.command("""
    CREATE TABLE IF NOT EXISTS agent_sessions (
        session_id String,
        agent_id String,
        total_input_tokens UInt64,
        model String
    ) ENGINE = MergeTree() ORDER BY session_id
""")

# Insert
client.insert("agent_sessions",
    [["s1", "agent_001", 500, "gpt-4o"], ["s2", "agent_002", 1000, "gpt-4o-mini"]],
    column_names=["session_id", "agent_id", "total_input_tokens", "model"],
)

# Aggregate
result = client.query("SELECT agent_id, SUM(total_input_tokens) FROM agent_sessions GROUP BY agent_id")
for row in result.result_rows:
    print(f"  {row[0]}: {row[1]} tokens")

7. JSON File DefaultSessionStore

Simplest option — zero external dependencies. Stores sessions as JSON files on disk.

from praisonaiagents.session.store import DefaultSessionStore

store = DefaultSessionStore(session_dir="./sessions")

# Add messages
store.add_message("demo", "user", "What is ML?")
store.add_message("demo", "assistant", "ML is a subset of AI.")

# Store metadata
session = store.get_session("demo")
session.metadata = {
    "agent_id": "json_agent",
    "total_input_tokens": 450,
    "compute_instance_id": "local_001",
}
store._save_session(session)

# Resume from disk
store2 = DefaultSessionStore(session_dir="./sessions")
history = store2.get_chat_history("demo")
print(f"Recovered {len(history)} messages")
meta = store2.get_session("demo").metadata
print(f"Agent: {meta['agent_id']}, Tokens: {meta['total_input_tokens']}")

Installation

When to Use Which Backend

A single script that demonstrates the full PraisonAI Managed Agent lifecycle — automatic resource management, streaming, tool selection, custom tool callbacks, web search, package installation, interrupts, and usage tracking. All in half the code of the raw Claude API.

Architecture

%%{init: {"theme": "base", "themeVariables": {"background": "transparent", "lineColor": "#000000"}}}%%
graph TD
    A[Agent + ManagedAgent] --> B[Auto Setup]
    B --> C[agent start]
    C --> D[Stream / Result]

    subgraph Auto Setup
        E[Create Agent] --> F[Create Environment]
        F --> G[Create Session]
    end

    B --> E

    classDef agent fill:#8B0000,color:#fff
    classDef tool fill:#189AB4,color:#fff

    class A,D agent
    class B,C,E,F,G tool

How to Run

Follow these steps to run the full demo:

Step 1: Install PraisonAI

pip install praisonai

Step 2: Set your API key

export ANTHROPIC_API_KEY=your-api-key

Step 3: Save the code below as app.py

Step 4: Run it

python app.py

Full Code

import json
from praisonai import Agent, ManagedAgent, ManagedConfig

# 1. Create an agent

managed = ManagedAgent()
agent = Agent(name="teacher", backend=managed)
result = agent.start("Say hello briefly", stream=True)

print(f"[1] Agent created: {managed.agent_id} (v{managed.agent_version})")

# 2. Update the agent

managed.update_agent(
    name="Teaching Agent v2",
    system="You are a senior Python developer. Write clean, production-quality code.",
)

print(f"[2] Agent updated: Teaching Agent v2 (v{managed.agent_version})")

# 3-4. Environment + Session are created automatically (already done in step 1)

print(f"[3] Environment created: {managed.environment_id}")
print(f"[4] Session created: {managed.session_id}")

# 5. Stream a response

print("\n[5] Streaming response...")

result = agent.start("Write a Python script that prints 'Hello from Managed Agents!' and run it", stream=True)

# 6. Multi-turn conversation (same session remembers context)

print("\n[6] Multi-turn: sending follow-up...")

result = agent.start("Now modify that script to accept a name argument and greet that person", stream=True)

# 7. Track usage

info = managed.retrieve_session()
print("\n[7] Usage report:")
if info.get("usage"):
    print(f"    Input tokens:  {info['usage']['input_tokens']}")
    print(f"    Output tokens: {info['usage']['output_tokens']}")
else:
    print(f"    Input tokens:  {managed.total_input_tokens}")
    print(f"    Output tokens: {managed.total_output_tokens}")

# 8. List sessions

sessions = managed.list_sessions()
print(f"\n[8] Total sessions: {len(sessions)}")
for s in sessions[:3]:
    print(f"    {s['id']} | {s['status']} | {s['title']}")

# 9. Selective tools (only bash + read + write)

bash_managed = ManagedAgent(
    config=ManagedConfig(
        name="Bash Only Agent",
        model="claude-haiku-4-5",
        system="You can only use bash, read, and write tools.",
        tools=[
            {
                "type": "agent_toolset_20260401",
                "default_config": {"enabled": False},
                "configs": [
                    {"name": "bash", "enabled": True},
                    {"name": "read", "enabled": True},
                    {"name": "write", "enabled": True},
                ],
            },
        ],
    ),
)

bash_agent = Agent(name="bash-only", backend=bash_managed)

print("\n[9] Bash-only agent streaming...")
result = bash_agent.start("Show the current date and Python version using bash", stream=True)

# 10. Disable specific tools (web disabled, everything else on)

no_web_managed = ManagedAgent(
    config=ManagedConfig(
        name="No Web Agent",
        model="claude-haiku-4-5",
        system="You are a coding assistant. You cannot access the web.",
        tools=[
            {
                "type": "agent_toolset_20260401",
                "configs": [
                    {"name": "web_fetch", "enabled": False},
                    {"name": "web_search", "enabled": False},
                ],
            },
        ],
    ),
)

no_web_agent = Agent(name="no-web", backend=no_web_managed)

print("\n[10] No-web agent streaming...")
result = no_web_agent.start("Write a Python one-liner that calculates 2**100 and print the result", stream=True)

# 11. Custom tools (you define the tool, PraisonAI calls your callback)


def handle_weather(tool_name, tool_input):
    print(f"\n  [Custom tool: {tool_name} | Input: {json.dumps(tool_input)}]")
    return "Tokyo: 22°C, sunny, humidity 55%"


custom_managed = ManagedAgent(
    config=ManagedConfig(
        name="Weather Agent",
        model="claude-haiku-4-5",
        system="You are a weather assistant. Use the get_weather tool to check weather.",
        tools=[
            {"type": "agent_toolset_20260401"},
            {
                "type": "custom",
                "name": "get_weather",
                "description": "Get current weather for a location",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "location": {"type": "string", "description": "City name"},
                    },
                    "required": ["location"],
                },
            },
        ],
    ),
    on_custom_tool=handle_weather,
)

custom_agent = Agent(name="weather", backend=custom_managed)

print("\n[11] Custom tool agent streaming...")
result = custom_agent.start("What is the weather in Tokyo?", stream=True)

# 12. Web search agent

search_managed = ManagedAgent(
    config=ManagedConfig(
        name="Search Agent",
        model="claude-haiku-4-5",
        system="You are a research assistant. Search the web and summarize.",
    ),
)

search_agent = Agent(name="searcher", backend=search_managed)

print("\n[12] Web search agent streaming...")
result = search_agent.start("Search the web for Python 3.13 new features and give me 3 bullet points", stream=True)

# 13. Environment with pre-installed packages

data_managed = ManagedAgent(
    config=ManagedConfig(
        name="Data Science Agent",
        model="claude-haiku-4-5",
        system="You are a data science assistant.",
        packages={"pip": ["pandas", "numpy"]},
    ),
)

data_agent = Agent(name="data-scientist", backend=data_managed)

print("\n[13] Data science environment streaming...")
result = data_agent.start("Use pandas to create a small DataFrame with 3 rows of sample data and print it", stream=True)

# 14. Interrupt a session

interrupt_managed = ManagedAgent(
    config=ManagedConfig(
        name="Interruptable Agent",
        model="claude-haiku-4-5",
        system="You are a helpful coding assistant.",
    ),
)

interrupt_agent = Agent(name="interruptable", backend=interrupt_managed)

print("\n[14] Interrupt demo...")
result = interrupt_agent.start("Write a Python script that prints numbers 1 to 10", stream=True)
interrupt_managed.interrupt()
print("  [Interrupt sent]")

# Final usage summary

print("\n" + "=" * 60)
print("FINAL USAGE SUMMARY")
print("=" * 60)

all_backends = [
    ("Teaching Agent v2", managed),
    ("Bash Only Agent", bash_managed),
    ("No Web Agent", no_web_managed),
    ("Weather Agent", custom_managed),
    ("Search Agent", search_managed),
    ("Data Science Agent", data_managed),
    ("Interruptable Agent", interrupt_managed),
]

total_input = 0
total_output = 0

for name, backend in all_backends:
    info = backend.retrieve_session()
    usage = info.get("usage", {})
    inp = usage.get("input_tokens", backend.total_input_tokens)
    out = usage.get("output_tokens", backend.total_output_tokens)
    total_input += inp
    total_output += out
    print(f"  {name:30s} | in: {inp:6d} | out: {out:6d}")

print(f"  {'TOTAL':30s} | in: {total_input:6d} | out: {total_output:6d}")
print("=" * 60)

What Each Step Does

Claude API vs PraisonAI

A single script that walks through the entire Claude Managed Agents lifecycle — agent creation, environment setup, session management, streaming, tool selection, custom tools, web search, package installation, interrupts, and usage tracking.

Architecture

%%{init: {"theme": "base", "themeVariables": {"background": "transparent", "lineColor": "#000000"}}}%%
graph TD
    A[Create Agent] --> B[Update Agent]
    B --> C[Create Environment]
    C --> D[Create Session]
    D --> E[Stream Response]
    E --> F[Multi-Turn]
    F --> G[Track Usage]
    G --> H[List Sessions]
    H --> I[Selective Tools]
    I --> J[Disable Tools]
    J --> K[Custom Tools]
    K --> L[Web Search]
    L --> M[Package Install]
    M --> N[Interrupt]
    N --> O[Usage Summary]

    classDef agent fill:#8B0000,color:#fff
    classDef tool fill:#189AB4,color:#fff

    class A,B,C,D,O agent
    class E,F,G,H,I,J,K,L,M,N tool

How to Run

Follow these steps to run the full demo:

Step 1: Install the SDK

pip install anthropic

Step 2: Set your API key

export ANTHROPIC_API_KEY=your-api-key

Step 3: Save the code below as app.py

Step 4: Run it

python app.py

Full Code

from anthropic import Anthropic

client = Anthropic()

# 1. Create an agent

agent = client.beta.agents.create(
    name="Teaching Agent",
    model="claude-haiku-4-5",
    system="You are a helpful coding assistant.",
    tools=[
        {"type": "agent_toolset_20260401"},
    ],
)

print(f"[1] Agent created: {agent.id} (v{agent.version})")

# 2. Update the agent

agent = client.beta.agents.update(
    agent.id,
    version=agent.version,
    name="Teaching Agent v2",
    system="You are a senior Python developer. Write clean, production-quality code.",
)

print(f"[2] Agent updated: {agent.name} (v{agent.version})")

# 3. Create an environment

environment = client.beta.environments.create(
    name="teaching-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
    },
)

print(f"[3] Environment created: {environment.id}")

# 4. Create a session

session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    title="Teaching session",
)

print(f"[4] Session created: {session.id}")

# 5. Stream a response

print(f"\n[5] Streaming response...")

with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "Write a Python script that prints 'Hello from Managed Agents!' and run it"}],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                print(f"\n  [Tool: {event.name}]")
            case "session.status_idle":
                print()
                break

# 6. Multi-turn conversation (same session remembers context)

print(f"\n[6] Multi-turn: sending follow-up...")

with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "Now modify that script to accept a name argument and greet that person"}],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                print(f"\n  [Tool: {event.name}]")
            case "session.status_idle":
                print()
                break

# 7. Track usage

result = client.beta.sessions.retrieve(session.id)
print(f"\n[7] Usage report:")
print(f"    Input tokens:  {result.usage.input_tokens}")
print(f"    Output tokens: {result.usage.output_tokens}")

# 8. List sessions

sessions = client.beta.sessions.list()
print(f"\n[8] Total sessions: {len(sessions.data)}")
for s in sessions.data[:3]:
    print(f"    {s.id} | {s.status} | {s.title}")

# 9. Selective tools (new agent with only bash + read + write)

bash_agent = client.beta.agents.create(
    name="Bash Only Agent",
    model="claude-haiku-4-5",
    system="You can only use bash, read, and write tools.",
    tools=[
        {
            "type": "agent_toolset_20260401",
            "default_config": {"enabled": False},
            "configs": [
                {"name": "bash", "enabled": True},
                {"name": "read", "enabled": True},
                {"name": "write", "enabled": True},
            ],
        },
    ],
)

bash_session = client.beta.sessions.create(
    agent=bash_agent.id,
    environment_id=environment.id,
    title="Bash only session",
)

print(f"\n[9] Bash-only agent streaming...")

with client.beta.sessions.events.stream(bash_session.id) as stream:
    client.beta.sessions.events.send(
        bash_session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "Show the current date and Python version using bash"}],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                print(f"\n  [Tool: {event.name}]")
            case "session.status_idle":
                print()
                break

# 10. Disable specific tools (web disabled, everything else on)

no_web_agent = client.beta.agents.create(
    name="No Web Agent",
    model="claude-haiku-4-5",
    system="You are a coding assistant. You cannot access the web.",
    tools=[
        {
            "type": "agent_toolset_20260401",
            "configs": [
                {"name": "web_fetch", "enabled": False},
                {"name": "web_search", "enabled": False},
            ],
        },
    ],
)

no_web_session = client.beta.sessions.create(
    agent=no_web_agent.id,
    environment_id=environment.id,
    title="No web session",
)

print(f"\n[10] No-web agent streaming...")

with client.beta.sessions.events.stream(no_web_session.id) as stream:
    client.beta.sessions.events.send(
        no_web_session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "Write a Python one-liner that calculates 2**100 and print the result"}],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                print(f"\n  [Tool: {event.name}]")
            case "session.status_idle":
                print()
                break

# 11. Custom tools (you define the tool, you provide the result)

custom_agent = client.beta.agents.create(
    name="Weather Agent",
    model="claude-haiku-4-5",
    system="You are a weather assistant. Use the get_weather tool to check weather.",
    tools=[
        {"type": "agent_toolset_20260401"},
        {
            "type": "custom",
            "name": "get_weather",
            "description": "Get current weather for a location",
            "input_schema": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "City name"},
                },
                "required": ["location"],
            },
        },
    ],
)

custom_session = client.beta.sessions.create(
    agent=custom_agent.id,
    environment_id=environment.id,
    title="Custom tool session",
)

print(f"\n[11] Custom tool agent streaming...")
import json
with client.beta.sessions.events.stream(custom_session.id) as stream:
    client.beta.sessions.events.send(
        custom_session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "What is the weather in Tokyo?"}],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.custom_tool_use":
                print(f"\n  [Custom tool: {event.name} | Input: {json.dumps(event.input)}]")
                client.beta.sessions.events.send(
                    custom_session.id,
                    events=[
                        {
                            "type": "user.custom_tool_result",
                            "custom_tool_use_id": event.id,
                            "content": [
                                {
                                    "type": "text",
                                    "text": "Tokyo: 22°C, sunny, humidity 55%",
                                },
                            ],
                        },
                    ],
                )
            case "session.status_idle":
                if hasattr(event, "stop_reason") and event.stop_reason and event.stop_reason.type == "requires_action":
                    continue
                print()
                break

# 12. Web search agent

search_agent = client.beta.agents.create(
    name="Search Agent",
    model="claude-haiku-4-5",
    system="You are a research assistant. Search the web and summarize.",
    tools=[
        {"type": "agent_toolset_20260401"},
    ],
)

search_session = client.beta.sessions.create(
    agent=search_agent.id,
    environment_id=environment.id,
    title="Search session",
)

print(f"\n[12] Web search agent streaming...")

with client.beta.sessions.events.stream(search_session.id) as stream:
    client.beta.sessions.events.send(
        search_session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "Search the web for Python 3.13 new features and give me 3 bullet points"}],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                print(f"\n  [Tool: {event.name}]")
            case "session.status_idle":
                print()
                break

# 13. Environment with pre-installed packages

data_env = client.beta.environments.create(
    name="data-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
        "packages": {
            "pip": ["pandas", "numpy"],
        },
    },
)

data_session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=data_env.id,
    title="Data science session",
)

print(f"\n[13] Data science environment streaming...")

with client.beta.sessions.events.stream(data_session.id) as stream:
    client.beta.sessions.events.send(
        data_session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "Use pandas to create a small DataFrame with 3 rows of sample data and print it"}],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                print(f"\n  [Tool: {event.name}]")
            case "session.status_idle":
                print()
                break

# 14. Interrupt a session

interrupt_session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    title="Interrupt session",
)

print(f"\n[14] Interrupt demo...")

with client.beta.sessions.events.stream(interrupt_session.id) as stream:
    client.beta.sessions.events.send(
        interrupt_session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "Write a Python script that prints numbers 1 to 10"}],
            },
        ],
    )

    tool_count = 0
    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                tool_count += 1
                print(f"\n  [Tool: {event.name}]")
                if tool_count >= 1:
                    print("  [Sending interrupt!]")
                    client.beta.sessions.events.send(
                        interrupt_session.id,
                        events=[{"type": "user.interrupt"}],
                    )
            case "session.status_idle":
                print()
                break

# Final usage summary

print("\n" + "=" * 60)
print("FINAL USAGE SUMMARY")
print("=" * 60)

all_sessions = [session.id, bash_session.id, no_web_session.id, custom_session.id, search_session.id, data_session.id, interrupt_session.id]

total_input = 0
total_output = 0

for sid in all_sessions:
    s = client.beta.sessions.retrieve(sid)
    if s.usage:
        total_input += s.usage.input_tokens
        total_output += s.usage.output_tokens
        print(f"  {s.title:30s} | in: {s.usage.input_tokens:6d} | out: {s.usage.output_tokens:6d}")

print(f"  {'TOTAL':30s} | in: {total_input:6d} | out: {total_output:6d}")
print("=" * 60)

What Each Step Does

PraisonAI wraps Claude Managed Agents with a simple, Pythonic interface — automatic agent, environment, and session management in just a few lines. Each section is a standalone, runnable script.

Architecture

%%{init: {"theme": "base", "themeVariables": {"background": "transparent", "lineColor": "#000000"}}}%%
graph TD
    A[PraisonAI Agent] --> B[ManagedAgent Backend]
    B --> C[Auto-Create Agent]
    B --> D[Auto-Create Environment]
    B --> E[Auto-Create Session]
    C --> F[Claude API]
    D --> F
    E --> F
    F --> G[Stream Response]
    G --> H[Result]

    classDef agent fill:#8B0000,color:#fff
    classDef tool fill:#189AB4,color:#fff
    classDef decision fill:#444,color:#fff

    class A,H agent
    class B,C,D,E tool
    class F,G decision

Prerequisites

pip install praisonai

Set your API key as an environment variable:

export ANTHROPIC_API_KEY=your-api-key

basic

The simplest possible managed agent — 3 lines of code.

from praisonai import Agent, ManagedAgent

agent = Agent(name="teacher", backend=ManagedAgent())
result = agent.start("Write a Python script that prints 'Hello from Managed Agents!' and run it")
print(result)

01_create_agent

Zero config — defaults to name=”Agent” and model=”claude-haiku-4-5″. Access agent_id and version after first call.

from praisonai import Agent, ManagedAgent

# Zero config — defaults: name="Agent", model="claude-haiku-4-5"
managed = ManagedAgent()
agent = Agent(name="coder", backend=managed)
result = agent.start("Say hello")

print(f"Agent ID: {managed.agent_id}")
print(f"Version: {managed.agent_version}")

02_create_environment

Environment is created automatically — just specify packages and networking in ManagedConfig.

from praisonai import Agent, ManagedAgent, ManagedConfig

# Environment is created automatically — just specify packages/networking in config
managed = ManagedAgent(
    config=ManagedConfig(
        name="Coding Assistant",
        model="claude-haiku-4-5",
        system="You are a helpful coding assistant. Write clean, well-documented code.",
        networking={"type": "unrestricted"},
    ),
)

agent = Agent(name="coder", backend=managed)
result = agent.start("Say hello")

print(f"Agent ID: {managed.agent_id}")
print(f"Environment ID: {managed.environment_id}")

03_create_session

Agent, environment, and session are all created automatically on first use.

from praisonai import Agent, ManagedAgent, ManagedConfig

# Agent, environment, and session are all created automatically on first use
managed = ManagedAgent(
    config=ManagedConfig(
        name="Coding Assistant",
        model="claude-haiku-4-5",
        system="You are a helpful coding assistant. Write clean, well-documented code.",
        session_title="Quickstart session",
    ),
)

agent = Agent(name="coder", backend=managed)
result = agent.start("Say hello")

print(f"Agent ID: {managed.agent_id}")
print(f"Environment ID: {managed.environment_id}")
print(f"Session ID: {managed.session_id}")

04_stream_response

Stream events in real time by passing stream=True to agent.start().

from praisonai import Agent, ManagedAgent, ManagedConfig

managed = ManagedAgent(
    config=ManagedConfig(
        name="Coding Assistant",
        model="claude-haiku-4-5",
        system="You are a helpful coding assistant. Write clean, well-documented code.",
    ),
)

agent = Agent(name="coder", backend=managed)

result = agent.start(
    "Create a Python script that generates the first 20 Fibonacci numbers and saves them to fibonacci.txt",
    stream=True,
)

print("\nAgent finished.")

05_select_tools

Select specific tools: only bash, read, write — everything else disabled.

from praisonai import Agent, ManagedAgent, ManagedConfig

# Select specific tools: only bash, read, write — everything else disabled
managed = ManagedAgent(
    config=ManagedConfig(
        name="Bash Only Agent",
        model="claude-haiku-4-5",
        system="You are a helpful assistant that can only use bash commands.",
        tools=[
            {
                "type": "agent_toolset_20260401",
                "default_config": {"enabled": False},
                "configs": [
                    {"name": "bash", "enabled": True},
                    {"name": "read", "enabled": True},
                    {"name": "write", "enabled": True},
                ],
            },
        ],
    ),
)

agent = Agent(name="bash-agent", backend=managed)
result = agent.start("List the current directory contents and show system info using uname -a")

print(result)
print("\nAgent finished.")

06_disable_tools

Disable specific tools: web access disabled, everything else stays on.

from praisonai import Agent, ManagedAgent, ManagedConfig

# Disable specific tools: web access disabled, everything else stays on
managed = ManagedAgent(
    config=ManagedConfig(
        name="No Web Agent",
        model="claude-haiku-4-5",
        system="You are a coding assistant. You cannot access the web.",
        tools=[
            {
                "type": "agent_toolset_20260401",
                "configs": [
                    {"name": "web_fetch", "enabled": False},
                    {"name": "web_search", "enabled": False},
                ],
            },
        ],
    ),
)

agent = Agent(name="no-web-agent", backend=managed)
result = agent.start("Write a hello world Python script and run it")

print(result)
print("\nAgent finished.")

07_custom_tools

Define a custom tool and handle its invocation with an on_custom_tool callback.

import json
from praisonai import Agent, ManagedAgent, ManagedConfig


def handle_weather(tool_name, tool_input):
    """Custom tool callback — PraisonAI calls this when the agent uses get_weather."""
    location = tool_input.get("location", "Unknown")
    print(f"\n[Custom tool call: {tool_name}]")
    print(f"[Input: {json.dumps(tool_input)}]")
    return f"Weather in {location}: 15°C, partly cloudy, humidity 72%"


managed = ManagedAgent(
    config=ManagedConfig(
        name="Weather Agent",
        model="claude-haiku-4-5",
        system="You are a weather assistant. Use the get_weather tool to check weather.",
        tools=[
            {"type": "agent_toolset_20260401"},
            {
                "type": "custom",
                "name": "get_weather",
                "description": "Get current weather for a location",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "location": {"type": "string", "description": "City name"},
                    },
                    "required": ["location"],
                },
            },
        ],
    ),
    on_custom_tool=handle_weather,
)

agent = Agent(name="weather-agent", backend=managed)
result = agent.start("What is the weather in London?")

print(result)
print("\nAgent finished.")

08_update_agent

Create an agent, then update its name and system prompt without recreating it.

from praisonai import Agent, ManagedAgent, ManagedConfig

managed = ManagedAgent(
    config=ManagedConfig(
        name="My Agent v1",
        model="claude-haiku-4-5",
        system="You are a helpful assistant.",
    ),
)

agent = Agent(name="updatable", backend=managed)

# First call creates the agent
result = agent.start("Say hello briefly")
print(f"Created: {managed.agent_id}, version: {managed.agent_version}")

# Update the agent's system prompt (no need to recreate)
managed.update_agent(
    name="My Agent v2",
    system="You are a senior Python developer. Write production-quality code.",
)
print(f"Updated: {managed.agent_id}, version: {managed.agent_version}")

09_list_sessions

List all sessions for the current agent.

from praisonai import Agent, ManagedAgent, ManagedConfig

managed = ManagedAgent(
    config=ManagedConfig(
        name="Session Demo Agent",
        model="claude-haiku-4-5",
        system="You are a helpful assistant.",
    ),
)

agent = Agent(name="session-demo", backend=managed)

# Run a task to create agent + session
agent.start("Say hello briefly")

# List sessions for this agent
sessions = managed.list_sessions()
print(f"Total sessions: {len(sessions)}")
for s in sessions[:5]:
    print(f"  ID: {s['id']}")
    print(f"  Status: {s['status']}")
    print(f"  Title: {s['title']}")
    print()

10_web_search

Create a research agent that searches the web — web tools are enabled by default.

from praisonai import Agent, ManagedAgent, ManagedConfig

managed = ManagedAgent(
    config=ManagedConfig(
        name="Research Agent",
        model="claude-haiku-4-5",
        system="You are a research assistant. Search the web for information.",
    ),
)

agent = Agent(name="researcher", backend=managed)
result = agent.start(
    "Search the web for the latest Python 3.13 features and summarize them in 3 bullet points"
)

print(result)
print("\nAgent finished.")

11_multi_turn

Multi-turn: each call reuses the same session — agent remembers context.

from praisonai import Agent, ManagedAgent, ManagedConfig

managed = ManagedAgent(
    config=ManagedConfig(
        name="Multi Turn Agent",
        model="claude-haiku-4-5",
        system="You are a helpful coding assistant.",
    ),
)

agent = Agent(name="multi-turn", backend=managed)
print(f"Session ID: {managed.session_id}")

# Multi-turn: each call reuses the same session — agent remembers context
messages = [
    "Create a file called greeting.py that prints 'Hello World'",
    "Now modify greeting.py to accept a name as a command line argument",
    "Run greeting.py with the argument 'Claude'",
]

for i, message in enumerate(messages):
    print(f"\n--- Turn {i + 1}: {message} ---\n")
    result = agent.start(message)
    print(result)
    print("\nAgent finished turn.")

12_environment_setup

Just add packages to the config — environment is created automatically with pip packages pre-installed.

from praisonai import Agent, ManagedAgent, ManagedConfig

# Just add packages to the config — environment is created automatically
managed = ManagedAgent(
    config=ManagedConfig(
        name="Data Science Agent",
        model="claude-haiku-4-5",
        system="You are a data science assistant.",
        packages={"pip": ["pandas", "numpy"]},
    ),
)

agent = Agent(name="data-scientist", backend=managed)

print(f"Environment ID: {managed.environment_id}")

result = agent.start(
    "Create a Python script that generates random sales data with pandas and saves a summary to sales_summary.txt"
)

print(result)
print("\nAgent finished.")

13_interrupt_session

Interrupt a running agent with a single method call.

from praisonai import Agent, ManagedAgent, ManagedConfig

managed = ManagedAgent(
    config=ManagedConfig(
        name="Interruptable Agent",
        model="claude-haiku-4-5",
        system="You are a helpful coding assistant.",
    ),
)

agent = Agent(name="interruptable", backend=managed)

# Start a task
result = agent.start("Write a Python script that prints numbers 1 to 5")
print(result)

# Interrupt the agent mid-work
print("\n[Sending interrupt...]")
managed.interrupt()
print("Agent stopped.")

14_track_usage

Token usage is tracked automatically — access via properties or the API.

from praisonai import Agent, ManagedAgent, ManagedConfig

managed = ManagedAgent(
    config=ManagedConfig(
        name="Usage Tracker Agent",
        model="claude-haiku-4-5",
        system="You are a helpful assistant.",
    ),
)

agent = Agent(name="tracker", backend=managed)
result = agent.start("Write a one-line Python script that prints the current date and run it")

print(result)
print("\nAgent finished.")

# Usage tracked automatically
print("\n--- Usage Report ---")
print(f"Input tokens:  {managed.total_input_tokens}")
print(f"Output tokens: {managed.total_output_tokens}")

# Or retrieve detailed session info from the API
info = managed.retrieve_session()
print(f"\nSession ID: {info.get('id')}")
print(f"Status: {info.get('status')}")
if "usage" in info:
    print(f"API Input tokens:  {info['usage']['input_tokens']}")
    print(f"API Output tokens: {info['usage']['output_tokens']}")

15_resume_session

Save agent/environment/session IDs to disk, then resume the session in a later run. Run the script twice — first run saves, second run resumes.

import json
import pathlib
from praisonai import Agent, ManagedAgent, ManagedConfig

IDS_FILE = pathlib.Path("managed_ids.json")

if IDS_FILE.exists():
    saved = json.loads(IDS_FILE.read_text())
    print(f"Resuming session: {saved['session_id']}\n")

    managed = ManagedAgent()
    managed.resume_session(saved["session_id"])

    agent = Agent(name="coder", backend=managed)
    result = agent.start("What is my favourite number?", stream=True)

else:
    managed = ManagedAgent(
        config=ManagedConfig(
            name="Persistent Coder",
            model="claude-haiku-4-5",
            system="You are a helpful coding assistant.",
        ),
    )
    agent = Agent(name="coder", backend=managed)
    result = agent.start("Remember this: my favourite number is 42.", stream=True)

    ids = managed.save_ids()
    IDS_FILE.write_text(json.dumps(ids, indent=2))

    print(f"\nSaved IDs to {IDS_FILE}")
    print(f"  agent_id      : {managed.agent_id}")
    print(f"  environment_id: {managed.environment_id}")
    print(f"  session_id    : {managed.session_id}")
    print("\nRun this script again to resume the session.")

16_session_ids_explained

Inspect all resource IDs and save them with save_ids() for later reuse.

from praisonai import Agent, ManagedAgent, ManagedConfig

managed = ManagedAgent(
    config=ManagedConfig(
        name="Session Demo",
        model="claude-haiku-4-5",
        system="You are a helpful assistant.",
    ),
)
agent = Agent(name="demo", backend=managed)

agent.start("Say hello and confirm you are ready.", stream=True)

print("\n--- IDs ---")
print(f"agent_id      : {managed.agent_id}")
print(f"environment_id: {managed.environment_id}")
print(f"session_id    : {managed.session_id}")
print(f"chat_history  : {len(agent.chat_history)} messages")

ids = managed.save_ids()
print(f"\nsave_ids(): {ids}")

Summary

Claude vs PraisonAI Comparison

A step-by-step guide to Claude Managed Agents — from creating your first agent to custom tools, multi-turn conversations, and usage tracking. Each section is a standalone, runnable Python script.

Architecture

%%{init: {"theme": "base", "themeVariables": {"background": "transparent", "lineColor": "#000000"}}}%%
graph TD
    A[Create Agent] --> B[Create Environment]
    B --> C[Create Session]
    C --> D[Stream Events]
    D --> E{Event Type}
    E --> F[agent.message]
    E --> G[agent.tool_use]
    E --> H[agent.custom_tool_use]
    E --> I[session.status_idle]
    I --> J[Done]

    classDef agent fill:#8B0000,color:#fff
    classDef tool fill:#189AB4,color:#fff
    classDef decision fill:#444,color:#fff

    class A,B,C,D,J agent
    class F,G,H tool
    class E decision

Prerequisites

pip install anthropic

Set your API key as an environment variable:

export ANTHROPIC_API_KEY=your-api-key

01_create_agent

Create a basic Claude managed agent with the Anthropic SDK.

from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="Coding Assistant",
    model="claude-haiku-4-5",
    system="You are a helpful coding assistant. Write clean, well-documented code.",
    tools=[
        {"type": "agent_toolset_20260401"},
    ],
)

print(f"Agent ID: {agent.id}")
print(f"Version: {agent.version}")

02_create_environment

Create an agent and a cloud environment with unrestricted networking.

from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="Coding Assistant",
    model="claude-haiku-4-5",
    system="You are a helpful coding assistant. Write clean, well-documented code.",
    tools=[
        {"type": "agent_toolset_20260401"},
    ],
)

print(f"Agent ID: {agent.id}")

environment = client.beta.environments.create(
    name="quickstart-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
    },
)

print(f"Environment ID: {environment.id}")

03_create_session

Create an agent, environment, and session — the three resources needed to run tasks.

from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="Coding Assistant",
    model="claude-haiku-4-5",
    system="You are a helpful coding assistant. Write clean, well-documented code.",
    tools=[
        {"type": "agent_toolset_20260401"},
    ],
)

print(f"Agent ID: {agent.id}")

environment = client.beta.environments.create(
    name="quickstart-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
    },
)

print(f"Environment ID: {environment.id}")

session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    title="Quickstart session",
)

print(f"Session ID: {session.id}")

04_stream_response

Stream events from a session in real time — messages, tool use, and idle status.

from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="Coding Assistant",
    model="claude-haiku-4-5",
    system="You are a helpful coding assistant. Write clean, well-documented code.",
    tools=[
        {"type": "agent_toolset_20260401"},
    ],
)

print(f"Agent ID: {agent.id}")

environment = client.beta.environments.create(
    name="quickstart-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
    },
)

print(f"Environment ID: {environment.id}")

session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    title="Quickstart session",
)

print(f"Session ID: {session.id}")

with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [
                    {
                        "type": "text",
                        "text": "Create a Python script that generates the first 20 Fibonacci numbers and saves them to fibonacci.txt",
                    },
                ],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                print(f"\n[Using tool: {event.name}]")
            case "session.status_idle":
                print("\n\nAgent finished.")
                break

05_select_tools

Selectively enable only specific tools (bash, read, write) while disabling everything else.

from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="Bash Only Agent",
    model="claude-haiku-4-5",
    system="You are a helpful assistant that can only use bash commands.",
    tools=[
        {
            "type": "agent_toolset_20260401",
            "default_config": {"enabled": False},
            "configs": [
                {"name": "bash", "enabled": True},
                {"name": "read", "enabled": True},
                {"name": "write", "enabled": True},
            ],
        },
    ],
)

print(f"Agent ID: {agent.id}")

environment = client.beta.environments.create(
    name="bash-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
    },
)

session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    title="Bash only session",
)

with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [
                    {
                        "type": "text",
                        "text": "List the current directory contents and show system info using uname -a",
                    },
                ],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                print(f"\n[Using tool: {event.name}]")
            case "session.status_idle":
                print("\n\nAgent finished.")
                break

06_disable_tools

Disable specific tools (web access) while keeping everything else enabled by default.

from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="No Web Agent",
    model="claude-haiku-4-5",
    system="You are a coding assistant. You cannot access the web.",
    tools=[
        {
            "type": "agent_toolset_20260401",
            "configs": [
                {"name": "web_fetch", "enabled": False},
                {"name": "web_search", "enabled": False},
            ],
        },
    ],
)

print(f"Agent ID: {agent.id}")

environment = client.beta.environments.create(
    name="no-web-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
    },
)

session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    title="No web session",
)

with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [
                    {
                        "type": "text",
                        "text": "Write a hello world Python script and run it",
                    },
                ],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                print(f"\n[Using tool: {event.name}]")
            case "session.status_idle":
                print("\n\nAgent finished.")
                break

07_custom_tools

Define a custom tool (get_weather) and handle its invocation with a callback to return results to the agent.

import json
from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="Weather Agent",
    model="claude-haiku-4-5",
    system="You are a weather assistant. Use the get_weather tool to check weather.",
    tools=[
        {"type": "agent_toolset_20260401"},
        {
            "type": "custom",
            "name": "get_weather",
            "description": "Get current weather for a location",
            "input_schema": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "City name"},
                },
                "required": ["location"],
            },
        },
    ],
)

print(f"Agent ID: {agent.id}")

environment = client.beta.environments.create(
    name="weather-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
    },
)

session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    title="Weather session",
)

with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [
                    {
                        "type": "text",
                        "text": "What is the weather in London?",
                    },
                ],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                print(f"\n[Using tool: {event.name}]")
            case "agent.custom_tool_use":
                print(f"\n[Custom tool call: {event.name}]")
                print(f"[Input: {json.dumps(event.input)}]")
                client.beta.sessions.events.send(
                    session.id,
                    events=[
                        {
                            "type": "user.custom_tool_result",
                            "custom_tool_use_id": event.id,
                            "content": [
                                {
                                    "type": "text",
                                    "text": "Weather in London: 15°C, partly cloudy, humidity 72%",
                                },
                            ],
                        },
                    ],
                )
            case "session.status_idle":
                if hasattr(event, "stop_reason") and event.stop_reason and event.stop_reason.type == "requires_action":
                    continue
                print("\n\nAgent finished.")
                break

08_update_agent

Create an agent, then update its name and system prompt. Retrieve and list agents.

from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="My Agent v1",
    model="claude-haiku-4-5",
    system="You are a helpful assistant.",
    tools=[
        {"type": "agent_toolset_20260401"},
    ],
)

print(f"Created: {agent.id}, version: {agent.version}")

updated = client.beta.agents.update(
    agent.id,
    version=agent.version,
    name="My Agent v2",
    system="You are a senior Python developer. Write production-quality code.",
)

print(f"Updated: {updated.id}, version: {updated.version}")

retrieved = client.beta.agents.retrieve(agent.id)
print(f"Name: {retrieved.name}")
print(f"System: {retrieved.system}")

agents = client.beta.agents.list()
print(f"\nTotal agents: {len(agents.data)}")
for a in agents.data[:3]:
    print(f"  - {a.name} (v{a.version})")

09_list_sessions

List all sessions and display their status, title, and token usage.

from anthropic import Anthropic

client = Anthropic()

sessions = client.beta.sessions.list()

print(f"Total sessions: {len(sessions.data)}")
for s in sessions.data[:5]:
    print(f"  ID: {s.id}")
    print(f"  Status: {s.status}")
    print(f"  Title: {s.title}")
    if s.usage:
        print(f"  Input tokens: {s.usage.input_tokens}")
        print(f"  Output tokens: {s.usage.output_tokens}")
    print()

10_web_search

Create a research agent that searches the web and summarizes results.

from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="Research Agent",
    model="claude-haiku-4-5",
    system="You are a research assistant. Search the web for information.",
    tools=[
        {"type": "agent_toolset_20260401"},
    ],
)

environment = client.beta.environments.create(
    name="research-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
    },
)

session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    title="Research session",
)

with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [
                    {
                        "type": "text",
                        "text": "Search the web for the latest Python 3.13 features and summarize them in 3 bullet points",
                    },
                ],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                print(f"\n[Using tool: {event.name}]")
            case "session.status_idle":
                print("\n\nAgent finished.")
                break

11_multi_turn

Run multiple turns in the same session — the agent remembers context from previous interactions.

from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="Multi Turn Agent",
    model="claude-haiku-4-5",
    system="You are a helpful coding assistant.",
    tools=[
        {"type": "agent_toolset_20260401"},
    ],
)

environment = client.beta.environments.create(
    name="multi-turn-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
    },
)

session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    title="Multi turn session",
)

print(f"Session ID: {session.id}")

messages = [
    "Create a file called greeting.py that prints 'Hello World'",
    "Now modify greeting.py to accept a name as a command line argument",
    "Run greeting.py with the argument 'Claude'",
]

for i, message in enumerate(messages):
    print(f"\n--- Turn {i + 1}: {message} ---\n")

    with client.beta.sessions.events.stream(session.id) as stream:
        client.beta.sessions.events.send(
            session.id,
            events=[
                {
                    "type": "user.message",
                    "content": [{"type": "text", "text": message}],
                },
            ],
        )

        for event in stream:
            match event.type:
                case "agent.message":
                    for block in event.content:
                        print(block.text, end="")
                case "agent.tool_use":
                    print(f"\n[Using tool: {event.name}]")
                case "session.status_idle":
                    print("\n\nAgent finished turn.")
                    break

12_environment_setup

Pre-install pip packages (pandas, numpy) in the cloud environment before the agent runs.

from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="Data Science Agent",
    model="claude-haiku-4-5",
    system="You are a data science assistant.",
    tools=[
        {"type": "agent_toolset_20260401"},
    ],
)

environment = client.beta.environments.create(
    name="data-science-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
        "packages": {
            "pip": ["pandas", "numpy"],
        },
    },
)

print(f"Environment ID: {environment.id}")

session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    title="Data science session",
)

with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [
                    {
                        "type": "text",
                        "text": "Create a Python script that generates random sales data with pandas and saves a summary to sales_summary.txt",
                    },
                ],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                print(f"\n[Using tool: {event.name}]")
            case "session.status_idle":
                print("\n\nAgent finished.")
                break

13_interrupt_session

Interrupt a running agent mid-execution by sending a user.interrupt event.

from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="Interruptable Agent",
    model="claude-haiku-4-5",
    system="You are a helpful coding assistant.",
    tools=[
        {"type": "agent_toolset_20260401"},
    ],
)

environment = client.beta.environments.create(
    name="interrupt-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
    },
)

session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    title="Interrupt session",
)

print(f"Session ID: {session.id}")

with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [
                    {
                        "type": "text",
                        "text": "Write a Python script that prints numbers 1 to 5",
                    },
                ],
            },
        ],
    )

    tool_count = 0
    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                tool_count += 1
                print(f"\n[Using tool: {event.name}]")
                if tool_count >= 1:
                    print("\n[Sending interrupt...]")
                    client.beta.sessions.events.send(
                        session.id,
                        events=[{"type": "user.interrupt"}],
                    )
            case "session.status_idle":
                print("\n\nAgent stopped.")
                break

14_track_usage

Track token usage (input, output, cache) after a session completes.

from anthropic import Anthropic

client = Anthropic()

agent = client.beta.agents.create(
    name="Usage Tracker Agent",
    model="claude-haiku-4-5",
    system="You are a helpful assistant.",
    tools=[
        {"type": "agent_toolset_20260401"},
    ],
)

environment = client.beta.environments.create(
    name="usage-env",
    config={
        "type": "cloud",
        "networking": {"type": "unrestricted"},
    },
)

session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    title="Usage tracking session",
)

with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [
                    {
                        "type": "text",
                        "text": "Write a one-line Python script that prints the current date and run it",
                    },
                ],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    print(block.text, end="")
            case "agent.tool_use":
                print(f"\n[Using tool: {event.name}]")
            case "session.status_idle":
                print("\n\nAgent finished.")
                break

result = client.beta.sessions.retrieve(session.id)
print(f"\n--- Usage Report ---")
print(f"Session ID: {result.id}")
print(f"Status: {result.status}")
if result.usage:
    print(f"Input tokens: {result.usage.input_tokens}")
    print(f"Output tokens: {result.usage.output_tokens}")
    print(f"Cache creation tokens: {result.usage.cache_creation}")
    print(f"Cache read tokens: {result.usage.cache_read_input_tokens}")

Summary