Runtime#

Source: openprogram/agentic_programming/runtime.py

The LLM runtime. Wraps an LLM provider, automatically computes context from the session DAG, calls the LLM, and writes the reply back to the DAG.

Class: Runtime#

class Runtime(call=None, model="default", max_retries=None, api_key=None, skills=None)

Constructor parameters#

Attributes#

Methods#

exec()#

Runtime.exec(content, context=None, response_format=None, model=None,
             tools=None, toolset=None, tools_source=None, tools_allow=None,
             tools_deny=None, tool_choice="auto", parallel_tool_calls=True,
             max_iterations=20, choices=None, timeout_s=None, on_retry=None,
             web_search=False, stream_fn=None) -> Any

Calls the LLM, with context computed automatically from the session DAG.

When called inside an @agentic_function:

1. Starting from the current function's DAG node, render_context uses expose / render_range to determine which historical nodes to read this time
2. render_dag_messages renders those nodes into messages
3. _call() is invoked to send the request
4. The reply is written into the llm node that exec() opened at the start of the call

When called with no DAG store installed (standalone scripts, no dispatcher): content is wrapped into a single user message and sent as a single-turn call; nothing is recorded.

A single @agentic_function can call exec() multiple times; each call is a new llm node on the DAG.

Parameters

Content block format

{"type": "text",  "text": "Find the login button."}
{"type": "image", "path": "screenshot.png"}
{"type": "image", "data": "<base64>", "mime_type": "image/png"}
{"type": "video", "path": "clip.mp4"}
{"type": "audio", "path": "recording.wav"}

Media blocks take either a path (read and base64-encoded automatically, mime type guessed from the extension) or inline data + mime_type. A text block may carry "role": "system" to contribute to the system prompt, and text/image blocks accept cache_control for provider prompt caching. Unknown block types are skipped silently.

Return value

With response_format=None, returns the LLM reply as str. With a structured response_format, returns the locally validated Python JSON value (dict, list, scalar, or None). With choices, returns the parsed decision result (the return value of the selected function, or the selected value itself).

Exceptions

• RuntimeError — the runtime is closed (close() was called)
• TypeError / NotImplementedError — raised immediately, never retried (programming errors: wrong call signature, no provider configured)
• LLMError — raised when retries are exhausted or a non-retryable error is hit; structured fields include reason / retryable / http_status / retry_after_s / attempts / elapsed_s / provider / model, etc.
• StructuredOutputSchemaError — invalid public schema or envelope; stable code="invalid_schema".
• StructuredOutputUnsupportedError — no verified lossless provider/fallback contract; stable code="unsupported".
• StructuredOutputValidationError — invalid JSON, schema mismatch, or hidden-submit contract failure; stable codes are invalid_json, validation_failed, missing_submission, and mixed_submission.
• StructuredOutputGenerationError — refusal or incomplete provider terminal; stable codes are refusal and incomplete.

All structured-output errors expose the stable .code and bounded .issues fields. Provider candidate text is not part of the public diagnostic contract.

async_exec()#

await Runtime.async_exec(content, context=None, response_format=None, model=None,
                         timeout_s=None, on_retry=None) -> Any

The async version of exec(). It has the same response_format=None text return and structured Python JSON return/error contract. Internally it calls _async_call(); a call= function may be synchronous or asynchronous, and the default provider path uses the same AgentSession structured lifecycle. Same timeout_s / on_retry semantics as exec(); retries sleep with asyncio.sleep, so external cancellation works. It has no public tool-loop parameters.

_call()#

Runtime._call(content, model="default", response_format=None) -> Any

The method that actually calls the LLM once (no retry — exec() wraps it in the retry loop). The default implementation routes through the provider layer (AgentSession) when a provider model or call= function is configured, and raises NotImplementedError otherwise. Override this method when subclassing.

Parameters

Return value

Raw text for ordinary calls; the validated Python JSON value when a structured format is active.

_async_call()#

await Runtime._async_call(content, model="default", response_format=None) -> Any

The async version of _call(). Override this method when subclassing to support an async provider.

close()#

Releases resources and ends the session; after close(), exec() raises RuntimeError. Runtime is also a context manager (with Runtime(...) as rt: closes on exit). Subclasses override it to clean up provider-specific resources.

Asking the user#

When a front-end session is connected, a runtime can block on user input mid-function: runtime.ask(prompt, options=..., multi=..., questions=[...], timeout=300.0, default=None) (one question or several in one card), runtime.confirm(prompt, default=False) (yes/no), and runtime.form(prompt, fields) (multi-field form). runtime.can_ask() reports whether anyone is there to answer (False on headless runs). A declined question raises UserDeclined; a timeout returns default when given, else raises AskTimeout.

Usage#

Option 1: Pass in a call function#

from openprogram import agentic_function
from openprogram.agentic_programming.runtime import Runtime

def my_llm(content, model="sonnet", response_format=None):
    # Convert content into your provider's format and send the request
    texts = [b["text"] for b in content if b["type"] == "text"]
    return call_my_api("\n".join(texts), model=model)

runtime = Runtime(call=my_llm, model="sonnet")

@agentic_function
def observe(task):
    """Look at the screen."""
    return runtime.exec(content=[
        {"type": "text", "text": f"Find: {task}"},
        {"type": "image", "path": "screenshot.png"},
    ])

Option 2: Subclass#

class MyAnthropicRuntime(Runtime):
    def __init__(self, api_key, model="sonnet"):
        super().__init__(model=model)
        self.client = anthropic.Anthropic(api_key=api_key)

    def _call(self, content, model="sonnet", response_format=None):
        messages_content = []
        for block in content:
            if block["type"] == "text":
                messages_content.append({"type": "text", "text": block["text"]})
        response = self.client.messages.create(
            model=model, max_tokens=1024,
            messages=[{"role": "user", "content": messages_content}],
        )
        return response.content[0].text

runtime = MyAnthropicRuntime(api_key="sk-...", model="claude-sonnet-4-6")

Multiple Runtimes coexisting#

fast = Runtime(call=gemini_call, model="gemini-2.5-flash")
strong = Runtime(call=claude_call, model="sonnet")

@agentic_function
def observe(task):
    """Quick observation with cheap model."""
    return fast.exec(content=[...])

@agentic_function
def plan(goal):
    """Complex planning with strong model."""
    return strong.exec(content=[...])

Retry mechanism#

exec() and async_exec() have built-in automatic retries to handle transient LLM API errors (network timeouts, rate limits, server errors, etc.).

Configuration#

# Default: max_retries=None → read env var OPENPROGRAM_MAX_RETRIES, defaulting to 6 if unset
rt = Runtime(call=my_llm)

# No retries (raise an exception on the first failure)
rt = Runtime(call=my_llm, max_retries=1)

# Multiple retries (for an unstable API)
rt = Runtime(call=my_llm, max_retries=5)

Behavior rules#

Error report format#

When all retries are exhausted, the LLMError raised contains the error information for each attempt; its structured fields (reason / retryable / http_status / attempts / elapsed_s, etc.) can be read directly:

LLMError: exec() failed after 3 attempt(s):
Attempt 1: ConnectionError: timeout
Attempt 2: RateLimitError: 429 Too Many Requests
Attempt 3: ConnectionError: timeout

The boundaries of retrying#

max_retries only handles transient failures at the API level (network timeouts, rate limits, etc.). If the problem lies in the function's own logic or output format, retrying won't fix it — edit the function code directly; see skills/agentic-programming/SKILL.md.