Agent Runtime Protocol
Swap the engine that runs an agent turn — built-in, custom, or plugin runtime.
from praisonaiagents import Agent
agent = Agent(name="runtime-agent", instructions="Execute tasks via the runtime protocol.")
agent.start("Process this request using the standard runtime protocol.")One protocol, many runtimes — pick the engine that runs a single agent turn without changing your Agent code.
The user swaps runtimes without changing agent code; the registry resolves the engine and streams RuntimeResult deltas.
Quick Start
- Run a turn with the built-in runtime
python import asyncio from praisonaiagents import resolve_runtime async def main(): runtime = resolve_runtime("praisonai") result = await runtime.run_turn("Who built the Eiffel Tower?") print(result.content) asyncio.run(main()) - Stream the response
python import asyncio from praisonaiagents import resolve_runtime async def main(): runtime = resolve_runtime("praisonai") async for delta in runtime.stream_turn("Tell me a story"): if delta.type == "text": print(delta.content, end="") asyncio.run(main()) - Register and use a custom runtime
python import asyncio from praisonaiagents import register_runtime, resolve_runtime from praisonaiagents.runtime import RuntimeResult, RuntimeDelta class EchoRuntime: def supports(self, model_ref=None): return True async def run_turn(self, prompt, **kwargs): return RuntimeResult(content=f"echo: {prompt}") async def stream_turn(self, prompt, **kwargs): yield RuntimeDelta(type="text", content=f"echo: {prompt}") register_runtime("echo", lambda: EchoRuntime()) async def main(): runtime = resolve_runtime("echo") result = await runtime.run_turn("Hello!") print(result.content) asyncio.run(main())
How It Works
The registry auto-discovers built-in and plugin runtimes, then hands back an instance implementing AgentRuntimeProtocol.
| Stage | What happens |
|---|---|
| Discover | Registry auto-loads built-ins and plugin runtimes from praisonai.runtimes entry points |
| Resolve | resolve_runtime("praisonai") returns an AgentRuntimeProtocol instance |
| Execute | await runtime.run_turn(prompt) returns a RuntimeResult; runtime.stream_turn(prompt) yields RuntimeDeltas |
| Fail-closed | Unknown runtime IDs raise ValueError with the list of available IDs |
Picking a Runtime
| You want… | Use |
|---|---|
| Default behaviour — wraps existing Agent | resolve_runtime("praisonai") |
| Different engine in-process | Implement AgentRuntimeProtocol, register_runtime("my-id", factory) |
| Different engine shipped as a package | Declare a praisonai.runtimes entry point |
Configuration Options
AgentRuntimeProtocol
Any object implementing these three methods is a valid runtime:
| Method | Signature | Notes |
|---|---|---|
supports | (model_ref=None) -> bool | Return True if this runtime can handle the given model |
run_turn | (prompt, *, system_prompt=None, model_ref=None, **kwargs) -> RuntimeResult | Async; run one agent turn |
stream_turn | (prompt, **kwargs) -> AsyncIterator[RuntimeDelta] | Async generator; stream one agent turn |
**kwargs understood by the built-in runtime: tools, max_tokens, temperature.
RuntimeConfig
| Field | Type | Default | Notes |
|---|---|---|---|
runtime_id | str | required | Unique runtime identifier |
metadata | Dict[str, Any] | {} | Optional runtime metadata |
RuntimeResult
| Field | Type | Default | Notes |
|---|---|---|---|
content | str | required | Agent response text |
metadata | Dict[str, Any] | {} | Includes model, agent_id, runtime |
error | Optional[str] | None | Set on failure — runtime returns errors, does not raise |
RuntimeDelta
| Field | Type | Default | Notes |
|---|---|---|---|
type | str | required | "text", "tool_call", "thinking", "error" |
content | str | "" | Delta payload |
metadata | Dict[str, Any] | {} | Runtime metadata |
RuntimeRegistryEntry
| Field | Type | Default | Notes |
|---|---|---|---|
runtime_id | str | required | |
display_name | Optional[str] | falls back to runtime_id | |
description | Optional[str] | None | |
is_builtin | bool | False | True for the praisonai runtime |
metadata | Dict[str, Any] | {} |
RuntimeRegistry methods
| Method | Purpose |
|---|---|
register(runtime_id, factory_or_instance, **metadata) | Register a runtime; raises ValueError if already registered |
unregister(runtime_id) -> bool | Remove a runtime and any aliases pointing to it |
is_registered(runtime_id) -> bool | Check if registered (or aliased) |
is_available(runtime_id) -> bool | Alias for is_registered |
list_runtimes() -> List[RuntimeRegistryEntry] | All registered runtimes with metadata |
list_names() -> List[str] | IDs only |
get_entry(runtime_id) -> Optional[RuntimeRegistryEntry] | Metadata for one runtime |
resolve(runtime_id, config_overrides=None) | Instantiate runtime; raises ValueError with available IDs if not found |
add_alias(alias, canonical_runtime_id) | Alias an existing runtime |
clear() | Reset (for tests) |
Common Patterns
Implement a custom runtime
from praisonaiagents import register_runtime
from praisonaiagents.runtime import RuntimeResult, RuntimeDelta
class MyRuntime:
def supports(self, model_ref=None):
return True
async def run_turn(self, prompt, *, system_prompt=None, model_ref=None, **kwargs):
response = await my_llm_client.complete(prompt, system=system_prompt)
if response.error:
return RuntimeResult(content="", error=response.error)
return RuntimeResult(content=response.text)
async def stream_turn(self, prompt, **kwargs):
async for token in my_llm_client.stream(prompt):
yield RuntimeDelta(type="text", content=token)
register_runtime("my-llm", lambda: MyRuntime())Plugin runtime via pyproject.toml
Ship your runtime as an installable package. The registry discovers it automatically at import time.
[project.entry-points."praisonai.runtimes"]
my-runtime = "my_package.runtime:MyRuntimeClass"# my_package/runtime.py
from praisonaiagents.runtime import RuntimeResult, RuntimeDelta
class MyRuntimeClass:
def supports(self, model_ref=None):
return True
async def run_turn(self, prompt, **kwargs):
return RuntimeResult(content=f"handled: {prompt}")
async def stream_turn(self, prompt, **kwargs):
yield RuntimeDelta(type="text", content=f"handled: {prompt}")Alias a runtime
from praisonaiagents.runtime.registry import add_runtime_alias
add_runtime_alias("default", "praisonai")Replace the default runtime
from praisonaiagents.runtime.registry import unregister_runtime
from praisonaiagents import register_runtime
unregister_runtime("praisonai")
register_runtime("praisonai", lambda: MyCustomRuntime())List available runtimes
from praisonaiagents import list_runtimes
for runtime_id in list_runtimes():
print(runtime_id)Check if a runtime is available
from praisonaiagents.runtime.registry import is_runtime_available
if is_runtime_available("my-runtime"):
runtime = resolve_runtime("my-runtime")Best Practices
Return errors, don't raise
The built-in runtime returns RuntimeResult(error=...) instead of raising. Custom runtimes should follow the same pattern so callers can handle failure uniformly with a single if result.error: check.
Use a factory for stateful runtimes
Pass register_runtime("id", lambda: MyRuntime()) so each resolve_runtime call gets a fresh instance. This is important for runtimes that hold per-turn state or open connections.
Emit typed deltas in stream_turn
stream_turn yields RuntimeDeltas — emit type="text" for tokens, type="error" on failure, type="tool_call" or type="thinking" for richer UIs. Always end with an error delta rather than raising.
Use registry.clear() only in tests
The registry is process-wide and thread-safe. Call RuntimeRegistry().clear() only in test teardown to avoid affecting other parts of your application.