Managed Agent Lifecycle

Complete CRUD operations for agents, environments, and sessions

Managed agent lifecycle provides full control over agent definitions, environments, and running sessions with comprehensive CRUD operations.

from praisonaiagents import Agent

agent = Agent(name="ops-agent", instructions="Manage hosted agent resources.")
agent.start("List active sessions and archive the stale ones.")

The user manages agents through CRUD APIs; definitions, environments, and sessions move through create → run → archive flows.

​

CRUD Operations

​

Quick Start

1. Basic Agent Lifecycle

   Create, use, and manage a simple agent.

   from praisonai import AnthropicManagedAgent, ManagedConfig
   
   managed = AnthropicManagedAgent(config=ManagedConfig(
       model="claude-sonnet-4-6",
       system="You are a helpful coding assistant.",
   ))
   
   # Create agent and environment
   agent_id = await managed.create_agent(managed._cfg)
   env_id = await managed.create_environment(managed._cfg)
   
   # Create session
   session_id = await managed.create_session(agent_id, env_id)
   
   # Use session...
   await managed.send_event(session_id, {"type": "user.message", "content": [{"type": "text", "text": "Hello!"}]})
   
   # Clean up when done
   await managed.archive_session(session_id)
   await managed.archive_agent()  # Clears cached IDs
   
2. Agent Version Pinning

   Pin agents to specific versions for consistent behavior.

   # Create agent and set version preference
   managed = AnthropicManagedAgent(config=config)
   agent_id = await managed.create_agent(config)
   
   # Pin to specific version
   managed.agent_version = 3
   
   # Sessions will use version 3
   session_id = await managed.create_session(agent_id, env_id)
   
   # List all versions of this agent
   versions = await managed.list_agent_versions()
   # [{"version": 1, "created_at": "..."}, {"version": 2, "..."}, {"version": 3, "..."}]
   

---

​

How It Works

---

​

Agent Management

​

Agent CRUD Operations

Method	Purpose	State Changes
create_agent(config)	Deploy agent definition	Sets agent_id, agent_version
retrieve_agent()	Get agent metadata	None
list_agents(**kwargs)	List all agents	None
archive_agent()	Mark agent inactive	Clears agent_id, agent_version, _session_id
list_agent_versions()	Get version history	None

​

Agent Version Management

# Create agent (version 1)
agent_id = await managed.create_agent(config)
print(f"Created agent {agent_id} v{managed.agent_version}")

# Update agent (creates version 2)
updated_config = ManagedConfig(
    model="claude-haiku-4-5",  # Changed model
    system="Updated system prompt"
)
agent_id = await managed.create_agent(updated_config)
print(f"Updated agent {agent_id} v{managed.agent_version}")

# Pin to specific version
managed.agent_version = 1
session_id = await managed.create_session(agent_id, env_id)  # Uses version 1

# List all versions
versions = await managed.list_agent_versions()
for v in versions:
    print(f"Version {v['version']} created at {v['created_at']}")

​

Agent Metadata Retrieval

# Get current agent details
agent_info = await managed.retrieve_agent()
print(f"Agent: {agent_info['name']} ({agent_info['model']})")
print(f"System: {agent_info['system'][:50]}...")
print(f"Created: {agent_info['created_at']}")

# List agents with filtering
agents = await managed.list_agents(limit=10, name_contains="coding")
for agent in agents:
    print(f"{agent['name']}: {agent['id']} (v{agent['version']})")

---

​

Environment Management

​

Environment CRUD Operations

Method	Purpose	State Changes
create_environment(config)	Provision sandbox	Sets environment_id
retrieve_environment()	Get environment metadata	None
list_environments(**kwargs)	List all environments	None
archive_environment()	Mark environment inactive	Clears environment_id, _session_id
delete_environment()	Destroy environment permanently	Clears environment_id, _session_id

​

Environment Lifecycle

# Create environment
env_id = await managed.create_environment(ManagedConfig(
    packages={"pip": ["pandas", "numpy"]},
    networking={"type": "limited", "allowed_hosts": ["api.openai.com"]}
))

# Check environment status
env_info = await managed.retrieve_environment()
print(f"Environment {env_info['id']} status: {env_info['status']}")
print(f"Packages: {env_info.get('packages', {})}")

# List environments
environments = await managed.list_environments(status="active")
for env in environments:
    print(f"Environment {env['name']}: {env['id']}")

# Archive vs Delete
await managed.archive_environment()  # Preserves data, marks inactive
# OR
await managed.delete_environment()   # Destroys environment completely

---

​

Session Management

​

Session CRUD Operations

Method	Purpose	Returns
create_session(agent_id, env_id)	Start agent session	session_id: str
retrieve_session(session_id?)	Get session metadata	Dict[str, Any]
list_sessions(**filters)	List sessions with filtering	List[Dict[str, Any]]
archive_session(session_id?)	Mark session inactive	None
delete_session(session_id?)	Delete session permanently	None

​

Session Lifecycle

# Create session with agent version pinning
managed.agent_version = 2  # Pin to version 2
session_id = await managed.create_session(agent_id, env_id)

# Check session status
session_info = await managed.retrieve_session(session_id)
print(f"Session {session_info['id']} status: {session_info['status']}")
print(f"Agent version: {session_info.get('agent_version')}")
print(f"Usage: {session_info.get('usage', {})}")

# List sessions with filtering
active_sessions = await managed.list_sessions(status="active", limit=10)
for session in active_sessions:
    print(f"Session {session['title']}: {session['id']}")

# Session cleanup
await managed.archive_session(session_id)  # Preserves conversation history
# OR  
await managed.delete_session(session_id)   # Removes all data permanently

---

​

Common Patterns

​

Multi-Agent System

class MultiAgentSystem:
    def __init__(self):
        self.agents = {}
        self.environments = {}
        self.sessions = {}
    
    async def create_coding_agent(self):
        """Create specialized coding agent."""
        managed = AnthropicManagedAgent(config=ManagedConfig(
            name="coder",
            model="claude-sonnet-4-6",
            system="You are an expert Python developer.",
            tools=[{"type": "agent_toolset_20260401"}]
        ))
        
        agent_id = await managed.create_agent(managed._cfg)
        env_id = await managed.create_environment(ManagedConfig(
            env_name="python-dev",
            packages={"pip": ["pandas", "numpy", "pytest"]}
        ))
        
        self.agents["coder"] = {"managed": managed, "agent_id": agent_id}
        self.environments["python-dev"] = env_id
        
        return managed, agent_id, env_id
    
    async def create_research_agent(self):
        """Create specialized research agent."""
        managed = AnthropicManagedAgent(config=ManagedConfig(
            name="researcher", 
            model="claude-sonnet-4-6",
            system="You are a research assistant.",
            tools=[{"type": "web_search_20260401"}]
        ))
        
        agent_id = await managed.create_agent(managed._cfg)
        env_id = await managed.create_environment(ManagedConfig(
            env_name="research",
            networking={"type": "unrestricted"}
        ))
        
        self.agents["researcher"] = {"managed": managed, "agent_id": agent_id}
        self.environments["research"] = env_id
        
        return managed, agent_id, env_id

​

Session Pool Management

class SessionPool:
    def __init__(self, managed_agent, max_sessions=5):
        self.managed = managed_agent
        self.max_sessions = max_sessions
        self.active_sessions = []
    
    async def get_session(self):
        """Get available session or create new one."""
        # Clean up idle sessions
        await self._cleanup_idle_sessions()
        
        if len(self.active_sessions) < self.max_sessions:
            # Create new session
            session_id = await self.managed.create_session(
                self.managed.agent_id, 
                self.managed.environment_id
            )
            self.active_sessions.append({
                "id": session_id,
                "created_at": time.time(),
                "last_used": time.time()
            })
            return session_id
        else:
            # Reuse oldest session
            oldest = min(self.active_sessions, key=lambda s: s["last_used"])
            oldest["last_used"] = time.time()
            return oldest["id"]
    
    async def _cleanup_idle_sessions(self):
        """Archive sessions idle for >30 minutes."""
        cutoff = time.time() - 1800  # 30 minutes
        idle_sessions = [s for s in self.active_sessions if s["last_used"] < cutoff]
        
        for session in idle_sessions:
            await self.managed.archive_session(session["id"])
            self.active_sessions.remove(session)

​

Resource Monitoring

async def monitor_resources(managed: AnthropicManagedAgent):
    """Monitor and report resource usage."""
    # List all resources
    agents = await managed.list_agents()
    environments = await managed.list_environments()
    sessions = await managed.list_sessions()
    
    print(f"📊 Resource Usage Summary")
    print(f"Agents: {len(agents)} total")
    print(f"Environments: {len(environments)} total") 
    print(f"Sessions: {len(sessions)} total")
    
    # Active sessions breakdown
    active_sessions = [s for s in sessions if s["status"] == "active"]
    archived_sessions = [s for s in sessions if s["status"] == "archived"]
    
    print(f"\nSession Status:")
    print(f"  Active: {len(active_sessions)}")
    print(f"  Archived: {len(archived_sessions)}")
    
    # Environment status
    for env in environments:
        print(f"\nEnvironment {env['name']}:")
        print(f"  Status: {env['status']}")
        print(f"  Packages: {len(env.get('packages', {}))}")

---

​

Best Practices

Resource Management

Always clean up resources to avoid billing and quota issues:

# Good: Explicit cleanup
try:
    session_id = await managed.create_session(agent_id, env_id)
    # Use session...
finally:
    await managed.archive_session(session_id)
    await managed.archive_agent()  # Clears cached IDs

# Good: Context manager pattern
class ManagedAgentContext:
    def __init__(self, managed_agent):
        self.managed = managed_agent
        
    async def __aenter__(self):
        self.agent_id = await self.managed.create_agent(self.managed._cfg)
        self.env_id = await self.managed.create_environment(self.managed._cfg)
        self.session_id = await self.managed.create_session(self.agent_id, self.env_id)
        return self
        
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.managed.archive_session(self.session_id)
        await self.managed.archive_agent()

Version Management

Use agent versioning for reliable deployments:

# Pin to specific versions in production
production_agent_version = 5
managed.agent_version = production_agent_version

# Test new versions in staging
staging_managed = AnthropicManagedAgent(config=new_config)
test_agent_id = await staging_managed.create_agent(staging_managed._cfg)
# If tests pass, promote to production

State Management

Understand state clearing behavior:

# archive_agent() clears ALL cached IDs
await managed.archive_agent()
# Now: agent_id=None, agent_version=None, _session_id=None

# archive_environment() clears environment + session IDs
await managed.archive_environment() 
# Now: environment_id=None, _session_id=None (agent_id preserved)

# delete_environment() clears environment + session IDs  
await managed.delete_environment()
# Now: environment_id=None, _session_id=None (agent_id preserved)

Error Recovery

Handle failures gracefully with proper resource tracking:

try:
    agent_id = await managed.create_agent(config)
    env_id = await managed.create_environment(config)
    session_id = await managed.create_session(agent_id, env_id)
except Exception as e:
    # Clean up any partially created resources
    if hasattr(managed, 'agent_id') and managed.agent_id:
        await managed.archive_agent()
    if hasattr(managed, 'environment_id') and managed.environment_id:
        await managed.delete_environment()
    raise e

---

​

Thread safety

A shared AnthropicManagedAgent or LocalManagedAgent is safe to execute() from multiple threads; agent / environment / session creation is guarded by per-instance locks, so only one of any resource is ever created. Each execute() call takes a per-call timeout= and stream_live= kwarg since PR #5774 — see Managed Runtime Protocol → Per-call execute() kwargs. On timeout the backend’s interrupt() is called on a daemon thread and asyncio.TimeoutError re-raises.

from concurrent.futures import ThreadPoolExecutor
from praisonai import AnthropicManagedAgent, ManagedConfig

managed = AnthropicManagedAgent(config=ManagedConfig(name="shared-agent"))

with ThreadPoolExecutor(max_workers=8) as pool:
    for _ in range(8):
        pool.submit(managed.execute, "Summarise today's news")

# Exactly one agent / environment / session was created in Anthropic's cloud.

The locks wrap only check + create + assign, so a stream reader is never blocked by another caller’s create. Long-running stream reads hold no lock.

---

​

Related

Managed Runtime Protocol

Core protocol defining agent lifecycle operations

Managed Vault

Secure credential storage for agent integrations