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

Parameter	Type	Default	Description
call	Callable | None	None	A user-supplied LLM function. Signature: fn(content: list[dict], model: str, response_format: dict) -> str. Internally it is wrapped into the standard provider path via a CallableModel, so DAG recording and history rendering work the same as with a real provider. If neither call nor a "provider:model_id" model is given, subclass and override _call()
model	str	"default"	The default model. Two forms: "provider:model_id" (e.g. "anthropic:claude-sonnet-4-6") resolves through openprogram.providers and streams via the provider layer; any other string is only meaningful together with call= or a subclass. Unknown "provider:model_id" values raise ValueError
max_retries	int | None	None	Maximum number of exec() attempts (including the first call, and must be >= 1). None = read the environment variable OPENPROGRAM_MAX_RETRIES, defaulting to 6 if unset
api_key	str | None	None	API key for the provider path. None = resolved from the credential store (openprogram providers login)
skills	bool | list[str] | None	None	Skill discovery for the system prompt. None / False = disabled; True = probe the default skill directories (user + repo); list[str] = explicit directory list. When enabled, an <available_skills> block is appended to the system prompt on every exec()

Attributes

Attribute	Type	Description
model	str	The default model name
max_retries	int	The resolved retry budget
system	str	Assignable system prompt used by exec() on the provider path (the @agentic_function(system=...) decorator sets it for the duration of a call)
thinking_level	str	Reasoning-effort knob: "off" (default) / "low" / "medium" / "high" / "xhigh"; passed through to the provider
session_id	str	Stable id across successive exec() calls ("op-<hex>"); providers use it as the prompt-cache key
on_stream	Callable | None	Optional callback fn(event_dict) for streaming events (text / thinking / tool_use / tool_result)
last_usage	dict | None	Token usage of the last call: {input_tokens, output_tokens, total_tokens, cache_read, cache_create, ...}

---

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

Parameter	Type	Default	Description
content	list[dict] | str	(required)	List of content blocks (see format below). A plain string is wrapped into one text block
context	str | None	None	Legacy parameter, ignored — the provider path builds history from the DAG
response_format	dict | JsonSchemaOutput | None	None	A bare JSON Schema or normalized JsonSchemaOutput envelope. Verified provider/model combinations use their registered native mapping; otherwise fallback="auto" may use the verified hidden strict-tool path, while prompt fallback requires explicit fallback="prompt". Unsupported or lossy combinations fail closed. The terminal value is parsed and validated locally against the original schema and returned as a Python JSON value. max_validation_retries is 0 or 1. Also forwarded to _call() for subclasses
model	str | None	None	Override the default model
tools	list | None	None	The tools available to the LLM for this call. Entries may be @agentic_functions, {"spec":..., "execute":...} dicts, or objects with .spec / .execute. If set, the tool loop runs until the model returns plain text. Default (None) is not "no tools": the call gets the full registered toolset; pass toolset="none" for a reasoning-only call, or tools=[] for an explicit empty list
toolset / tools_source / tools_allow / tools_deny	—	None	Toolset preset and policy filtering: toolset names a preset ("full" is the implicit default, "none" opts out), tools_source filters per channel source, tools_allow / tools_deny are name allow/deny lists
tool_choice	str | dict	"auto"	"auto" / "required" / "none" / {"type":"function","name":"X"} to force a specific tool. Passed through to the provider (OpenAI / Anthropic / Gemini / Bedrock each map it to their own protocol form)
parallel_tool_calls	bool	True	Allow multiple tool calls in a single turn; False is passed through to providers that support the switch
max_iterations	int	20	Upper bound on tool-loop iterations (one iteration = one model call plus its tool execution). The effective value is max(1, min(50, max_iterations)), where 50 is the hard limit in agent_loop.py
choices	dict | list | None	None	If set, constrains the end of the turn: after the model finishes the full turn, its final reply must pick one of choices; exec parses and returns the result of that choice. See next-step-decision for details
timeout_s	float | None	None	The wall-clock time budget for the entire exec() (including all retry sleeps); on timeout, raises LLMError (reason=TIMEOUT, retryable=False). None = fall back to the OPENPROGRAM_EXEC_TIMEOUT_S environment variable (unset or 0 = unbounded)
on_retry	Callable | None	None	An observation callback invoked before each backoff sleep (once per failed attempt that has a retry queued), receiving a RetryInfo; not fired for the terminal failure. Exceptions raised inside the callback are swallowed
web_search	bool	False	Enable the provider's native web-search tool for this call, where supported
stream_fn	—	None	Per-call stream-function override (used by the dispatcher and tests to inject a fake or pre-built stream); None = the real provider

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

Parameter	Type	Description
content	list[dict]	The current turn's content blocks (history is rendered from the DAG by the provider path)
model	str	The model name
response_format	dict | None	Output format constraint

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

Situation	Handling
API call succeeds	Return the result
API raises a transient exception	Record the failed attempt, sleep with exponential backoff (base 1.5 s x 2^attempt, +/-25% jitter; a server Retry-After hint is honored as a lower bound; base tunable via OPENPROGRAM_RETRY_BACKOFF_BASE), then retry until max_retries is reached
Permanent error (bad image data, expired auth, invalid API key, or the provider marked the exception retryable=False)	Raised immediately as LLMError with retryable=False, no retry
The provider already exhausted its own transport-retry budget (transport_exhausted)	Not re-retried — raised as LLMError
TypeError or NotImplementedError	Raised immediately, no retry (usually a problem with the provider implementation or the way it's called)
All retries fail	Raise a structured LLMError (fields such as reason / retryable / http_status / attempts), with a full attempt report attached

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.