enterprise_ai.loop

Source: sdks/python/src/enterprise_ai/loop/__init__.py. Back to the Python SDK reference index.

note

This page is generated by tools/docs/gen_sdk_reference.py from the SDK sources and the OpenAPI text folders. Do not edit it by hand: change the source and regenerate.

Model LoopBudget

Local budgets; None means unlimited.

Fields:

Name	Type	Required	Description
max_steps	int	no	Maximum number of model calls (plan steps). Default 20.
max_tool_calls	int | None	no	Maximum number of tool calls over the run; None is unlimited.
max_output_tokens	int | None	no	Maximum number of output tokens over the run; None is unlimited.
max_cost	Decimal | None	no	Maximum measured cost of the run, as a Decimal; None is unlimited.
max_recoveries	int	no	Maximum number of recoveries after a failed tool call or a rejected answer before the run fails. Default 3.
max_duration	float | None	no	Maximum duration of the run, in seconds; None is unlimited.
max_seconds	float | None	no	Alias of max_duration, kept for compatibility.

LoopBudget.from_manifest

from_manifest(manifest: dict[str, Any]) -> LoopBudget

Budgets from spec.autonomy, spec.budgets and spec.runtime.limits.

Name	Type	Default	Required	Description
manifest	dict[str, Any]		yes	Agent manifest as a dict, read under its spec key.

Returns: LoopBudget. A LoopBudget: Local budgets; None means unlimited.

Model LoopState

Serializable state of a run (stored in checkpoints).

Fields:

Name	Type	Required	Description
messages	list[dict[str, Any]]	no	Chat messages of the run, tool results included.
step	int	no	Current step, free text (up to 500 characters).
tool_calls	int	no	Number of tool calls executed so far.
input_tokens	int	no	Input tokens counted over the run.
output_tokens	int	no	Output tokens counted over the run.
cost	str	no	Measured cost so far, as a decimal string. Default "0".
pending	list[dict[str, Any]]	no	Tool calls asked by the model and not executed yet.
final	Any	no	True when this is the last step of the run. Default false.
status	str	no	Status of the run: running (default), completed, failed, budget_exhausted or stopped.
recoveries	int	no	Number of recoveries so far (failed tool calls or rejected answers).
last_error	str | None	no	Last failure or the name of the spent budget, or None.
phase	str	no	Next phase to run: plan (default), act, verify or done.
started_at	float	no	Monotonic clock value at the start, used for max_duration; not saved in checkpoints.

LoopState.from_dict

from_dict(data: dict[str, Any]) -> LoopState

Rebuild a state from a checkpoint dict; unknown keys and started_at are ignored.

Name	Type	Default	Required	Description
data	dict[str, Any]		yes	Dict produced by to_dict, for example the state of a checkpoint.

Returns: LoopState. A LoopState: Serializable state of a run (stored in checkpoints).

LoopState.to_dict

to_dict() -> dict[str, Any]

Serializable form of the state for a checkpoint, without started_at.

This element takes no parameter.

Returns: dict[str, Any]. Dict of every field except started_at.

Model LoopResult

Outcome of :meth:AgentLoop.run.

Fields:

Name	Type	Required	Description
status	str	yes	Final status of the run: completed, failed, budget_exhausted or stopped.
output	Any	yes	Output of the operation (transcript, image references...).
state	LoopState	yes	Serializable state of a run (stored in checkpoints).
reason	str | None	no	Why the action is taken. Written to the audit trail with the actor and the time.

Attributes:

Name	Type	Required	Description
ok	bool	yes	True when status is completed.

Class AgentLoop

Run an agent with budgets and governance hooks.

Attributes:

Name	Type	Required	Description
agent_id		yes	Identifier of the agent watched by the kill switch, or None.
approval_for		yes	Glob patterns of tools that need a human approval.
approval_timeout		yes	Maximum wait for an approval decision, in seconds.
budget		yes	Local budgets (LoopBudget) applied by route.
bus		yes	Message bus read by serve, or None.
checkpoint_every		yes	Number of steps between two checkpoints; 0 saves only the final one.
client		yes	Enterprise AI client used for checkpoints, approvals, kill switch, sessions and usage, or None to run offline.
context_budget		yes	ContextBudget used by prepare_context, or None.
final_verifier		yes	Function checking the final answer, or None to accept every answer.
history_limit		yes	Number of last session messages loaded as history; 0 loads every one.
kill_poll_every		yes	Number of steps between two polls of the agent status; 0 disables the poll.
last_runner	Any	yes	Last AgentRunner built by runner, or None.
model		yes	Model identifier, or None.
model_call		yes	Function (messages, tools) -> completion called in the plan phase.
on_event		yes	Callback (kind, data) called on each loop event, or None.
poll_interval		yes	Pause between two intake cycles of serve, in seconds.
processed_runs	set[str]	yes	Identifiers of the pending runs already taken by serve, so none is run twice.
record_steps		yes	When True, each phase is recorded as a run step on the server.
remote_tool		yes	Function (name, arguments) running the tools that are not local, or None.
remote_tool_specs		yes	OpenAI tools entries of the remote tools.
run_id		yes	Identifier of the server run, or None.
runner_options		yes	Extra keyword arguments passed to every AgentRunner.
session_cursors	dict[str, str | None]	yes	Watched sessions with the identifier of the last message processed (None before the first).
session_id		yes	Default session of the loop, or None.
stop_event		yes	Event that stops the loop when set (heartbeat or killed agent).
tools	dict[str, LocalTool]	yes	Local tools by name (a dict of LocalTool).
verifier		yes	Function checking each tool result, or None to accept every result.

Constructor AgentLoop

AgentLoop(model_call: ModelCall | None = None, model: str | None = None, session_id: str | None = None, bus: Any = None, context_budget: Any = None, history_limit: int = 50, poll_interval: float = 2.0, watch_sessions: list[str] | None = None, runner_options: dict[str, Any] | None = None, tools: dict[str, LocalTool] | list[LocalTool] | None = None, remote_tool: Callable[[str, dict[str, Any]], Any] | None = None, remote_tool_specs: list[dict[str, Any]] | None = None, budget: LoopBudget | None = None, verifier: Verifier | None = None, final_verifier: Callable[[Any], bool | str] | None = None, approval_for: list[str] | None = None, client: Any = None, agent_id: str | None = None, run_id: str | None = None, checkpoint_every: int = 1, kill_poll_every: int = 1, record_steps: bool = True, stop_event: threading.Event | None = None, on_event: Callable[[str, dict[str, Any]], None] | None = None, approval_timeout: float = 3600.0) -> None

Configure the loop: model, local and remote tools, budgets, verifiers, approvals and the client used for checkpoints, kill switch and run steps.

Name	Type	Default	Required	Description
model_call	ModelCall | None	None	no	Function (messages, tools) -> completion returning an OpenAI style completion; an optional eai_cost key is added to the cost.
model	str | None	None	no	Model identifier used by the runner and the context builder.
session_id	str | None	None	no	Default session whose history is loaded and to which answers are posted.
bus	Any	None	no	Message bus read by serve: inbound messages are answered and the answer is published outbound; None disables it.
context_budget	Any	None	no	ContextBudget used to fit the system prompt, history and input in prepare_context; None keeps every message.
history_limit	int	50	no	Number of last session messages loaded as history; 0 loads every one. Default 50.
poll_interval	float	2.0	no	Pause between two intake cycles of serve, in seconds. Default 2.0.
watch_sessions	list[str] | None	None	no	Sessions whose new messages are processed by serve, from the start.
runner_options	dict[str, Any] | None	None	no	Extra keyword arguments passed to every AgentRunner built by runner.
tools	dict[str, LocalTool] | list[LocalTool] | None	None	no	Local tools, as a list of LocalTool or a dict by name.
remote_tool	Callable[[str, dict[str, Any]], Any] | None	None	no	Function (name, arguments) that runs a tool which is not local, for example through the Tool Gateway.
remote_tool_specs	list[dict[str, Any]] | None	None	no	OpenAI tools entries of the remote tools, offered to the model next to the local ones.
budget	LoopBudget | None	None	no	Local budgets (LoopBudget); by default 20 steps and 3 recoveries, nothing else limited.
verifier	Verifier | None	None	no	Function (name, result) checking each tool result: True accepts it, any other value is a failure reason and triggers a recovery.
final_verifier	Callable[[Any], bool | str] | None	None	no	Function checking the final answer: True accepts it, any other value is the rejection reason sent back to the model.
approval_for	list[str] | None	None	no	Glob patterns of tools that need a human approval, matched against name and tool:<name>; empty by default.
client	Any	None	no	Parent client.
agent_id	str | None	None	no	Identifier of the agent whose status is polled by the kill switch and reported in usage events.
run_id	str | None	None	no	Identifier of the server run used for checkpoints, run steps and usage events.
checkpoint_every	int	1	no	Save a checkpoint after the act phase every N steps; 0 saves only the final one. Default 1.
kill_poll_every	int	1	no	Poll the agent status for the kill switch every N steps; 0 disables the poll. Default 1.
record_steps	bool	True	no	When True (default) and a run id is set, each phase is recorded as a run step on the server.
stop_event	threading.Event | None	None	no	Event that stops the loop when set (heartbeat); a new threading.Event by default.
on_event	Callable[[str, dict[str, Any]], None] | None	None	no	Callback (kind, data) called on each loop event (plan, tool, approval_requested, done).
approval_timeout	float	3600.0	no	Maximum wait for an approval decision, in seconds. Default 3600.0.

AgentLoop.act

act(state: LoopState) -> LoopState

Execute pending tool calls; failures and rejections become tool messages.

Name	Type	Default	Required	Description
state	LoopState		yes	Current LoopState, whose pending tool calls are executed; it is updated in place.

Returns: LoopState. A LoopState: Serializable state of a run (stored in checkpoints).

AgentLoop.check_budget

check_budget(state: LoopState) -> str | None

Name of the exhausted budget, or None.

Name	Type	Default	Required	Description
state	LoopState		yes	Current LoopState, compared with the limits of budget.

Returns: str \| None. max_steps, max_tool_calls, max_output_tokens, max_cost or max_duration, or None when no budget is spent.

AgentLoop.check_kill_switch

check_kill_switch(state: LoopState) -> None

Raise :class:AgentStopped when the heartbeat or the agent status asks to stop.

Name	Type	Default	Required	Description
state	LoopState		yes	Current LoopState; its step number decides whether the agent status is polled this time.

Returns: None.

Raises: AgentStopped.

AgentLoop.checkpoint

checkpoint(state: LoopState) -> Any

POST the state to the server checkpoints API (when a run id is known).

Name	Type	Default	Required	Description
state	LoopState		yes	State to save, sent with the label step-<n>.

Returns: Any. Answer of the checkpoints API, or None without a client or a run id.

AgentLoop.history

history(session_id: str) -> list[dict[str, Any]]

List the messages of a session. Messages in order, paginated with cursor. after (message id or timestamp) lets a runtime poll the new messages only; an unknown message id or a malformed timestamp answers 400 invalid_request.

Route: /v1/sessions/{session_id}/messages (GET).

Name	Type	Default	Required	Description
session_id	str		yes	Identifier (UUID) of the session.

Returns: list[dict[str, Any]]. Decoded JSON answer of the operation "List the messages of a session".

Raises: the typed errors of the transport (Errors).

AgentLoop.needs_approval

needs_approval(name: str) -> bool

Tell whether a tool call needs a human approval: the local tool requires one or its name matches approval_for.

Name	Type	Default	Required	Description
name	str		yes	Name of the tool, matched as name and tool:<name>.

Returns: bool. True when the call must be approved first.

AgentLoop.plan

plan(state: LoopState) -> LoopState

Call the model; store tool calls in pending or the answer in final.

Name	Type	Default	Required	Description
state	LoopState		yes	Current LoopState; its messages are sent to the model and it is updated in place.

Returns: LoopState. A LoopState: Serializable state of a run (stored in checkpoints).

AgentLoop.poll_once

poll_once() -> list[TurnResult]

List the runs. Runs of the tenant, newest first, paginated with cursor. Filters: agent_id (id or name; agent is an alias), state, session_id, parent_run_id. An unknown state or a malformed UUID answers 400 invalid_request.

Route: /v1/runs (GET).

This element takes no parameter.

Returns: list[TurnResult]. A TurnResult: Outcome of :meth:AgentRunner.run_turn.

Raises: the typed errors of the transport (Errors).

AgentLoop.prepare_context

prepare_context(input: Any, session_id: str | None = None, system: str | None = None) -> list[dict[str, Any]]

System prompt + session history + the new input, fitted in context_budget when set.

Name	Type	Default	Required	Description
input	Any		yes	New user input; a non text value is serialized as JSON, None adds no user message.
session_id	str | None	None	no	Session whose history is loaded; None uses no history.
system	str | None	None	no	System prompt placed first, or None.

Returns: list[dict[str, Any]]. List of chat messages (role, content) ready for the model.

AgentLoop.request_approval

request_approval(name: str, arguments: dict[str, Any]) -> bool

Create an approval and wait for the decision. True when approved.

Name	Type	Default	Required	Description
name	str		yes	Name of the tool, sent as the resource of the approval (tools:invoke action).
arguments	dict[str, Any]		yes	Arguments of the tool call, sent as the reason of the approval (JSON, at most 500 characters).

Returns: bool. True when the decision is approved; raises ApprovalRequired without a client.

Raises: ApprovalRequired.

AgentLoop.resume

resume(run_id: str) -> TurnResult

List the checkpoints of a run. Checkpoints of the run in sequence order. One page only. 404 run_not_found.

Route: /v1/runs/{run_id}/checkpoints (GET).

Name	Type	Default	Required	Description
run_id	str		yes	Identifier (UUID) of the run.

Returns: TurnResult. A TurnResult: Outcome of :meth:AgentRunner.run_turn.

Raises: ValueError, the typed errors of the transport (Errors).

AgentLoop.route

route(state: LoopState) -> str

Next phase: plan, act, verify or done (applies budgets and recovery limits).

Name	Type	Default	Required	Description
state	LoopState		yes	Current LoopState; its status and last_error are set when the loop ends.

Returns: str. Name of the next phase: plan, act, verify or done.

AgentLoop.run_legacy

run_legacy(task: Any = None, system: str | None = None, messages: list[dict[str, Any]] | None = None, state: LoopState | None = None, raise_on_budget: bool = False) -> LoopResult

Drive the loop to completion. Resumes from state when given (checkpoint restore).

Name	Type	Default	Required	Description
task	Any	None	no	Task given to the agent as a user message; a non text value is serialized as JSON.
system	str | None	None	no	System prompt, added first when the messages hold none.
messages	list[dict[str, Any]] | None	None	no	Previous chat messages to start from.
state	LoopState | None	None	no	State restored from a checkpoint; when given, task, system and messages are ignored.
raise_on_budget	bool	False	no	When True, a spent budget raises BudgetExhausted instead of returning a result. Default False.

Returns: LoopResult. A LoopResult: Outcome of :meth:AgentLoop.run.

Raises: BudgetExhausted.

AgentLoop.run_once

run_once(input: Any = None, session_id: str | None = None, run_id: str | None = None, system: str | None = None, state: dict[str, Any] | None = None) -> TurnResult

Run one Navin turn (:class:AgentRunner) and post the answer to the session.

Name	Type	Default	Required	Description
input	Any	None	no	New user input of the turn; ignored when state is given.
session_id	str | None	None	no	Session of the turn, used for history and to post the answer; defaults to the session_id of the loop.
run_id	str | None	None	no	Server run of the turn; defaults to the run_id of the loop.
system	str | None	None	no	System prompt of the turn, or None.
state	dict[str, Any] | None	None	no	Runner state restored from a checkpoint to resume the turn.

Returns: TurnResult. A TurnResult: Outcome of :meth:AgentRunner.run_turn.

AgentLoop.runner

runner(run_id: str | None = None, session_id: str | None = None) -> AgentRunner

An :class:AgentRunner sharing this loop's model, tools, budget and stop event.

Name	Type	Default	Required	Description
run_id	str | None	None	no	Server run of the runner; defaults to the run_id of the loop.
session_id	str | None	None	no	Session of the runner; defaults to the session_id of the loop.

Returns: AgentRunner. A AgentRunner: Run agent turns against the platform gateway with tools, rules and governance.

AgentLoop.serve

async serve(max_cycles: int | None = None) -> None

Serve until :meth:shutdown: bus inbound messages, pending runs and new session messages.

Name	Type	Default	Required	Description
max_cycles	int | None	None	no	Stop after this number of intake cycles; None serves until shutdown.

Returns: None.

AgentLoop.shutdown

shutdown() -> None

Ask :meth:serve to exit after the current turn.

This element takes no parameter.

Returns: None.

AgentLoop.start_state

start_state(task: Any, system: str | None = None, messages: list[dict[str, Any]] | None = None) -> LoopState

Build the initial LoopState: previous messages, the system prompt when none is present, then the task as a user message.

Name	Type	Default	Required	Description
task	Any		yes	Task appended as a user message; a non text value is serialized as JSON, None adds nothing.
system	str | None	None	no	System prompt, inserted first when the messages hold none.
messages	list[dict[str, Any]] | None	None	no	Previous chat messages, copied.

Returns: LoopState. A LoopState: Serializable state of a run (stored in checkpoints).

AgentLoop.tool_specs

tool_specs() -> list[dict[str, Any]]

OpenAI tools list: local tools plus remote Tool Gateway tools.

This element takes no parameter.

Returns: list[dict[str, Any]]. List of OpenAI tools entries, local tools first.

AgentLoop.verify

verify(state: LoopState) -> LoopState

Check the final answer; a rejected answer is sent back to the model (recover).

Name	Type	Default	Required	Description
state	LoopState		yes	Current LoopState, whose final answer is checked; it is updated in place.

Returns: LoopState. A LoopState: Serializable state of a run (stored in checkpoints).

AgentLoop.watch

watch(session_id: str, after: str | None = None) -> None

Process new messages of session_id in :meth:serve.

Name	Type	Default	Required	Description
session_id	str		yes	Session to watch.
after	str | None	None	no	Identifier of the last message already seen; None processes the session from the start.

Returns: None.