Managed Runtime Protocol

Protocol for remote agent runtime where the entire loop runs on managed infrastructure

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

from praisonaiagents import Agent
from praisonai import AnthropicManagedAgent, ManagedConfig

runtime = AnthropicManagedAgent(config=ManagedConfig(model="claude-sonnet-4-6"))
agent = Agent(name="Remote Coder", instructions="Write and run code on managed infrastructure.", llm=runtime)
agent.start("Create a hello-world script and run it.")

The user delegates a task; the full agent loop executes on managed provider infrastructure via the runtime protocol.

Protocol Choice

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

Protocol Methods

Agent Management

Environment Management

Session Management

Event Handling

Per-call execute() kwargs

AnthropicManagedAgent.execute() and LocalManagedAgent.execute() honour two per-call keyword arguments that ComputeManagedAgent has always honoured.

import asyncio
from praisonai.integrations.managed_agents import AnthropicManagedAgent

async def main():
    backend = AnthropicManagedAgent()
    # On expiry the backend is interrupted (best-effort) before asyncio.TimeoutError re-raises.
    answer = await backend.execute("Summarise today's news", timeout=30)
    live = await backend.execute("Write a report", stream_live=True)

asyncio.run(main())

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

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)

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)

Related