API reference

The public API surface is generated from inline docstrings via mkdocstrings. Modules below are considered stable — see API stability for the deprecation policy.

Core runtime

Coordinate model decisions, tools, policies, events, and checkpoints.

One runtime instance serializes its own runs because a provider may maintain cursor state. Different runtime instances can operate independently.

Source code in src/avo/runtime.py

class AgentRuntime:
    """Coordinate model decisions, tools, policies, events, and checkpoints.

    One runtime instance serializes its own runs because a provider may
    maintain cursor state. Different runtime instances can operate
    independently.
    """

    def __init__(
        self,
        *,
        provider: ModelProvider,
        tools: Iterable[Tool] = (),
        policy: LoopPolicy | None = None,
        event_store: EventStore | None = None,
        clock: Clock = utc_now,
        approval_callback: ApprovalCallback | None = None,
        memory: LetheMemoryAdapter | None = None,
        hooks: HookRegistry | None = None,
        stream_callback: Callable[[str], None] | None = None,
        stream_interrupt_callback: Callable[[], None] | None = None,
    ) -> None:
        self.provider = provider
        self.tools = ToolRegistry(tools)
        self.policy = policy or LoopPolicy()
        self.event_store = event_store or InMemoryEventStore()
        self._clock = clock
        self._approval_callback = approval_callback or _always_approve
        # Purely observational display plumbing (see handle_model_pending);
        # public so callers can swap them per turn like ``provider``.
        # Each run/resume snapshots both at entry, so a swap only binds
        # from the next run — never to an in-flight one.
        # ``stream_interrupt_callback`` fires when a stream that already
        # forwarded text deltas dies mid-tokens, before the retry's deltas.
        self.stream_callback = stream_callback
        self.stream_interrupt_callback = stream_interrupt_callback
        self.memory = memory
        self.hooks = hooks if hooks is not None else HookRegistry()
        self._breaker: CircuitBreaker | None = (
            CircuitBreaker(self.policy.circuit_breaker)
            if self.policy.circuit_breaker is not None
            else None
        )
        self._execution_lock = asyncio.Lock()

    # ------------------------------------------------------------------
    # Public entry points
    # ------------------------------------------------------------------

    async def run(
        self,
        task: str,
        *,
        system_prompt: str | None = None,
        user_state: dict[str, JsonValue] | None = None,
        run_id: str | None = None,
        stream_callback: Callable[[str], None] | None = None,
        stream_interrupt_callback: Callable[[], None] | None = None,
    ) -> RunResult:
        """Create and execute a run until it reaches one explicit terminal state.

        ``system_prompt`` is sent as a native ``system`` message before the
        user task. ``stream_callback``/``stream_interrupt_callback`` passed here win
        over the instance attributes for this run only, so a chat turn can
        bind its printer without a concurrent background run — or a later
        turn — ever inheriting it. ``None`` means "inherit the instance".
        """

        async with self._execution_lock:
            now = self._now()
            values: dict[str, object] = {
                "task": task,
                "user_state": user_state or {},
                "created_at": now,
                "updated_at": now,
            }
            if run_id is not None:
                values["run_id"] = run_id
            record = RunRecord.model_validate(values)
            messages: list[dict[str, JsonValue]] = []
            if system_prompt and system_prompt.strip():
                messages.append({"role": "system", "content": system_prompt.strip()})
            if self.memory is not None:
                recalled = self.memory.recall_text(task)
                if recalled:
                    messages.append(
                        {
                            "role": "system",
                            "content": "Relevant memories:\n- " + "\n- ".join(recalled),
                        }
                    )
            messages.append({"role": "user", "content": task})
            context = _RunContext(
                run=record,
                policy=self.policy,
                messages=messages,
                next_step=1,
                token_usage=TokenUsage(),
                token_accounting_available=True,
                consecutive_errors=0,
                detector=ProgressDetector(),
                completed_tool_call_ids=set(),
                user_state=dict(user_state or {}),
                stream_callback=(
                    stream_callback if stream_callback is not None else self.stream_callback
                ),
                stream_interrupt_callback=(
                    stream_interrupt_callback
                    if stream_interrupt_callback is not None
                    else self.stream_interrupt_callback
                ),
            )
            created = AgentEvent(
                run_id=record.run_id,
                event_type=EventType.RUN_CREATED,
                created_at=now,
                payload={
                    "task": task,
                    "policy": cast(dict[str, JsonValue], self.policy.model_dump(mode="json")),
                },
            )
            persisted = await self.event_store.create_run(record, created)
            await self._event_persisted(persisted)

            try:
                await runtime_persistence.transition(self, context, RunState.MODEL_PENDING)
                await runtime_persistence.checkpoint(self, context)
                return await self._drive(context)
            except asyncio.CancelledError:
                if not is_terminal(context.run.state):
                    await runtime_persistence.terminate(
                        self,
                        context,
                        RunState.CANCELLED,
                        StopReason.USER_CANCELLED,
                        error="Run cancelled by the caller.",
                    )
                raise
            except StorageError:
                raise
            except Exception as exc:
                if not is_terminal(context.run.state):
                    await runtime_persistence.terminate(
                        self,
                        context,
                        RunState.FAILED,
                        StopReason.INTERNAL_ERROR,
                        error=f"{type(exc).__name__}: {exc}",
                    )
                return self._result(context.run)

    async def resume(
        self,
        run_id: str,
        *,
        stream_callback: Callable[[str], None] | None = None,
        stream_interrupt_callback: Callable[[], None] | None = None,
    ) -> RunResult:
        """Resume a persisted non-terminal run from its latest safe checkpoint.

        Per-call stream callbacks override the instance attributes for this
        resumed run exactly as in :meth:`run`.
        """

        async with self._execution_lock:
            record = await self.event_store.get_run(run_id)
            if is_terminal(record.state):
                raise RunAlreadyTerminalError(
                    f"Run {run_id!r} is already terminal ({record.state.value}) and cannot resume."
                )
            checkpoint_obj = await self.event_store.get_latest_checkpoint(run_id)
            if checkpoint_obj is None:
                raise CheckpointNotFoundError(
                    f"Run {run_id!r} has no checkpoint. Resume requires at least one "
                    "successfully persisted checkpoint."
                )

            policy = LoopPolicy.model_validate(checkpoint_obj.policy)
            context = _RunContext(
                run=record,
                policy=policy,
                messages=[dict(message) for message in checkpoint_obj.messages],
                next_step=checkpoint_obj.next_step,
                token_usage=checkpoint_obj.token_usage.model_copy(),
                token_accounting_available=checkpoint_obj.token_accounting_available,
                consecutive_errors=checkpoint_obj.consecutive_errors,
                detector=ProgressDetector(
                    action_history=checkpoint_obj.repeated_action_history,
                    observation_history=checkpoint_obj.observation_fingerprints,
                    model_history=checkpoint_obj.model_response_fingerprints,
                    progress_markers=checkpoint_obj.progress_markers,
                ),
                completed_tool_call_ids=set(checkpoint_obj.completed_tool_call_ids),
                user_state=dict(checkpoint_obj.user_state),
                pending_response=checkpoint_obj.pending_response,
                stream_callback=(
                    stream_callback if stream_callback is not None else self.stream_callback
                ),
                stream_interrupt_callback=(
                    stream_interrupt_callback
                    if stream_interrupt_callback is not None
                    else self.stream_interrupt_callback
                ),
            )
            runtime_persistence.restore_provider(self, checkpoint_obj.provider_metadata)
            await runtime_persistence.reconcile_after_checkpoint(self, context, checkpoint_obj)

            resumed = await runtime_persistence.append(
                self,
                context,
                EventType.RUN_RESUMED,
                {
                    "checkpoint_id": checkpoint_obj.checkpoint_id,
                    "checkpoint_sequence": checkpoint_obj.last_event_sequence,
                    "state": context.run.state.value,
                },
            )
            del resumed
            try:
                return await self._drive(context)
            except asyncio.CancelledError:
                if not is_terminal(context.run.state):
                    await runtime_persistence.terminate(
                        self,
                        context,
                        RunState.CANCELLED,
                        StopReason.USER_CANCELLED,
                        error="Resumed run cancelled by the caller.",
                    )
                raise
            except StorageError:
                raise
            except Exception as exc:
                if not is_terminal(context.run.state):
                    await runtime_persistence.terminate(
                        self,
                        context,
                        RunState.FAILED,
                        StopReason.INTERNAL_ERROR,
                        error=f"{type(exc).__name__}: {exc}",
                    )
                return self._result(context.run)

    async def inspect(self, run_id: str) -> RunTrace:
        """Build a chronological trace for a stored run."""

        from avo.tracing import TraceInspector

        return await TraceInspector(self.event_store).inspect(run_id)

    # ------------------------------------------------------------------
    # Internal driver and small helpers
    # ------------------------------------------------------------------

    async def _drive(self, context: _RunContext) -> RunResult:
        handlers = {
            RunState.CREATED: runtime_handlers.handle_created,
            RunState.MODEL_PENDING: runtime_handlers.handle_model_pending,
            RunState.DECISION_RECEIVED: runtime_handlers.handle_decision_received,
            RunState.TOOL_PENDING: runtime_handlers.handle_tool_pending,
            RunState.APPROVAL_PENDING: runtime_handlers.handle_approval_pending,
            RunState.TOOL_EXECUTING: runtime_handlers.handle_tool_executing,
            RunState.OBSERVATION_RECORDED: runtime_handlers.handle_observation_recorded,
            RunState.PAUSED: runtime_handlers.handle_paused,
        }
        provider_name = getattr(self.provider, "name", None) or "unknown"
        model_name = getattr(self.provider, "model", None)
        with span_for_turn(context.run.run_id, provider=provider_name, model=model_name) as span:
            while not is_terminal(context.run.state):
                handler = handlers.get(context.run.state)
                if handler is None:
                    raise RuntimeError(f"No state handler exists for {context.run.state.value!r}.")
                await handler(self, context)
            record_usage(
                span,
                input_tokens=context.token_usage.input_tokens,
                output_tokens=context.token_usage.output_tokens,
            )
        await self.hooks.fire(
            HookContext(event=HookEvent.STOP, run_id=context.run.run_id, run=context.run)
        )
        return self._result(context.run)

    def _active_record(
        self,
        context: _RunContext,
        *,
        state: RunState | None = None,
        steps: int | None = None,
        error: str | None = None,
    ) -> RunRecord:
        return RunRecord.model_validate(
            {
                **context.run.model_dump(),
                "state": state or context.run.state,
                "steps": context.run.steps if steps is None else steps,
                "token_usage": context.token_usage,
                "token_accounting_available": context.token_accounting_available,
                "user_state": context.user_state,
                "error": error,
                "updated_at": self._now(),
            }
        )

    def _operation_boundary_reason(self, context: _RunContext) -> StopReason | None:
        return context.policy.runtime_reason(self._elapsed(context))

    def _elapsed(self, context: _RunContext, *, now: datetime | None = None) -> float:
        current = now or self._now()
        return max(0.0, (current - context.run.created_at).total_seconds())

    def _now(self) -> datetime:
        value = self._clock()
        if value.tzinfo is None or value.utcoffset() is None:
            raise ValueError("Runtime clock must return a timezone-aware datetime.")
        return value

    @staticmethod
    def _assistant_message(response: ModelResponse) -> dict[str, JsonValue]:
        if response.tool_call is not None:
            return {
                "role": "assistant",
                "tool_call": cast(JsonValue, response.tool_call.model_dump(mode="json")),
            }
        return {"role": "assistant", "content": response.content}

    @staticmethod
    def _require_pending_response(context: _RunContext) -> ModelResponse:
        if context.pending_response is None:
            raise RuntimeError(
                f"State {context.run.state.value!r} requires a pending model response."
            )
        return context.pending_response

    @staticmethod
    def _require_tool_call(response: ModelResponse) -> ToolCall:
        if response.tool_call is None:
            raise RuntimeError("The current model decision does not contain a tool call.")
        return response.tool_call

    @staticmethod
    def _result(run: RunRecord) -> RunResult:
        if run.stop_reason is None:
            raise RuntimeError("Cannot build RunResult before a stop reason is persisted.")
        return RunResult(
            run_id=run.run_id,
            status=run.state,
            stop_reason=run.stop_reason,
            output=run.output,
            error=run.error,
            steps=run.steps,
            token_usage=run.token_usage,
            token_accounting_available=run.token_accounting_available,
        )

    async def _event_persisted(self, event: AgentEvent) -> None:
        """Hook called after each durable event; useful for failure-injection tests."""

        del event

    # ------------------------------------------------------------------
    # Persistence facade — delegates to runtime_persistence. Kept here
    # so existing callers (and runtime_handlers) can keep using
    # ``self._append``, ``self._checkpoint`` etc.
    # ------------------------------------------------------------------

    async def _append(
        self,
        context: _RunContext,
        event_type: EventType,
        payload: dict[str, JsonValue],
    ) -> AgentEvent:
        return await runtime_persistence.append(self, context, event_type, payload)

    async def _append_with_run(
        self,
        context: _RunContext,
        event_type: EventType,
        payload: dict[str, JsonValue],
    ) -> AgentEvent:
        return await runtime_persistence.append_with_run(self, context, event_type, payload)

    async def _transition(self, context: _RunContext, state: RunState) -> AgentEvent:
        return await runtime_persistence.transition(self, context, state)

    async def _checkpoint(self, context: _RunContext):  # type: ignore[no-untyped-def]
        return await runtime_persistence.checkpoint(self, context)

    def _record_tool_result(self, context: _RunContext, result: ToolResult) -> None:
        runtime_persistence.record_tool_result(self, context, result)

    def _provider_snapshot(self) -> dict[str, JsonValue]:
        return runtime_persistence.provider_snapshot(self)

    def _restore_provider(self, metadata: dict[str, JsonValue]) -> None:
        runtime_persistence.restore_provider(self, metadata)

    async def _reconcile_after_checkpoint(
        self,
        context: _RunContext,
        checkpoint_obj: Checkpoint,
    ) -> None:
        await runtime_persistence.reconcile_after_checkpoint(self, context, checkpoint_obj)

    async def _trigger_policy(
        self,
        context: _RunContext,
        reason: StopReason,
    ) -> None:
        await runtime_persistence.trigger_policy(self, context, reason)

    async def _terminate(
        self,
        context: _RunContext,
        state: RunState,
        reason: StopReason,
        *,
        output: str | None = None,
        error: str | None = None,
    ) -> None:
        await runtime_persistence.terminate(
            self, context, state, reason, output=output, error=error
        )

run async

run(
    task: str,
    *,
    system_prompt: str | None = None,
    user_state: dict[str, JsonValue] | None = None,
    run_id: str | None = None,
    stream_callback: Callable[[str], None] | None = None,
    stream_interrupt_callback: Callable[[], None]
    | None = None,
) -> RunResult

Create and execute a run until it reaches one explicit terminal state.

system_prompt is sent as a native system message before the user task. stream_callback/stream_interrupt_callback passed here win over the instance attributes for this run only, so a chat turn can bind its printer without a concurrent background run — or a later turn — ever inheriting it. None means "inherit the instance".

Source code in src/avo/runtime.py

async def run(
    self,
    task: str,
    *,
    system_prompt: str | None = None,
    user_state: dict[str, JsonValue] | None = None,
    run_id: str | None = None,
    stream_callback: Callable[[str], None] | None = None,
    stream_interrupt_callback: Callable[[], None] | None = None,
) -> RunResult:
    """Create and execute a run until it reaches one explicit terminal state.

    ``system_prompt`` is sent as a native ``system`` message before the
    user task. ``stream_callback``/``stream_interrupt_callback`` passed here win
    over the instance attributes for this run only, so a chat turn can
    bind its printer without a concurrent background run — or a later
    turn — ever inheriting it. ``None`` means "inherit the instance".
    """

    async with self._execution_lock:
        now = self._now()
        values: dict[str, object] = {
            "task": task,
            "user_state": user_state or {},
            "created_at": now,
            "updated_at": now,
        }
        if run_id is not None:
            values["run_id"] = run_id
        record = RunRecord.model_validate(values)
        messages: list[dict[str, JsonValue]] = []
        if system_prompt and system_prompt.strip():
            messages.append({"role": "system", "content": system_prompt.strip()})
        if self.memory is not None:
            recalled = self.memory.recall_text(task)
            if recalled:
                messages.append(
                    {
                        "role": "system",
                        "content": "Relevant memories:\n- " + "\n- ".join(recalled),
                    }
                )
        messages.append({"role": "user", "content": task})
        context = _RunContext(
            run=record,
            policy=self.policy,
            messages=messages,
            next_step=1,
            token_usage=TokenUsage(),
            token_accounting_available=True,
            consecutive_errors=0,
            detector=ProgressDetector(),
            completed_tool_call_ids=set(),
            user_state=dict(user_state or {}),
            stream_callback=(
                stream_callback if stream_callback is not None else self.stream_callback
            ),
            stream_interrupt_callback=(
                stream_interrupt_callback
                if stream_interrupt_callback is not None
                else self.stream_interrupt_callback
            ),
        )
        created = AgentEvent(
            run_id=record.run_id,
            event_type=EventType.RUN_CREATED,
            created_at=now,
            payload={
                "task": task,
                "policy": cast(dict[str, JsonValue], self.policy.model_dump(mode="json")),
            },
        )
        persisted = await self.event_store.create_run(record, created)
        await self._event_persisted(persisted)

        try:
            await runtime_persistence.transition(self, context, RunState.MODEL_PENDING)
            await runtime_persistence.checkpoint(self, context)
            return await self._drive(context)
        except asyncio.CancelledError:
            if not is_terminal(context.run.state):
                await runtime_persistence.terminate(
                    self,
                    context,
                    RunState.CANCELLED,
                    StopReason.USER_CANCELLED,
                    error="Run cancelled by the caller.",
                )
            raise
        except StorageError:
            raise
        except Exception as exc:
            if not is_terminal(context.run.state):
                await runtime_persistence.terminate(
                    self,
                    context,
                    RunState.FAILED,
                    StopReason.INTERNAL_ERROR,
                    error=f"{type(exc).__name__}: {exc}",
                )
            return self._result(context.run)

resume async

resume(
    run_id: str,
    *,
    stream_callback: Callable[[str], None] | None = None,
    stream_interrupt_callback: Callable[[], None]
    | None = None,
) -> RunResult

Resume a persisted non-terminal run from its latest safe checkpoint.

Per-call stream callbacks override the instance attributes for this resumed run exactly as in :meth:run.

Source code in src/avo/runtime.py

async def resume(
    self,
    run_id: str,
    *,
    stream_callback: Callable[[str], None] | None = None,
    stream_interrupt_callback: Callable[[], None] | None = None,
) -> RunResult:
    """Resume a persisted non-terminal run from its latest safe checkpoint.

    Per-call stream callbacks override the instance attributes for this
    resumed run exactly as in :meth:`run`.
    """

    async with self._execution_lock:
        record = await self.event_store.get_run(run_id)
        if is_terminal(record.state):
            raise RunAlreadyTerminalError(
                f"Run {run_id!r} is already terminal ({record.state.value}) and cannot resume."
            )
        checkpoint_obj = await self.event_store.get_latest_checkpoint(run_id)
        if checkpoint_obj is None:
            raise CheckpointNotFoundError(
                f"Run {run_id!r} has no checkpoint. Resume requires at least one "
                "successfully persisted checkpoint."
            )

        policy = LoopPolicy.model_validate(checkpoint_obj.policy)
        context = _RunContext(
            run=record,
            policy=policy,
            messages=[dict(message) for message in checkpoint_obj.messages],
            next_step=checkpoint_obj.next_step,
            token_usage=checkpoint_obj.token_usage.model_copy(),
            token_accounting_available=checkpoint_obj.token_accounting_available,
            consecutive_errors=checkpoint_obj.consecutive_errors,
            detector=ProgressDetector(
                action_history=checkpoint_obj.repeated_action_history,
                observation_history=checkpoint_obj.observation_fingerprints,
                model_history=checkpoint_obj.model_response_fingerprints,
                progress_markers=checkpoint_obj.progress_markers,
            ),
            completed_tool_call_ids=set(checkpoint_obj.completed_tool_call_ids),
            user_state=dict(checkpoint_obj.user_state),
            pending_response=checkpoint_obj.pending_response,
            stream_callback=(
                stream_callback if stream_callback is not None else self.stream_callback
            ),
            stream_interrupt_callback=(
                stream_interrupt_callback
                if stream_interrupt_callback is not None
                else self.stream_interrupt_callback
            ),
        )
        runtime_persistence.restore_provider(self, checkpoint_obj.provider_metadata)
        await runtime_persistence.reconcile_after_checkpoint(self, context, checkpoint_obj)

        resumed = await runtime_persistence.append(
            self,
            context,
            EventType.RUN_RESUMED,
            {
                "checkpoint_id": checkpoint_obj.checkpoint_id,
                "checkpoint_sequence": checkpoint_obj.last_event_sequence,
                "state": context.run.state.value,
            },
        )
        del resumed
        try:
            return await self._drive(context)
        except asyncio.CancelledError:
            if not is_terminal(context.run.state):
                await runtime_persistence.terminate(
                    self,
                    context,
                    RunState.CANCELLED,
                    StopReason.USER_CANCELLED,
                    error="Resumed run cancelled by the caller.",
                )
            raise
        except StorageError:
            raise
        except Exception as exc:
            if not is_terminal(context.run.state):
                await runtime_persistence.terminate(
                    self,
                    context,
                    RunState.FAILED,
                    StopReason.INTERNAL_ERROR,
                    error=f"{type(exc).__name__}: {exc}",
                )
            return self._result(context.run)

Models

Bases: AvoModel

Provider-neutral input for one model generation.

Source code in src/avo/models.py

class ModelRequest(AvoModel):
    """Provider-neutral input for one model generation."""

    request_id: str = Field(default_factory=new_id, min_length=1)
    run_id: str = Field(min_length=1)
    step: int = Field(ge=1)
    messages: list[dict[str, JsonValue]]
    tools: list[ToolMetadata] = Field(default_factory=list)
    cache: bool = False
    cache_prefix_messages: int | None = Field(default=None, ge=0)

Bases: AvoModel

Provider-neutral final answer or single tool-call decision.

Source code in src/avo/models.py

class ModelResponse(AvoModel):
    """Provider-neutral final answer or single tool-call decision."""

    response_id: str = Field(default_factory=new_id, min_length=1)
    content: str | None = None
    tool_call: ToolCall | None = None
    usage: TokenUsage | None = None

    @model_validator(mode="after")
    def validate_decision(self) -> ModelResponse:
        """Require exactly one of final content and a tool call."""

        has_content = self.content is not None
        has_call = self.tool_call is not None
        if has_content == has_call:
            raise ValueError("model response must contain exactly one of content or tool_call")
        return self

    @property
    def is_final(self) -> bool:
        """Return whether this response is a final answer."""

        return self.content is not None

is_final property

is_final: bool

Return whether this response is a final answer.

validate_decision

validate_decision() -> ModelResponse

Require exactly one of final content and a tool call.

Source code in src/avo/models.py

@model_validator(mode="after")
def validate_decision(self) -> ModelResponse:
    """Require exactly one of final content and a tool call."""

    has_content = self.content is not None
    has_call = self.tool_call is not None
    if has_content == has_call:
        raise ValueError("model response must contain exactly one of content or tool_call")
    return self

Loop policy

Bases: BaseModel

Runtime limits enforced at deterministic operation boundaries.

Source code in src/avo/policies.py

class LoopPolicy(BaseModel):
    """Runtime limits enforced at deterministic operation boundaries."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    max_steps: int = Field(default=20, gt=0)
    max_runtime_seconds: float | None = Field(default=300, gt=0)
    max_input_tokens: int | None = Field(default=None, gt=0)
    max_output_tokens: int | None = Field(default=None, gt=0)
    max_total_tokens: int | None = Field(default=None, gt=0)
    repeated_action_limit: int = Field(default=3, gt=0)
    consecutive_error_limit: int = Field(default=3, gt=0)
    no_progress_window: int = Field(default=5, gt=0)
    checkpoint_every_step: bool = True
    provider_timeout_seconds: float | None = Field(default=60, gt=0)
    tool_timeout_seconds: float | None = Field(default=60, gt=0)
    circuit_breaker: CircuitBreakerPolicy | None = Field(default=None)

    def token_budget_reason(
        self,
        usage: TokenUsage,
        *,
        accounting_available: bool,
    ) -> StopReason | None:
        """Return a token-budget stop reason when reported usage exceeds a limit.

        Missing provider usage does not invent a zero count. The runtime records
        accounting_available=False and cannot enforce a numeric token budget.
        """

        if not accounting_available:
            return None
        if self.max_input_tokens is not None and usage.input_tokens > self.max_input_tokens:
            return StopReason.TOKEN_BUDGET_EXCEEDED
        if self.max_output_tokens is not None and usage.output_tokens > self.max_output_tokens:
            return StopReason.TOKEN_BUDGET_EXCEEDED
        if self.max_total_tokens is not None and usage.total_tokens > self.max_total_tokens:
            return StopReason.TOKEN_BUDGET_EXCEEDED
        return None

    def runtime_reason(self, elapsed_seconds: float) -> StopReason | None:
        """Return a runtime-budget stop reason at an operation boundary."""

        if self.max_runtime_seconds is not None and elapsed_seconds >= self.max_runtime_seconds:
            return StopReason.MAX_RUNTIME
        return None

token_budget_reason

token_budget_reason(
    usage: TokenUsage, *, accounting_available: bool
) -> StopReason | None

Return a token-budget stop reason when reported usage exceeds a limit.

Missing provider usage does not invent a zero count. The runtime records accounting_available=False and cannot enforce a numeric token budget.

Source code in src/avo/policies.py

def token_budget_reason(
    self,
    usage: TokenUsage,
    *,
    accounting_available: bool,
) -> StopReason | None:
    """Return a token-budget stop reason when reported usage exceeds a limit.

    Missing provider usage does not invent a zero count. The runtime records
    accounting_available=False and cannot enforce a numeric token budget.
    """

    if not accounting_available:
        return None
    if self.max_input_tokens is not None and usage.input_tokens > self.max_input_tokens:
        return StopReason.TOKEN_BUDGET_EXCEEDED
    if self.max_output_tokens is not None and usage.output_tokens > self.max_output_tokens:
        return StopReason.TOKEN_BUDGET_EXCEEDED
    if self.max_total_tokens is not None and usage.total_tokens > self.max_total_tokens:
        return StopReason.TOKEN_BUDGET_EXCEEDED
    return None

runtime_reason

runtime_reason(elapsed_seconds: float) -> StopReason | None

Return a runtime-budget stop reason at an operation boundary.

Source code in src/avo/policies.py

def runtime_reason(self, elapsed_seconds: float) -> StopReason | None:
    """Return a runtime-budget stop reason at an operation boundary."""

    if self.max_runtime_seconds is not None and elapsed_seconds >= self.max_runtime_seconds:
        return StopReason.MAX_RUNTIME
    return None

Providers

Bases: Protocol

An async provider capable of producing one agent-loop decision.

Source code in src/avo/providers/base.py

@runtime_checkable
class ModelProvider(Protocol):
    """An async provider capable of producing one agent-loop decision."""

    async def generate(self, request: ModelRequest) -> ModelResponse:
        """Generate a final answer or one tool call."""

generate async

generate(request: ModelRequest) -> ModelResponse

Generate a final answer or one tool call.

Source code in src/avo/providers/base.py

async def generate(self, request: ModelRequest) -> ModelResponse:
    """Generate a final answer or one tool call."""

Tools

Bases: Generic[ArgumentsT]

Adapt a typed Pydantic input model and callable into a Tool.

Source code in src/avo/tools.py

class FunctionTool(Generic[ArgumentsT]):
    """Adapt a typed Pydantic input model and callable into a Tool."""

    def __init__(
        self,
        *,
        name: str,
        description: str,
        arguments_model: type[ArgumentsT],
        function: ToolCallable[ArgumentsT],
    ) -> None:
        self._arguments_model = arguments_model
        self._function = function
        self._metadata = ToolMetadata(
            name=name,
            description=description,
            input_schema=cast(dict[str, JsonValue], arguments_model.model_json_schema()),
        )

    @property
    def metadata(self) -> ToolMetadata:
        """Return the immutable name, description, and argument schema."""

        return self._metadata.model_copy(deep=True)

    async def invoke(self, arguments: dict[str, JsonValue]) -> JsonValue:
        """Validate arguments, invoke the function, and require JSON-safe output."""

        try:
            parsed = self._arguments_model.model_validate(arguments)
        except ValidationError as exc:
            raise ToolValidationError(
                f"Invalid arguments for tool {self._metadata.name!r}: {exc}"
            ) from exc

        try:
            value = self._function(parsed)
            if inspect.isawaitable(value):
                value = await value
        except Exception as exc:
            raise ToolExecutionError(
                f"Tool {self._metadata.name!r} raised {type(exc).__name__}: {exc}"
            ) from exc

        try:
            return _JSON_ADAPTER.validate_python(value)
        except ValidationError as exc:
            raise ToolValidationError(
                f"Tool {self._metadata.name!r} returned a non-JSON value: {exc}"
            ) from exc

metadata property

metadata: ToolMetadata

Return the immutable name, description, and argument schema.

invoke async

invoke(arguments: dict[str, JsonValue]) -> JsonValue

Validate arguments, invoke the function, and require JSON-safe output.

Source code in src/avo/tools.py

async def invoke(self, arguments: dict[str, JsonValue]) -> JsonValue:
    """Validate arguments, invoke the function, and require JSON-safe output."""

    try:
        parsed = self._arguments_model.model_validate(arguments)
    except ValidationError as exc:
        raise ToolValidationError(
            f"Invalid arguments for tool {self._metadata.name!r}: {exc}"
        ) from exc

    try:
        value = self._function(parsed)
        if inspect.isawaitable(value):
            value = await value
    except Exception as exc:
        raise ToolExecutionError(
            f"Tool {self._metadata.name!r} raised {type(exc).__name__}: {exc}"
        ) from exc

    try:
        return _JSON_ADAPTER.validate_python(value)
    except ValidationError as exc:
        raise ToolValidationError(
            f"Tool {self._metadata.name!r} returned a non-JSON value: {exc}"
        ) from exc

Name-indexed tool collection with idempotent invocation checks.

Source code in src/avo/tools.py

class ToolRegistry:
    """Name-indexed tool collection with idempotent invocation checks."""

    def __init__(self, tools: Iterable[Tool] = ()) -> None:
        self._tools: dict[str, Tool] = {}
        for item in tools:
            self.register(item)

    def register(self, item: Tool) -> None:
        """Register one tool and reject duplicate names."""

        name = item.metadata.name
        if name in self._tools:
            raise DuplicateToolError(
                f"Tool name {name!r} is already registered; tool names must be unique."
            )
        self._tools[name] = item

    @property
    def metadata(self) -> list[ToolMetadata]:
        """Return model-facing metadata in registration order."""

        return [item.metadata for item in self._tools.values()]

    def get(self, name: str) -> Tool:
        """Return a registered tool or raise a specific error."""

        try:
            return self._tools[name]
        except KeyError as exc:
            available = ", ".join(self._tools) or "(none)"
            raise ToolNotFoundError(
                f"Tool {name!r} is not registered. Available tools: {available}."
            ) from exc

    async def invoke(
        self,
        call: ToolCall,
        *,
        completed_tool_call_ids: set[str],
    ) -> ToolResult:
        """Invoke a call once and normalize success or expected failure details."""

        if call.tool_call_id in completed_tool_call_ids:
            raise ToolAlreadyCompletedError(
                f"Tool call {call.tool_call_id!r} was already completed and cannot run again."
            )

        started_at = utc_now()
        started_clock = time.perf_counter()
        try:
            tool = self.get(call.name)
            output = await tool.invoke(call.arguments)
        except (ToolNotFoundError, ToolValidationError, ToolExecutionError) as exc:
            finished_at = utc_now()
            return ToolResult(
                tool_call_id=call.tool_call_id,
                tool_name=call.name,
                success=False,
                error=str(exc),
                started_at=started_at,
                finished_at=finished_at,
                duration_ms=max(0.0, (time.perf_counter() - started_clock) * 1000),
            )
        finished_at = utc_now()
        return ToolResult(
            tool_call_id=call.tool_call_id,
            tool_name=call.name,
            success=True,
            output=output,
            started_at=started_at,
            finished_at=finished_at,
            duration_ms=max(0.0, (time.perf_counter() - started_clock) * 1000),
        )

metadata property

metadata: list[ToolMetadata]

Return model-facing metadata in registration order.

register

register(item: Tool) -> None

Register one tool and reject duplicate names.

Source code in src/avo/tools.py

def register(self, item: Tool) -> None:
    """Register one tool and reject duplicate names."""

    name = item.metadata.name
    if name in self._tools:
        raise DuplicateToolError(
            f"Tool name {name!r} is already registered; tool names must be unique."
        )
    self._tools[name] = item

get

get(name: str) -> Tool

Return a registered tool or raise a specific error.

Source code in src/avo/tools.py

def get(self, name: str) -> Tool:
    """Return a registered tool or raise a specific error."""

    try:
        return self._tools[name]
    except KeyError as exc:
        available = ", ".join(self._tools) or "(none)"
        raise ToolNotFoundError(
            f"Tool {name!r} is not registered. Available tools: {available}."
        ) from exc

invoke async

invoke(
    call: ToolCall, *, completed_tool_call_ids: set[str]
) -> ToolResult

Invoke a call once and normalize success or expected failure details.

Source code in src/avo/tools.py

async def invoke(
    self,
    call: ToolCall,
    *,
    completed_tool_call_ids: set[str],
) -> ToolResult:
    """Invoke a call once and normalize success or expected failure details."""

    if call.tool_call_id in completed_tool_call_ids:
        raise ToolAlreadyCompletedError(
            f"Tool call {call.tool_call_id!r} was already completed and cannot run again."
        )

    started_at = utc_now()
    started_clock = time.perf_counter()
    try:
        tool = self.get(call.name)
        output = await tool.invoke(call.arguments)
    except (ToolNotFoundError, ToolValidationError, ToolExecutionError) as exc:
        finished_at = utc_now()
        return ToolResult(
            tool_call_id=call.tool_call_id,
            tool_name=call.name,
            success=False,
            error=str(exc),
            started_at=started_at,
            finished_at=finished_at,
            duration_ms=max(0.0, (time.perf_counter() - started_clock) * 1000),
        )
    finished_at = utc_now()
    return ToolResult(
        tool_call_id=call.tool_call_id,
        tool_name=call.name,
        success=True,
        output=output,
        started_at=started_at,
        finished_at=finished_at,
        duration_ms=max(0.0, (time.perf_counter() - started_clock) * 1000),
    )

Circuit breaker

Three-state breaker — see module docstring.

Source code in src/avo/circuit_breaker.py

class CircuitBreaker:
    """Three-state breaker — see module docstring."""

    def __init__(
        self,
        policy: CircuitBreakerPolicy,
        *,
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        """Bind policy and an injectable monotonic clock for tests."""

        self.policy = policy
        self._state = CircuitState.CLOSED
        self._consecutive_failures = 0
        self._opened_at: float | None = None
        self._half_open_calls = 0
        self._clock = clock

    @property
    def state(self) -> CircuitState:
        """Return the current state, transitioning OPEN to HALF_OPEN if cooldown elapsed."""

        self._maybe_recover()
        return self._state

    @property
    def consecutive_failures(self) -> int:
        """Number of consecutive failures recorded in the current CLOSED window."""

        return self._consecutive_failures

    def allow(self) -> None:
        """Reserve a slot for one call. Raises :class:`BreakerOpen` when saturated."""

        self._maybe_recover()
        if self._state is CircuitState.OPEN:
            assert self._opened_at is not None
            raise BreakerOpen(
                self._state,
                self.policy.cooldown_seconds - (self._clock() - self._opened_at),
            )
        if self._state is CircuitState.HALF_OPEN:
            if self._half_open_calls >= self.policy.half_open_max_calls:
                raise BreakerOpen(self._state, retry_after_seconds=0.0)
            self._half_open_calls += 1

    def record_success(self) -> None:
        """Mark one call as successful; closes the breaker from HALF_OPEN."""

        self._consecutive_failures = 0
        self._opened_at = None
        self._half_open_calls = 0
        self._state = CircuitState.CLOSED

    def record_failure(self) -> None:
        """Mark one call as failed; opens the breaker after the threshold."""

        if self._state is CircuitState.HALF_OPEN:
            self._half_open_calls = max(0, self._half_open_calls - 1)
            self._open()
            return
        self._consecutive_failures += 1
        if self._consecutive_failures >= self.policy.failure_threshold:
            self._open()

    def reset(self) -> None:
        """Force the breaker back to CLOSED (test/admin use)."""

        self._state = CircuitState.CLOSED
        self._consecutive_failures = 0
        self._opened_at = None
        self._half_open_calls = 0

    def _open(self) -> None:
        self._state = CircuitState.OPEN
        self._opened_at = self._clock()
        self._half_open_calls = 0

    def _maybe_recover(self) -> None:
        if self._state is not CircuitState.OPEN:
            return
        assert self._opened_at is not None
        if self._clock() - self._opened_at >= self.policy.cooldown_seconds:
            self._state = CircuitState.HALF_OPEN
            self._half_open_calls = 0
            self._consecutive_failures = 0

state property

state: CircuitState

Return the current state, transitioning OPEN to HALF_OPEN if cooldown elapsed.

consecutive_failures property

consecutive_failures: int

Number of consecutive failures recorded in the current CLOSED window.

__init__

__init__(
    policy: CircuitBreakerPolicy,
    *,
    clock: Callable[[], float] = monotonic,
) -> None

Bind policy and an injectable monotonic clock for tests.

Source code in src/avo/circuit_breaker.py

def __init__(
    self,
    policy: CircuitBreakerPolicy,
    *,
    clock: Callable[[], float] = time.monotonic,
) -> None:
    """Bind policy and an injectable monotonic clock for tests."""

    self.policy = policy
    self._state = CircuitState.CLOSED
    self._consecutive_failures = 0
    self._opened_at: float | None = None
    self._half_open_calls = 0
    self._clock = clock

allow

allow() -> None

Reserve a slot for one call. Raises :class:BreakerOpen when saturated.

Source code in src/avo/circuit_breaker.py

def allow(self) -> None:
    """Reserve a slot for one call. Raises :class:`BreakerOpen` when saturated."""

    self._maybe_recover()
    if self._state is CircuitState.OPEN:
        assert self._opened_at is not None
        raise BreakerOpen(
            self._state,
            self.policy.cooldown_seconds - (self._clock() - self._opened_at),
        )
    if self._state is CircuitState.HALF_OPEN:
        if self._half_open_calls >= self.policy.half_open_max_calls:
            raise BreakerOpen(self._state, retry_after_seconds=0.0)
        self._half_open_calls += 1

record_success

record_success() -> None

Mark one call as successful; closes the breaker from HALF_OPEN.

Source code in src/avo/circuit_breaker.py

def record_success(self) -> None:
    """Mark one call as successful; closes the breaker from HALF_OPEN."""

    self._consecutive_failures = 0
    self._opened_at = None
    self._half_open_calls = 0
    self._state = CircuitState.CLOSED

record_failure

record_failure() -> None

Mark one call as failed; opens the breaker after the threshold.

Source code in src/avo/circuit_breaker.py

def record_failure(self) -> None:
    """Mark one call as failed; opens the breaker after the threshold."""

    if self._state is CircuitState.HALF_OPEN:
        self._half_open_calls = max(0, self._half_open_calls - 1)
        self._open()
        return
    self._consecutive_failures += 1
    if self._consecutive_failures >= self.policy.failure_threshold:
        self._open()

reset

reset() -> None

Force the breaker back to CLOSED (test/admin use).

Source code in src/avo/circuit_breaker.py

def reset(self) -> None:
    """Force the breaker back to CLOSED (test/admin use)."""

    self._state = CircuitState.CLOSED
    self._consecutive_failures = 0
    self._opened_at = None
    self._half_open_calls = 0

Bases: BaseModel

Tuning knobs for :class:CircuitBreaker.

Source code in src/avo/circuit_breaker.py

class CircuitBreakerPolicy(BaseModel):
    """Tuning knobs for :class:`CircuitBreaker`."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    failure_threshold: int = Field(default=5, ge=1)
    cooldown_seconds: float = Field(default=30.0, gt=0)
    half_open_max_calls: int = Field(default=1, ge=1)

Logging

Configure the named logger (or root if None).

Returns the JSON handler when json_mode is set, otherwise None. The handler is replaced on each call so callers can swap modes at runtime (e.g. avo chat switches to JSON when AVO_LOG_FORMAT=json is set).

Source code in src/avo/logging_config.py

def configure_logging(
    *,
    level: LogLevel = "INFO",
    json_mode: bool = False,
    logger_name: str | None = None,
) -> logging.Handler | None:
    """Configure the named logger (or root if ``None``).

    Returns the JSON handler when ``json_mode`` is set, otherwise
    ``None``. The handler is replaced on each call so callers can
    swap modes at runtime (e.g. ``avo chat`` switches to JSON when
    ``AVO_LOG_FORMAT=json`` is set).
    """

    logger = logging.getLogger(logger_name)
    for existing in list(logger.handlers):
        if isinstance(existing.formatter, JsonFormatter):
            logger.removeHandler(existing)
    if json_mode:
        return install_json_handler(logger, level=level)

    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
    handler.setLevel(level)
    logger.addHandler(handler)
    logger.setLevel(level)
    return None