Agent Runtime Protocol

Swap the engine that runs an agent turn — built-in, custom, or plugin runtime.

python
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.

Diagram

resolve_runtime ‘praisonai’

resolve_runtime ‘echo’

entry-point praisonai.runtimes

Agent + prompt

Runtime Registry

Built-in PraisonAIRuntime

Custom Runtime

Plugin Runtime

RuntimeResult / RuntimeDelta

Quick Start

  1. 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())
  2. 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())
  3. 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.

Diagram
AgentRuntimeProtocolRuntimeRegistryYour AppUserAgentRuntimeProtocolRuntimeRegistryYour AppUser“Who built the Eiffel Tower?”resolve_runtime(“praisonai”)PraisonAIRuntime instanceawait run_turn(prompt)RuntimeResult(content=…)response
StageWhat happens
DiscoverRegistry auto-loads built-ins and plugin runtimes from praisonai.runtimes entry points
Resolveresolve_runtime("praisonai") returns an AgentRuntimeProtocol instance
Executeawait runtime.run_turn(prompt) returns a RuntimeResult; runtime.stream_turn(prompt) yields RuntimeDeltas
Fail-closedUnknown runtime IDs raise ValueError with the list of available IDs

Picking a Runtime

Diagram

No – default behaviour

Yes – in-process

Yes – ship as a package

What do you need?

Different engine?

resolve_runtime praisonai

Implement AgentRuntimeProtocol\nregister_runtime my-id factory

Declare praisonai.runtimes\nentry point in pyproject.toml

✅ RuntimeResult

You want…Use
Default behaviour — wraps existing Agentresolve_runtime("praisonai")
Different engine in-processImplement AgentRuntimeProtocol, register_runtime("my-id", factory)
Different engine shipped as a packageDeclare a praisonai.runtimes entry point

Configuration Options

AgentRuntimeProtocol

Any object implementing these three methods is a valid runtime:

MethodSignatureNotes
supports(model_ref=None) -> boolReturn True if this runtime can handle the given model
run_turn(prompt, *, system_prompt=None, model_ref=None, **kwargs) -> RuntimeResultAsync; 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

FieldTypeDefaultNotes
runtime_idstrrequiredUnique runtime identifier
metadataDict[str, Any]{}Optional runtime metadata

RuntimeResult

FieldTypeDefaultNotes
contentstrrequiredAgent response text
metadataDict[str, Any]{}Includes model, agent_id, runtime
errorOptional[str]NoneSet on failure — runtime returns errors, does not raise

RuntimeDelta

FieldTypeDefaultNotes
typestrrequired"text", "tool_call", "thinking", "error"
contentstr""Delta payload
metadataDict[str, Any]{}Runtime metadata

RuntimeRegistryEntry

FieldTypeDefaultNotes
runtime_idstrrequired
display_nameOptional[str]falls back to runtime_id
descriptionOptional[str]None
is_builtinboolFalseTrue for the praisonai runtime
metadataDict[str, Any]{}

RuntimeRegistry methods

MethodPurpose
register(runtime_id, factory_or_instance, **metadata)Register a runtime; raises ValueError if already registered
unregister(runtime_id) -> boolRemove a runtime and any aliases pointing to it
is_registered(runtime_id) -> boolCheck if registered (or aliased)
is_available(runtime_id) -> boolAlias 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

python
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.

toml
[project.entry-points."praisonai.runtimes"]
my-runtime = "my_package.runtime:MyRuntimeClass"
python
# 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

python
from praisonaiagents.runtime.registry import add_runtime_alias

add_runtime_alias("default", "praisonai")

Replace the default runtime

python
from praisonaiagents.runtime.registry import unregister_runtime
from praisonaiagents import register_runtime

unregister_runtime("praisonai")
register_runtime("praisonai", lambda: MyCustomRuntime())

List available runtimes

python
from praisonaiagents import list_runtimes

for runtime_id in list_runtimes():
    print(runtime_id)

Check if a runtime is available

python
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.


Related