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

Component	Responsibility	Location
Agent	Model, system prompt, tool definitions	Remote provider
Environment	Runtime container, packages, networking	Remote provider
Session	Running instance of agent-in-environment	Remote provider
Runtime	Protocol implementation, API client	Local (your code)

---

Protocol Methods

Agent Management

Method	Purpose	Returns
create_agent(config)	Deploy agent definition	agent_id: str
retrieve_agent(agent_id)	Get agent metadata	Dict[str, Any]
list_agents(**filters)	List all agents	List[Dict[str, Any]]
archive_agent(agent_id)	Mark agent inactive	None

Environment Management

Method	Purpose	Returns
create_environment(config)	Provision sandbox	environment_id: str
retrieve_environment(env_id)	Get environment metadata	Dict[str, Any]
list_environments(**filters)	List all environments	List[Dict[str, Any]]
archive_environment(env_id)	Mark environment inactive	None
delete_environment(env_id)	Destroy environment	None

Session Management

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

Event Handling

Method	Purpose	Returns
send_event(session_id, event)	Send message to session	None
stream_events(session_id)	Stream responses	AsyncIterator[Dict[str, Any]]
interrupt(session_id)	Stop current processing. ManagedBackendBase.interrupt() is a public no-op default on the base mixin, so subclasses that don’t override it still satisfy the protocol — the stream-cancel path depends on it	None

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

Kwarg	Type	Default	Behaviour
stream_live	bool	False	Forwarded to _execute_sync(prompt, stream_live=...) — drives incremental stdout and emitter.llm_response events on both Anthropic and Local backends.
timeout	float | None	None (unbounded; the per-instance self.timeout still governs the inner SDK call where applicable)	Enforced by asyncio.wait_for around the thread-pool future. On timeout, interrupt() is called on a daemon thread (best-effort) and asyncio.TimeoutError re-raises.

Fixed in PraisonAI PR #5774 (issue #5773). Before #5774, both execute() methods were one-liners of the shape await loop.run_in_executor(None, self._execute_sync, prompt), so every caller-supplied kwarg was silently discarded. ComputeManagedAgent honoured timeout, AnthropicManagedAgent used its instance-level self.timeout, and LocalManagedAgent was effectively unbounded. #5774 routes both through a shared ManagedBackendBase._execute_in_executor helper that honours stream_live and timeout and calls interrupt() on timeout. Unknown kwargs are still accepted (not raised as TypeError) to avoid breaking existing callers.

timeout= on execute() bounds the async wait, not the worker thread. On timeout the backend is interrupted (AnthropicManagedAgent.interrupt sends user.interrupt to the active session) but the Python thread already dispatched into _execute_sync cannot be killed.

Since PraisonAI PR #3908, agent.start(prompt, stream=True) yields chunks from backend.stream() as they arrive, on both async and sync callers. Earlier releases buffered the full response in the sync (no running event loop) path. See Streaming → Managed-backend streaming.

---

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

Resource Management

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)

Error Handling

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)

Event Processing

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)

Performance Optimization

• 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

---

Related

Sandboxed Agent

Local agent loop with optional tool sandboxing

Managed Agent Lifecycle

Complete CRUD operations for agents, environments, sessions

Streaming

How backend= streams each chunk incrementally from sync code