Python API: Agents

The agent: a model, tools and instructions; runs, threads, streams and approvals. Generated from the docstrings and type hints of criogenai 0.1.7. See also: Python API overview, the same module in TypeScript.

Agents and threads: a model with tools, run by the Rust engine.

An Agent calls its model, runs the tools the model asks for, and repeats until the model answers. agent.run is one call that remembers nothing; agent.thread(id) is a conversation saved in the agent's storage, with approvals, retries, history and rewind.

from criogenai import (
    Agent,
    Thread,
    Stream,
    Reply,
    RunResult,
    Decision,
    ApprovalRequest,
    Retry,
    AgentTool,
    Event,
)

Contents#

Agent#

class Agent:
    def __init__(
        self,
        model: Any,
        *,
        instructions: Optional[str] = None,
        tools: Sequence[Any] = (),
        approve: Sequence[str] = (),
        output: Any = None,
        limits: Optional[Dict[str, Any]] = None,
        storage: Union[None, str, Storage] = None,
        memory: Optional[Memory] = None,
        on_tool_exception: Optional[str] = None,
        fallbacks: Sequence[Any] = (),
        fallback_on: Sequence[str] = (),
        model_retry: Optional[Retry] = None,
        tool_retry: Optional[Retry] = None,
        max_retry_after: Optional[float] = None,
        telemetry: Any = None,
        hooks: Sequence[Any] = (),
        redact: Any = None,
    )

A model with tools and instructions; the agent loop runs in Rust.

The agent calls the model, runs the tools it asks for (several at once when it asks for several), sends back the results and repeats until the model answers without calling a tool. Limits keep every run bounded. Use run for one-off calls and thread for conversations that are saved, can wait for approval and can be retried.

Parameters

• model (Any): The model: a model object (OpenAI("gpt-5-mini"), FakeModel([...]), ...) or a "provider:model" string such as "openai:gpt-5-mini". A string takes the provider's key from its usual environment variable.
• instructions (Optional[str], default None): The system prompt, sent ahead of the conversation on every model call.
• tools (Sequence[Any], default ()): What the model may call: functions decorated with @tool, other agents (agent.as_tool(...)), MCP servers or single MCP tools (criogenai.mcp), and collection search tools (Collection.as_tool).
• approve (Sequence[str], default ()): Names of tools that need a human decision before they run. The run stops with needs_approval set; continue it on a thread with approve, reject or decide.
• output (Any, default None): Structured output: a JSON Schema (dict) or a Pydantic model class. The run must end with JSON that matches it, which becomes result.output (an instance when you passed a Pydantic model).
• limits (Optional[Dict[str, Any]], default None): Bounds on each run, as a dict: model_calls (default 25), tool_calls (default 200), tokens (input plus output tokens across the run; no limit by default) and tool_timeout_secs (per tool call, default 60). Reaching a limit fails the run with LimitExceededError.
• storage (Union[None, str, Storage], default None): Where threads and memory are kept: a Storage, or a URL such as "sqlite://agent.db" or "postgres://...". None keeps them in this process's memory.
• memory (Optional[Memory], default None): Long-term memory (Memory.by_state(...), Memory.shared(...)): facts are recalled before each model call, and the model gets remember, forget and search_memory tools. Kept in storage.
• on_tool_exception (Optional[str], default None): What happens when a tool raises an exception other than ToolError (a ToolError always goes to the model). "return_to_model" (default): the model sees the exception's type and message, and the run goes on. "return_generic": the model only learns that the tool failed; the text stays in the tool result's artifact, which is never sent to a model. "fail": the run fails with ToolExecutionError, the exception as its __cause__.
• fallbacks (Sequence[Any], default ()): Models tried in order when the main model fails with a transient error (rate limit, overload, server error, timeout, network) after its retries. Same forms as model.
• fallback_on (Sequence[str], default ()): More model error kinds that also move to the next fallback, for example "context_overflow" (a model with a larger context window) or "content_filter" (another vendor).
• model_retry (Optional[Retry], default None): How transient model errors are retried. Default Retry(): 3 attempts, backoff from 0.5 s doubling up to 20 s, with jitter.
• tool_retry (Optional[Retry], default None): How tool calls are retried after a ToolError with retryable=True or a timeout. Only tools marked read_only or idempotent are retried. Default Retry().
• max_retry_after (Optional[float], default None): The longest Retry-After the agent waits for, in seconds (default 60). When a provider asks for longer, the call fails at once (or moves to a fallback) instead of hanging.
• telemetry (Any, default None): A Telemetry exporter; every run of this agent is sent to it.
• hooks (Sequence[Any], default ()): Objects whose methods run around the loop (on_start, before_model, after_model, before_tool, after_tool, on_end); see criogenai.hooks.
• redact (Any, default None): A function text -> text (or an async function) applied to every text in the transcript before each model call: the input, earlier replies and tool results. It runs before all hooks, so neither a hook nor a provider sees the original text, and the redacted transcript is what a thread saves. Use it to remove identifiers before anything leaves your systems.

Raises

• TypeError: An item of tools is not a tool, or an item of hooks has none of the hook methods.
• ValueError: model or a fallback is not a known "provider:model" string, or its API key is missing (that one is also a ValidationError with code invalid_value).
• ValidationError: The configuration is invalid, for example a limit of 0 or an unknown on_tool_exception.
• ModelError: A model cannot be built, for example Bedrock without an API key.

Example

from criogenai import Agent, FakeModel, FakeReply, tool

@tool
def add(a: int, b: int) -> int:
    "Adds two numbers."
    return a + b

model = FakeModel([FakeReply.tool_call("add", {"a": 2, "b": 3}), "2 + 3 = 5"])
agent = Agent(model, instructions="Use the tools for arithmetic.", tools=[add])
result = agent.run_sync("What is 2 + 3?")
print(result.text)  # 2 + 3 = 5

Agent.run#

async def run(input: str, **state: Any) -> RunResult

Runs the agent once on input, without a thread.

Nothing is kept between calls: each run starts from the instructions and input. A run that reaches a tool in approve ends with needs_approval set and cannot be continued; use a thread for approvals.

Parameters

• input (str): The user's message.
• **state (Any): Values for the run state, which hooks (ctx.state) and memory read. For example user_id="u-1" selects the namespace of Memory.by_state("user_id", ...).

Returns: The run's RunResult.

Raises

• ModelError: The model failed after retries and fallbacks.
• LimitExceededError: A run limit was reached.
• ToolExecutionError: A tool raised and on_tool_exception is "fail".
• CancelledError: The run was cancelled.

Example

result = await agent.run("Summarize the release notes.")
print(result.text, result.usage["output_tokens"])

Agent.run_sync#

def run_sync(input: str, **state: Any) -> RunResult

run for code without a running event loop, such as scripts.

Parameters

• input (str): The user's message.
• **state (Any): Values for the run state, as for run.

Returns: The run's RunResult.

Raises

• RuntimeError: Called inside a running event loop; await agent.run(...) there instead.

Agent.as_tool#

def as_tool(name: str, description: str) -> AgentTool

This agent as a tool of another agent.

The tool takes input (text), runs this agent on it and returns its answer. Agents calling agents may nest up to 8 levels deep.

Parameters

• name (str): The tool name the other agent's model sees.
• description (str): What this agent does, so the model knows when to call it.

Returns: An AgentTool to put in another agent's tools.

Example

researcher = Agent("openai:gpt-5-mini", instructions="Find facts.")
writer = Agent(
    "openai:gpt-5-mini",
    tools=[researcher.as_tool("research", "Finds facts on a topic.")],
)

Agent.thread#

def thread(thread_id: str, actor: Optional[str] = None) -> 'Thread'

The conversation thread_id in the agent's storage.

The thread is created by its first call. Thread objects with the same id share the stored conversation.

Parameters

• thread_id (str): Your id for the conversation, for example a chat session id.
• actor (Optional[str], default None): Who acts on the thread, recorded in the audit trail when the storage has audit=True. None records no actor.

Returns: A Thread.

Example

thread = agent.thread("session-42", actor="support-ui")
reply = await thread.send("Hello")
reply = await thread.send("What did I just say?")

Thread#

class Thread:
    def __init__(self, agent: Agent, thread_id: str, actor: Optional[str])

One conversation, kept in the agent's storage; made by Agent.thread.

Each call is saved as a checkpoint. While a call waits for approval, the thread takes only approve, reject or decide, and send raises ThreadSuspendedError. A failed or interrupted call is saved too, and retry continues it. A call on a thread that another run holds raises ThreadBusyError or waits, as the storage's on_busy says.

Parameters

• agent (Agent): The agent that runs the thread's calls.
• thread_id (str): The conversation's id.
• actor (Optional[str]): Who acts, for the audit trail.

Example

agent = Agent(model, tools=[delete_file], approve=["delete_file"])
thread = agent.thread("t-1")
reply = await thread.send("Delete a.txt")
if reply.needs_approval:
    reply = await thread.approve()
print(reply.text)

Thread.send#

async def send(input: str, **state: Any) -> Reply

Runs one turn: the stored conversation plus input.

Parameters

• input (str): The user's message.
• **state (Any): Values for the run state, as for Agent.run.

Returns: The Reply, saved as the thread's new head.

Raises

• ThreadSuspendedError: The thread waits for an approval decision.
• ThreadBusyError: Another run holds the thread.
• ModelError: The model failed after retries and fallbacks. The failed run is saved; retry continues it.

Thread.send_sync#

def send_sync(input: str, **state: Any) -> Reply

send for code without a running event loop.

Parameters

• input (str): The user's message.
• **state (Any): Values for the run state, as for Agent.run.

Returns: The Reply.

Raises

• RuntimeError: Called inside a running event loop; await thread.send(...) there instead.

Thread.approve#

async def approve() -> Reply

Approves every pending call and continues the run.

Returns: The Reply of the continued run.

Thread.reject#

async def reject(message: str = 'rejected') -> Reply

Rejects every pending call and continues the run.

The calls do not run; the model is told they were rejected.

Parameters

• message (str, default 'rejected'): The reason the model sees for each call.

Returns: The Reply of the continued run.

Thread.decide#

async def decide(decisions: Mapping[str, Dict[str, Any]]) -> Reply

Continues the run with one decision per pending call.

Parameters

• decisions (Mapping[str, Dict[str, Any]]): {call_id: decision}, with a Decision for every call in reply.approvals.

Returns: The Reply of the continued run.

Raises

• ValidationError: A call has no decision or a decision is invalid (resume_invalid); the approval stays pending.

Thread.retry#

async def retry() -> Reply

Continues a failed or interrupted run from where it stopped.

Tool calls that finished and were saved do not run again.

Returns: The Reply of the continued run.

Thread.history#

async def history(limit: int = 20) -> List[Dict[str, Any]]

The checkpoints of the thread's current branch, newest first.

Parameters

• limit (int, default 20): The most checkpoints to return.

Returns: One dict per checkpoint: id, parent_id, run_id, version, step, status (for example "completed" or "suspended"), source and created_at_ms.

Thread.messages#

async def messages() -> List[Dict[str, Any]]

The stored conversation: every message, as dicts.

Thread.pending_approvals#

async def pending_approvals() -> List[ApprovalRequest]

The tool calls waiting for a decision, read from storage.

Empty when the thread is not waiting. Use it to show a review queue without keeping the last Reply: for example after a restart, or in another process that shares the storage.

Returns: One ApprovalRequest per waiting call, in call order.

Example

waiting = await agent.thread("ticket-42").pending_approvals()
for request in waiting:
    print(request.tool, request.args)

Thread.rewind#

async def rewind(checkpoint_id: str) -> Dict[str, Any]

Makes an older checkpoint current again.

Writes a new head with that checkpoint's state, and its pending approval or failure if it had one. Newer checkpoints stay stored but leave the history. The next call runs from there again.

Parameters

• checkpoint_id (str): The id of a checkpoint from history.

Returns: The new head checkpoint, in the form of a history entry.

Thread.delete#

async def delete() -> bool

Deletes the thread and its checkpoints.

Returns: True if the thread existed.

Thread.stream#

def stream(input: str, **state: Any) -> Stream

send, streaming its events. Call it inside a running event loop.

Parameters

• input (str): The user's message.
• **state (Any): Values for the run state, as for Agent.run.

Returns: A Stream of Event objects; await stream.reply() gives the Reply.

Thread.stream_approve#

def stream_approve() -> Stream

approve, streaming the continued run's events.

Thread.stream_reject#

def stream_reject(message: str = 'rejected') -> Stream

reject, streaming the continued run's events.

Parameters

• message (str, default 'rejected'): The reason the model sees for each call.

Thread.stream_retry#

def stream_retry() -> Stream

retry, streaming the continued run's events.

Thread.stream_decide#

def stream_decide(decisions: Mapping[str, Dict[str, Any]]) -> Stream

decide, streaming the continued run's events.

Parameters

• decisions (Mapping[str, Dict[str, Any]]): {call_id: decision}, with a Decision for every call in reply.approvals.

Stream#

class Stream:
    def __init__(self, native: Any, output_model: Any)

A streamed thread call: its events as they happen, then its reply.

Made by Thread.stream and the other stream_* methods. Iterate with async for event in stream, then await stream.reply(). Leaving the loop early cancels the run; what it did so far is saved, and Thread.retry continues it.

Parameters

• native (Any): The engine's stream (internal).
• output_model (Any): The agent's Pydantic output model, if any (internal).

Example

stream = agent.thread("t-1").stream("Weather in Paris?")
async for event in stream:
    if event.type == "text":
        print(event.text, end="")
reply = await stream.reply()

Stream.reply#

async def reply() -> Reply

Waits for the call to end and returns its saved Reply.

Raises the errors Thread.send raises when the call failed.

Stream.cancel#

def cancel() -> None

Cancels the run. What it did so far stays saved.

Reply#

@dataclass
class Reply(RunResult):
    version: int = 0
    checkpoint_id: str = ''

A thread call's result, saved as the thread's new head.

Has every field of RunResult, plus the checkpoint it was saved as.

Attributes

• version (int, default 0): The version number of the thread's new head checkpoint.
• checkpoint_id (str, default ''): The id of that checkpoint: the id that Thread.history lists and Thread.rewind takes.

RunResult#

@dataclass
class RunResult:
    text: Optional[str]
    output: Any
    messages: List[Dict[str, Any]]
    usage: Dict[str, Any]
    stop_reason: Optional[str]
    needs_approval: bool
    approvals: List[ApprovalRequest]
    model_calls: int
    tool_calls: int
    run_id: str
    raw: Dict[str, Any] = field(repr=False, default_factory=dict)
    duration_ms: int = 0
    answered_by: Optional[Dict[str, str]] = None
    models: List[Dict[str, Any]] = field(default_factory=list)

What a run produced: the answer, the transcript and what it used.

Agent.run returns one; thread calls return a Reply, which adds the saved checkpoint.

Attributes

• text (Optional[str]): The final answer's text. None while the run waits for approval, and when structured output (output=) ended it.
• output (Any): The structured output when the agent has output=: a dict, or an instance of the Pydantic model you passed. When a hook ends the run, the output the hook gave. Otherwise None.
• messages (List[Dict[str, Any]]): The transcript as dicts (role, content blocks, id, metadata). For a thread, the whole stored conversation, earlier turns included.
• usage (Dict[str, Any]): Tokens used by the run's model calls: input_tokens, output_tokens, reasoning_tokens, cache_read_tokens, cache_write_tokens.
• stop_reason (Optional[str]): Why the run ended: "final_answer", "output" (structured output accepted), "return_direct", "content_filter", "hook_end" or "limit". None while the run waits for approval.
• needs_approval (bool): True when the run stopped before tool calls that need a decision (see approvals).
• approvals (List[ApprovalRequest]): The calls waiting for a decision; empty unless needs_approval.
• model_calls (int): Model calls in the run. Retries of one call count once.
• tool_calls (int): Tool calls executed in the run.
• run_id (str): The run's id, also used in telemetry and audit records.
• raw (Dict[str, Any], default field(repr=False, default_factory=dict)): The engine's result as a dict.
• duration_ms (int, default 0): Wall time of the call, in milliseconds.
• answered_by (Optional[Dict[str, str]], default None): {"provider", "model"} of the model that wrote the last answer; shows when a fallback answered. None if no model answered in this call.
• models (List[Dict[str, Any]], default field(default_factory=list)): Each model this call used, as {"provider", "model", "calls", "usage"}. estimate_cost reads it.

Decision#

class Decision

A reviewer's answer to one pending tool call, for Thread.decide.

Each method returns a plain dict. Thread.decide needs a decision for every pending call; a missing or invalid one fails with resume_invalid (ValidationError) and the approval stays pending.

Example

reply = await thread.send("Delete a.txt and b.txt")
decisions = {
    request.call_id: Decision.approve()
    if request.args.get("path") == "a.txt"
    else Decision.reject("Only a.txt may be deleted.")
    for request in reply.approvals
}
reply = await thread.decide(decisions)

Decision.approve#

@staticmethod
def approve() -> Dict[str, Any]

Runs the call as the model wrote it.

Decision.reject#

@staticmethod
def reject(message: Optional[str] = None) -> Dict[str, Any]

Does not run the call; the model is told it was rejected.

Parameters

• message (Optional[str], default None): Why, shown to the model. None sends a default text.

Decision.edit#

@staticmethod
def edit(args: Dict[str, Any], name: Optional[str] = None) -> Dict[str, Any]

Runs a changed call instead: other arguments, or another tool.

The new arguments are checked against the tool's schema. The result tells the model what actually ran.

Parameters

• args (Dict[str, Any]): The arguments to use instead of the model's.
• name (Optional[str], default None): Another tool to call instead. None keeps the tool the model chose.

Decision.respond#

@staticmethod
def respond(message: str) -> Dict[str, Any]

Does not run the call; message becomes the call's result.

Parameters

• message (str): The answer the model receives as the tool result. Must not be empty.

ApprovalRequest#

@dataclass(frozen=True)
class ApprovalRequest:
    call_id: str
    tool: str
    args: Dict[str, Any]
    description: Optional[str] = None
    allowed: Tuple[str, ...] = ()

A tool call waiting for a reviewer's decision.

A run stops before a tool listed in Agent(approve=[...]) runs. The result then has needs_approval set and one ApprovalRequest per waiting call in approvals. Answer with Thread.approve, Thread.reject or Thread.decide.

Attributes

• call_id (str): The tool call's id. Thread.decide takes decisions keyed by it.
• tool (str): The name of the tool the model wants to call.
• args (Dict[str, Any]): The arguments the model wrote.
• description (Optional[str], default None): Text for the reviewer, when the approval rule provides one; usually None.
• allowed (Tuple[str, ...], default ()): The decisions the reviewer may make: "approve", "edit", "reject", "respond".

Retry#

@dataclass(frozen=True)
class Retry:
    max_attempts: int = 3
    initial_backoff: float = 0.5
    factor: float = 2.0
    max_backoff: float = 20.0
    jitter: bool = True

How often and how long to retry: attempts and an exponential backoff.

Used by Agent(model_retry=..., tool_retry=...) and Gateway(retry=...). The wait grows from initial_backoff by factor per attempt, up to max_backoff. When a provider sends Retry-After, that wait is used instead (up to max_retry_after), and a request that never reached the provider is retried at once.

Attributes

• max_attempts (int, default 3): Attempts in total, including the first. 1 means no retry.
• initial_backoff (float, default 0.5): The first wait, in seconds.
• factor (float, default 2.0): How much each wait grows over the one before.
• max_backoff (float, default 20.0): The longest wait, in seconds.
• jitter (bool, default True): Randomize the waits, so that many clients do not retry at the same moment.

Example

agent = Agent(
    "openai:gpt-5-mini",
    model_retry=Retry(max_attempts=5, initial_backoff=1.0),
    tool_retry=Retry(max_attempts=2),
)

AgentTool#

class AgentTool:
    def __init__(self, agent: 'Agent', name: str, description: str)

An agent as a tool of another agent; made by Agent.as_tool.

The tool takes {"input": text} and returns the sub-agent's answer. Put it in the other agent's tools.

Parameters

• agent ('Agent'): The agent that runs when the tool is called.
• name (str): The tool name the model sees.
• description (str): What the tool does, for the model.

Event#

class Event:
    type: str
    data: Dict[str, Any]
    def __init__(self, data: Dict[str, Any])

One event of a streamed thread call.

The event's fields are also attributes (event.text). Reading a field this type of event does not have raises AttributeError. By type:

• text: text (a piece of the answer) and message_id.
• tool_call_started: call_id, tool and args.
• tool_call_finished: call_id, tool and status (for example "success" or "error").
• message: message, a whole message added to the transcript.
• approval_needed: requests, the calls waiting for a decision (as dicts).
• other: event (the engine event's name, e.g. "step_started"), node and step; never any content.

Parameters

• data (Dict[str, Any]): The event as a dict with a type key. Stream makes events; you do not build them.

Attributes

• type (str): The kind of event, one of the six above.
• data (Dict[str, Any]): The whole event as a dict.