praval.model_runtime

Provider-neutral model runtime.

The runtime is the stable execution boundary between agents and providers. It keeps the legacy string API working while exposing neutral request/response objects for newer provider features.

Functions

`acall_with_retries`(operation, fn, *, ...[, ...])	Async `call_with_retries`: `fn` is awaited afresh for each attempt.
`call_with_retries`(operation, fn, *, retries, ...)	Call `fn` (one provider request), retrying it while it is retryable.
`execute_legacy_tool_call`(*, hitl_context, ...)	Execute a legacy provider tool call and return the result content.
`execute_legacy_tool_call_async`(*, ...[, ...])	Execute a tool on the caller's event loop with optional HITL gating.
`legacy_tool_to_spec`(tool, *[, strict])	Convert a legacy Praval tool dict to a neutral ToolSpec.
`max_provider_retries`(config)	Retries Praval allows for one provider request (`config.retries`).
`normalize_content_parts`(value)	Normalize public multimodal content input to ContentPart instances.
`normalize_reasoning_config`(value)	Normalize public reasoning config values.
`normalize_structured_output_config`(value)	Normalize public structured-output config values.
`record_provider_retry`(attempt, error, *[, ...])	Log a provider retry and record it as a fact and a span event.

Classes

ModelRuntime(*, provider, provider_name, config)

Runtime wrapper for provider-neutral model execution.

praval.model_runtime.max_provider_retries(config)[source]

Retries Praval allows for one provider request (config.retries).

Return type:: int
Parameters:: config (Any)

praval.model_runtime.record_provider_retry(attempt, error, *, operation='invoke', backoff_seconds=0.0)[source]

Log a provider retry and record it as a fact and a span event.

Only the error type is recorded, never its message.

Return type:

None

Parameters:

attempt (int)
error (BaseException)
operation (str)
backoff_seconds (float)

praval.model_runtime.call_with_retries(operation, fn, *, retries, map_error, on_retry=None)[source]

Call fn (one provider request), retrying it while it is retryable.

map_error turns any exception from fn into a typed ProviderError. A retryable error is retried up to retries times after _retry_backoff_seconds; anything else is raised at once, with the original exception as its cause when it was mapped. HITL signals pass through unchanged. fn must be safe to repeat: it should send one request and must not run tools.

Return type:

TypeVar(_T)

Parameters:

operation (str)
fn (Callable[[], _T])
retries (int)
map_error (Callable[[Exception], ProviderError])
on_retry (Callable[[...], None] | None)

async praval.model_runtime.acall_with_retries(operation, fn, *, retries, map_error, on_retry=None)[source]

Async call_with_retries: fn is awaited afresh for each attempt.

Return type:

TypeVar(_T)

Parameters:

operation (str)
fn (Callable[[], Awaitable[_T]])
retries (int)
map_error (Callable[[Exception], ProviderError])
on_retry (Callable[[...], None] | None)

praval.model_runtime.normalize_structured_output_config(value)[source]

Normalize public structured-output config values.

Return type:: Optional[StructuredOutputConfig]
Parameters:: value (Any)

praval.model_runtime.normalize_reasoning_config(value)[source]

Normalize public reasoning config values.

Return type:: Optional[ReasoningConfig]
Parameters:: value (Any)

praval.model_runtime.normalize_content_parts(value)[source]

Normalize public multimodal content input to ContentPart instances.

Return type:: Any
Parameters:: value (Any)

praval.model_runtime.legacy_tool_to_spec(tool, *, strict=False)[source]

Convert a legacy Praval tool dict to a neutral ToolSpec.

Return type:

Optional[ToolSpec]

Parameters:

tool (Dict[str, Any])
strict (bool)

praval.model_runtime.execute_legacy_tool_call(*, hitl_context, tool_call_id, function_name, raw_args, available_tools, continuation_state=None, resume_intervention=None)[source]

Execute a legacy provider tool call and return the result content.

Provider adapters place the returned text directly in their tool messages. The runtime tool loop uses the typed ToolResult instead.

Return type:

str

Parameters:

hitl_context (Dict[str, Any] | None)
tool_call_id (str)
function_name (str)
raw_args (Any)
available_tools (List[Dict[str, Any]])
continuation_state (Dict[str, Any] | None)
resume_intervention (Dict[str, Any] | None)

async praval.model_runtime.execute_legacy_tool_call_async(*, hitl_context, tool_call_id, function_name, raw_args, available_tools, continuation_state=None, resume_intervention=None)[source]

Execute a tool on the caller’s event loop with optional HITL gating.

Return type:

ToolResult

Parameters:

hitl_context (Dict[str, Any] | None)
tool_call_id (str)
function_name (str)
raw_args (Any)
available_tools (List[Dict[str, Any]])
continuation_state (Dict[str, Any] | None)
resume_intervention (Dict[str, Any] | None)

class praval.model_runtime.ModelRuntime(*, provider, provider_name, config, usage=None, agent_name=None)[source]

Bases: object

Runtime wrapper for provider-neutral model execution.

Parameters:

provider (Any)
provider_name (str)
config (Any)
usage (Optional[UsageMeter])
agent_name (Optional[str])

__init__(*, provider, provider_name, config, usage=None, agent_name=None)[source]

Parameters:

provider (Any)
provider_name (str)
config (Any)
usage (UsageMeter | None)
agent_name (str | None)

Return type:

None

property capabilities: ProviderCapabilities[source]: Return provider capabilities if exposed.

invoke(*, messages, tools=None, hitl_context=None, response_schema=None, reasoning=None, provider_options=None, timeout=None, metadata=None, stream_options=None, stream=False, max_tool_rounds=None)[source]

Execute a model request and return a neutral response.

Return type:

ModelResponse

Parameters:

messages (List[Dict[str, Any]])
tools (List[Dict[str, Any]] | None)
hitl_context (Dict[str, Any] | None)
response_schema (StructuredOutputConfig | None)
reasoning (str | Dict[str, Any] | ReasoningConfig | None)
provider_options (Dict[str, Any] | None)
timeout (float | None)
metadata (Dict[str, Any] | None)
stream_options (Dict[str, Any] | None)
stream (bool)
max_tool_rounds (int | None)

generate_text(*, messages, tools=None, hitl_context=None, **kwargs)[source]

Execute a request and return text for legacy callers.

Return type:

str

Parameters:

messages (List[Dict[str, Any]])
tools (List[Dict[str, Any]] | None)
hitl_context (Dict[str, Any] | None)
kwargs (Any)

async ainvoke(*, messages, tools=None, hitl_context=None, response_schema=None, reasoning=None, provider_options=None, timeout=None, metadata=None, stream_options=None, max_tool_rounds=None)[source]

Execute providers and tools without moving async tools across loops.

Return type:

ModelResponse

Parameters:

messages (List[Dict[str, Any]])
tools (List[Dict[str, Any]] | None)
hitl_context (Dict[str, Any] | None)
response_schema (StructuredOutputConfig | None)
reasoning (str | Dict[str, Any] | ReasoningConfig | None)
provider_options (Dict[str, Any] | None)
timeout (float | None)
metadata (Dict[str, Any] | None)
stream_options (Dict[str, Any] | None)
max_tool_rounds (int | None)

stream(*, messages, tools=None, hitl_context=None, response_schema=None, reasoning=None, provider_options=None, timeout=None, metadata=None, stream_options=None, max_tool_rounds=None)[source]

Stream normalized model events.

Return type:

Iterator[ModelEvent]

Parameters:

messages (List[Dict[str, Any]])
tools (List[Dict[str, Any]] | None)
hitl_context (Dict[str, Any] | None)
response_schema (StructuredOutputConfig | None)
reasoning (str | Dict[str, Any] | ReasoningConfig | None)
provider_options (Dict[str, Any] | None)
timeout (float | None)
metadata (Dict[str, Any] | None)
stream_options (Dict[str, Any] | None)
max_tool_rounds (int | None)

async astream(*, messages, tools=None, hitl_context=None, response_schema=None, reasoning=None, provider_options=None, timeout=None, metadata=None, stream_options=None, max_tool_rounds=None)[source]

Asynchronously stream normalized model events.

Return type:

AsyncIterator[ModelEvent]

Parameters:

messages (List[Dict[str, Any]])
tools (List[Dict[str, Any]] | None)
hitl_context (Dict[str, Any] | None)
response_schema (StructuredOutputConfig | None)
reasoning (str | Dict[str, Any] | ReasoningConfig | None)
provider_options (Dict[str, Any] | None)
timeout (float | None)
metadata (Dict[str, Any] | None)
stream_options (Dict[str, Any] | None)
max_tool_rounds (int | None)

resolve_capabilities(request)[source]

Resolve effective capabilities for a request.

Return type:: ProviderCapabilities
Parameters:: request (ModelRequest)

validate_request(request)[source]

Validate a model request before provider execution.

Return type:: None
Parameters:: request (ModelRequest)

resume_tool_flow(suspended_state, tools, hitl_context=None)[source]

Resume a runtime-owned tool loop after a HITL decision.

Return type:

ModelResponse

Parameters:

suspended_state (Dict[str, Any])
tools (List[Dict[str, Any]] | None)
hitl_context (Dict[str, Any] | None)

async resume_tool_flow_async(suspended_state, tools, hitl_context=None)[source]

Resume a runtime-owned tool loop while preserving the event loop.

Return type:

ModelResponse

Parameters:

suspended_state (Dict[str, Any])
tools (List[Dict[str, Any]] | None)
hitl_context (Dict[str, Any] | None)