Skip to main content
Managed Runtime Protocol defines the interface for remote agent execution where the entire agent loop runs on provider infrastructure.

​
Quick Start

1

Basic Lifecycle

Create agent, environment, session, then send events and stream responses.
from praisonai import AnthropicManagedAgent, ManagedConfig

runtime = AnthropicManagedAgent(config=ManagedConfig(
    model="claude-sonnet-4-6",
    system="You are a coding assistant.",
))

# Create infrastructure
agent_id = await runtime.create_agent(config)
env_id = await runtime.create_environment(config)
session_id = await runtime.create_session(agent_id, env_id)

# Send message and stream responses
await runtime.send_event(session_id, {
    "type": "user.message",
    "content": [{"type": "text", "text": "Hello!"}]
})

async for event in runtime.stream_events(session_id):
    if event["type"] == "agent.message":
        print(event["content"])
2

With Advanced Configuration

Full configuration with packages, networking, and vaults.
from praisonai import AnthropicManagedAgent, ManagedConfig, NetworkingConfig, PackagesConfig

runtime = AnthropicManagedAgent(config=ManagedConfig(
    model="claude-sonnet-4-6",
    system="You are a data analysis expert.",
    packages=PackagesConfig(
        pip=["pandas", "numpy", "matplotlib"],
        npm=["express"]
    ),
    networking=NetworkingConfig(
        type="limited",
        allowed_hosts=["api.example.com"],
        allow_mcp_servers=True
    ),
    vault_ids=["vault_github_123"]  # OAuth credentials
))

# Full lifecycle with session management
agent_id = await runtime.create_agent(config)
env_id = await runtime.create_environment(config) 
session_id = await runtime.create_session(agent_id, env_id)

await runtime.send_event(session_id, {
    "type": "user.message",
    "content": [{"type": "text", "text": "Analyze sales data from GitHub repo"}]
})

# Handle events
async for event in runtime.stream_events(session_id):
    if event["type"] == "agent.message":
        print(f"Agent: {event['content']}")
    elif event["type"] == "session.status_idle":
        break

​
How It Works

ComponentResponsibilityLocation
AgentModel, system prompt, tool definitionsRemote provider
EnvironmentRuntime container, packages, networkingRemote provider
SessionRunning instance of agent-in-environmentRemote provider
RuntimeProtocol implementation, API clientLocal (your code)

​
Protocol Methods

​
Agent Management

MethodPurposeReturns
create_agent(config)Deploy agent definitionagent_id: str
retrieve_agent(agent_id)Get agent metadataDict[str, Any]
list_agents(**filters)List all agentsList[Dict[str, Any]]
archive_agent(agent_id)Mark agent inactiveNone

​
Environment Management

MethodPurposeReturns
create_environment(config)Provision sandboxenvironment_id: str
retrieve_environment(env_id)Get environment metadataDict[str, Any]
list_environments(**filters)List all environmentsList[Dict[str, Any]]
archive_environment(env_id)Mark environment inactiveNone
delete_environment(env_id)Destroy environmentNone

​
Session Management

MethodPurposeReturns
create_session(agent_id, env_id)Start agent sessionsession_id: str
retrieve_session(session_id)Get session statusDict[str, Any]
list_sessions(**filters)List all sessionsList[Dict[str, Any]]
archive_session(session_id)Mark session inactiveNone
delete_session(session_id)Delete session dataNone

​
Event Handling

MethodPurposeReturns
send_event(session_id, event)Send message to sessionNone
stream_events(session_id)Stream responsesAsyncIterator[Dict[str, Any]]
interrupt(session_id)Stop current processingNone

​
Common Patterns

​
Multi-Turn Conversation

# Create infrastructure once
agent_id = await runtime.create_agent(config)
env_id = await runtime.create_environment(config)
session_id = await runtime.create_session(agent_id, env_id)

# Multiple turns
for user_input in conversation:
    await runtime.send_event(session_id, {
        "type": "user.message",
        "content": [{"type": "text", "text": user_input}]
    })
    
    async for event in runtime.stream_events(session_id):
        if event["type"] == "agent.message":
            print(event["content"])
        elif event["type"] == "session.status_idle":
            break  # Ready for next input

​
Agent Version Pinning

# Pin to specific agent version
runtime = AnthropicManagedAgent(config=config)
runtime.agent_version = 5  # Use version 5

# Session will use pinned version
session_id = await runtime.create_session(agent_id, env_id)

​
Session Cleanup

# Archive preserves data but marks inactive
await runtime.archive_session(session_id)

# Delete removes all data permanently
await runtime.delete_session(session_id)

# List sessions with filters
active_sessions = await runtime.list_sessions(status="active")
archived_sessions = await runtime.list_sessions(status="archived")

​
Best Practices

Create agents and environments once, reuse across sessions:
# Create once
agent_id = await runtime.create_agent(config)
env_id = await runtime.create_environment(config)

# Reuse for multiple sessions
session1 = await runtime.create_session(agent_id, env_id)
session2 = await runtime.create_session(agent_id, env_id)
Handle network failures and provider errors gracefully:
try:
    await runtime.send_event(session_id, event)
except Exception as e:
    # Check if session still exists
    try:
        session = await runtime.retrieve_session(session_id)
        # Session exists, retry
    except:
        # Session lost, recreate
        session_id = await runtime.create_session(agent_id, env_id)
Handle different event types appropriately:
async for event in runtime.stream_events(session_id):
    match event["type"]:
        case "agent.message":
            handle_message(event["content"])
        case "agent.tool_use":
            handle_tool_call(event)
        case "session.status_idle":
            break  # Agent finished processing
        case "session.error":
            handle_error(event)
  • Reuse agent/environment definitions across sessions
  • Use appropriate session cleanup (archive vs delete)
  • Pin agent versions for consistent behavior
  • Filter list operations to reduce API calls

Sandboxed Agent

Local agent loop with optional tool sandboxing

Managed Agent Lifecycle

Complete CRUD operations for agents, environments, sessions