State¶

Learn: Keep the conversation. Concepts: State and history. Guide: Go back or try another path.

parent_id, parent_call_id, child and achild are about subagents: States that a tool call of this State started. Guide: Use subagents.

State ¶

State(*args: Never, id: str | None = None, messages: Iterable[Message] = (), extra_data: Mapping[str, Any] | None = None, history: Iterable[HistoryEntry] | None = None, **kwargs: Never)

One conversation's record: the messages the model sees (messages) and everything that happened (history).

Create it with the messages to start from, pass it to agent.run to keep it after the run, or to continue it after Ctrl+C. Its methods are the only way to change it, and every one is thread-safe, so tools may call them from worker threads. Read values with the properties, or take snapshot() for several values that must belong together.

Example

state = State(messages=[Message.user("Find the bug in this repo")], extra_data={"repo": "api"})
agent.run(state)
print(state.answer, state.stopped, state.usage.cost)

Start a State, empty or from messages, extra data or a history. Everything is keyword-only.

Parameters:

Name	Type	Description	Default
`id`	`str \| None`	The name a store saves this State under: letters, digits, `_` and `-`, at most 128 characters. A random id when left out. Continue a saved State with `store.load(id)`, not with a new State of the same id.	`None`
`messages`	`Iterable[Message]`	The conversation to start from, as `Message` objects. Recorded as one `import` entry. May be empty: add messages later with `add_message`.	`()`
`extra_data`	`Mapping[str, Any] \| None`	Your own values (JSON: dicts, lists, str, int, float, bool, None). The model never sees them. Recorded as one `extra_data` entry.	`None`
`history`	`Iterable[HistoryEntry] \| None`	Entries to rebuild a State from (`other.history`, or what a store loaded). It cannot be combined with `messages` or `extra_data`: the history already holds them.	`None`

Raises:

Type	Description
`TypeError`	A positional argument, a keyword that does not exist (the 0.4 ones, `task=`, `context=` and `data=`, get a message naming their replacement), `messages` that is not a list (a plain `str`) or has an item that is not a `Message`, an `extra_data` value that is not JSON, an item of `history` that is not a history entry, or `history` together with `messages` or `extra_data`.
`ValueError`	`id` has other characters, or `history` is inconsistent (it gives the index of the entry that cannot follow the ones before it).

id `property` ¶

id: str

The name a store saves this State under. Given with State(id=...), or random.

parent_id `property` ¶

parent_id: str | None

The id of the State whose tool call started this one as a subagent, or None for a State that no tool call started. It is the first link in the history (RunStartEntry.parent_id), so it is the same while it runs and after store.load.

parent_call_id `property` ¶

parent_call_id: str | None

The ToolCall.id of the parent's tool call that started this State, or None together with parent_id.

history `property` ¶

history: tuple[HistoryEntry, ...]

Everything that happened, oldest first: messages, model requests and replies, tool results, errors, context changes, runs and stops. Entries are only ever added, so shrinking the context never removes anything from here.

Each entry is one of the HistoryEntry classes (ModelReplyEntry, ToolResultEntry, ...). Checking kind tells which, so type checkers know what content is.

Values in history are read-only and in their JSON form from the moment they are recorded, on a live State as well as a loaded one: a tool call's args and a recorded answer are frozen dicts and lists (indexing, json.dumps and == work, a change raises TypeError; dict(value) makes an editable copy), a tuple becomes a list, and a dataclass or Pydantic model becomes a dict. A value that is not JSON raises TypeError when it is recorded.

Example

for entry in state.history:
    if entry.kind == "model_reply":
        print(entry.content.tool_calls)  # a Reply
    elif entry.kind == "tool_result":
        print(entry.call.name, entry.outcome)

messages `property` ¶

messages: tuple[Message, ...]

The messages the model will see at the next think.

Results of this turn's tool calls go in together once the last call has a result. Messages added in the meantime, or while think waits on the model, go in after them.

pending_calls `property` ¶

pending_calls: tuple[ToolCall, ...]

Tool calls the model requested this turn that have no result yet, in request order.

A call leaves this tuple when it gets a result, is denied or cancelled by a permission, or is closed by an exception.

Every tool call needs a result before the model is asked again. While calls are pending, think, compact, clear_tool_results and restore raise ValueError, add_message waits (the message goes into messages right after the results), and finish works as usual.

turn `property` ¶

turn: int

Number of turns so far: each think adds 1 while it waits on the model, and a think that fails takes it back.

usage `property` ¶

usage: Usage

Total tokens, requests and cost of every think, ask and compact on this State, and of the subagents its tool calls started (added when the call closes). restore does not lower it.

extra_data `property` ¶

extra_data: FrozenDict[str, Any]

A read-only dict for your own values (JSON). The model never sees it.

Change it with edit_extra_data(): state.extra_data["x"] = 1 raises TypeError.

finished `property` ¶

finished: bool

Whether finish() was called. A finished State stops its loop before the next turn, and cannot run again.

answer `property` ¶

answer: Any

The run's answer.

The value given to finish(answer) if there is one, otherwise the text of the latest model reply without tool calls, otherwise None. A think that fails does not change it.

It is in JSON form even before any save: a dataclass or Pydantic model given to finish comes back as a read-only dict (Review(**state.answer) rebuilds it). Use agent.ask(..., returns=Type) to get the object itself.

stopped `property` ¶

stopped: Stopped | None

Why this run is set to stop, or None until something decides that.

StoppedByFinish(answer): finish() was called (set right away)
StoppedByPermission(call, permission): a permission denied call with stop=True (set right away, when use_tools records the results)
StoppedByUntil(name): an until function returned true, e.g. StoppedByUntil("waiting_for_user") (set when @loop checks it before a turn)
StoppedByLimit(turns): the loop ran limit turns in this call; turns is the limit (set when @loop checks it before a turn)

run/arun reset it to None at the start, and it is None after a run that ended with an exception. A @loop stops before its next turn once it is set. A loop written without @loop checks it itself. A @loop called inside such a loop leaves its until/limit reason set, so a loop without @loop that calls one checks only for StoppedByFinish/StoppedByPermission.

Example

agent.run(state)
if isinstance(state.stopped, StoppedByLimit):
    print(f"gave up after {state.stopped.turns} turns")

def my_loop(agent, state):
    while state.stopped is None and not state.finished:
        agent.think(state)
        if state.pending_calls:
            agent.use_tools(state)

created_at `property` ¶

created_at: datetime | None

When the first thing was recorded: the at of the first history entry, in UTC. None for a State with no history yet.

updated_at `property` ¶

updated_at: datetime | None

When something was last recorded: the at of the latest history entry, in UTC. None for a State with no history yet.

Anything added to history moves it, including changes to extra_data.

snapshot ¶

snapshot() -> StateSnapshot

The current value of this State: a frozen StateSnapshot that never changes, and that restore can go back to.

Example

before = state.snapshot()
state.add_message(Message.user("Try the other approach"))
state.restore(before)

add_message ¶

add_message(message: Message) -> None

Add a user message or a notice to the conversation, for example the next request in a chat loop.

While tool calls are pending, or while think waits on the model, the message is held and goes into messages right after the results or the reply. A loop waiting for the user stays waiting until the model answers it.

Parameters:

Name	Type	Description	Default
`message`	`Message`	`Message.user(text, *images)`, or `Message.notice(text)` to tell the model something from your loop or a tool (a user message whose text starts with `"[notice] "`).	required

Raises:

Type	Description
`TypeError`	`message` is not a `Message`.
`ValueError`	`message` is not a user message, has blocks other than text and images, or has no text and no image.

Example

state.add_message(Message.user("Now fix it"))
state.add_message(Message.notice("The tests failed"))

finish ¶

finish(answer: Any = None) -> None

End the run. state.stopped becomes StoppedByFinish(answer) right away, and the loop stops before its next turn.

Tools may call it (a submit tool, for example), even while other calls are pending. Calling it again is allowed; the last answer given wins.

It ends the State for good: afterwards think, use_tools, ask and run raise ValueError, and only restore to a snapshot taken before it takes it back. If a tool calls it, end the loop body without another think. A stop by a permission does not do this: the State is not finished, so add a message and run again.

Parameters:

Name	Type	Description	Default
`answer`	`Any`	If not `None`, becomes `state.answer`. Any JSON value, a Pydantic model or a dataclass (turned into a dict), not only a string. `None` keeps an earlier answer.	`None`

Raises:

Type	Description
`TypeError`	`answer` cannot be turned into JSON (a set, `Path`, `datetime`, `Enum` or `bytes`, for example).

compact ¶

compact(summary: str) -> None

Replace the messages with the first user message and summary. History keeps everything.

Use it with a summary you wrote or got from agent.ask. agent.compact(state) asks the model for the summary.

Parameters:

Name	Type	Description	Default
`summary`	`str`	What the model should know about the work so far.	required

Raises:

Type	Description
`TypeError`	`summary` is not a string.
`ValueError`	Tool calls are pending, or `summary` is empty or whitespace only.

clear_tool_results ¶

clear_tool_results(keep_last: int = 5) -> None

Shrink the context by blanking old tool results. History keeps the full results.

Every tool result except the last keep_last is replaced with "(cleared: kept in history)". Images in a result are cleared with its text. Results that are already cleared still count toward keep_last. Nothing is recorded when nothing changes.

Parameters:

Name	Type	Description	Default
`keep_last`	`int`	How many of the most recent tool results to keep.	`5`

Raises:

Type	Description
`ValueError`	Tool calls are pending, or `keep_last` is not an integer >= 0.

edit_extra_data ¶

edit_extra_data() -> Iterator[dict[str, Any]]

Change extra_data inside a with block. It gives you an editable copy (plain dicts and lists), and when the block ends normally the changes are recorded as one history entry.

Edits take turns: a lock held for the whole block makes two threads' edits run one after the other, so data["n"] = data.get("n", 0) + 1 never loses an update. If the block raises, nothing changes and the exception propagates. If nothing changed, nothing is recorded.

Raises:

Type	Description
`TypeError`	At the end of the block, a value is not JSON (a set, an object, a non-string key, NaN). The message names the key.

Example

with state.edit_extra_data() as data:
    data.setdefault("notes", {})["bug"] = "b.py"
    data.pop("draft", None)

fork ¶

fork() -> State

A new State with the same history and so the same content, a new id, and nothing shared with this one: no store, no running Agent. Use it to try something on a copy.

The copy is the same State at that point, so a fork of a finished State is finished too (start a new State from state.messages to continue). Fork between turns: while a model request is being waited on there is no think() behind the copy, so fork() refuses (fork before think, or after it returns, e.g. in Reporter.on_think_end). State(history=snapshot.history) of such a snapshot is still allowed: it is a plain replay.

The copy is independent: running or changing it leaves this State as it was, and fork().snapshot() == state.snapshot(). With Agent(store=...), the copy is saved under its own id the first time the Agent runs it. The one thing it shares is the subagent States this State ran, which it can open with child too (they do not change because of the copy): the ones in memory, and, for a State from store.load, the ones in that store.

Example

attempt = state.fork()
agent.run(attempt)  # `state` is untouched

Raises:

Type	Description
`ValueError`	The model is being waited on.

child ¶

child(child_id: str) -> State

The subagent State with id child_id, one that a tool call of this State started (one of the child_ids of a ToolResultEntry in the history). A grandchild is state.child(a).child(b).

While the process that ran the subagents is alive, this is the very State object the subagent ran with (state.child(a) is state.child(a)). After store.load, or in a State that did not run them, it is loaded from the store this State is bound to. That State is not cached: every call loads a new one, so state.child(a) is state.child(a) is False then. A subagent whose Agent has a store of its own is not in this State's store, so it cannot be opened from here: open it with that store's load(child_id).

Example

for entry in state.history:
    if entry.kind == "tool_result":
        for child_id in entry.child_ids:
            print(entry.call.name, state.child(child_id).answer)

Raises:

Type	Description
`KeyError`	`child_id` is not a child of this State (a grandchild is not either).
`ValueError`	It is a child, but this State has no way to open it: it is not in memory and no store is bound to this State.
`TypeError`	The store works only asynchronously. Use `achild`.
`LookupError`	The store has no State with this id (a subagent with a store of its own).

achild `async` ¶

achild(child_id: str) -> State

The async version of child: the same lookup and the same errors, but a child that is not in memory is loaded with await store.aload(child_id). Use it with a store that works only asynchronously. Like child, it does not cache what it loads from the store.

Example

state = await store.aload("bug-hunt-1")
researcher_state = await state.achild(child_id)

restore ¶

restore(snapshot: StateSnapshot) -> None

Go back to snapshot, a state.snapshot() taken earlier from this State.

Every value goes back to what it was then (messages, turn, extra_data, finished, answer, stopped), except usage, which stays cumulative (the tokens were spent), and history, which keeps everything and gets one context_change entry saying the State went back.

Parameters:

Name	Type	Description	Default
`snapshot`	`StateSnapshot`	A snapshot of this State, taken between turns.	required

Raises:

Type	Description
`TypeError`	`snapshot` is not a `StateSnapshot`.
`ValueError`	`snapshot` is not from this State, tool calls are pending, the model is being waited on, a compaction is under way, or the snapshot was taken while the model was being waited on.

StateSnapshot¶

What state.snapshot() returns: the frozen value of a State at one moment. Pass it to state.restore(snapshot).

StateSnapshot `dataclass` ¶

StateSnapshot

The value of a State at one moment: what state.snapshot() returns, and what every state.<property> reads. It is frozen, so it never changes while you hold it, and it has no methods that change anything.

Every field is a pure function of history: building a State from the same history (State(history=...)) gives an equal snapshot. == compares the public fields, so equal histories give equal snapshots. Pass a snapshot to state.restore(snapshot) to go back to that point.

A snapshot lives in memory. To keep one across processes, keep its history: State(history=snapshot.history) rebuilds the same value, and so does a store.

Example

before = state.snapshot()
agent.run(state)
if before.usage != state.usage:
    state.restore(before)  # undo the run's effect on the context

history `class-attribute` `instance-attribute` ¶

history: tuple[HistoryEntry, ...] = ()

Everything that happened, oldest first. Entries are only ever added.

messages `class-attribute` `instance-attribute` ¶

messages: tuple[Message, ...] = ()

What the model will see at the next think. Results of this turn's tool calls go in together once the last call has a result. Messages added in the meantime, or while the model is being waited on, go in after them.

pending_calls `class-attribute` `instance-attribute` ¶

pending_calls: tuple[ToolCall, ...] = ()

The tool calls of the open turn that have no result yet, in request order.

turn `class-attribute` `instance-attribute` ¶

turn: int = 0

Completed model replies, plus one while a request is waiting on the model.

usage `class-attribute` `instance-attribute` ¶

usage: Usage = field(default_factory=Usage)

Tokens, requests and cost of every think, ask and compact. It is cumulative: restore does not lower it, because the tokens were spent.

extra_data `class-attribute` `instance-attribute` ¶

extra_data: FrozenDict[str, Any] = field(default_factory=FrozenDict)

Your own values (JSON), which the model never sees. Read-only: change them with state.edit_extra_data().

finished `class-attribute` `instance-attribute` ¶

finished: bool = False

Whether finish() was called (and not undone by restore). A finished State cannot run again.

answer `class-attribute` `instance-attribute` ¶

answer: Any = None

The value given to finish(answer) if there is one, otherwise the text of the latest reply without tool calls, otherwise None.

stopped `class-attribute` `instance-attribute` ¶

stopped: Stopped | None = None

Why the current or last run is set to stop, or None (a new run clears it).

created_at `class-attribute` `instance-attribute` ¶

created_at: datetime | None = field(default=None, compare=False)

The at of the first history entry, or None while the history is empty. Not compared by ==.

updated_at `class-attribute` `instance-attribute` ¶

updated_at: datetime | None = field(default=None, compare=False)

The at of the latest history entry, or None while the history is empty. Not compared by ==.

State¶

State ¶

id property ¶

parent_id property ¶

parent_call_id property ¶

history property ¶

messages property ¶

pending_calls property ¶

turn property ¶

usage property ¶

extra_data property ¶

finished property ¶

answer property ¶

stopped property ¶

created_at property ¶

updated_at property ¶

snapshot ¶

add_message ¶

finish ¶

compact ¶

clear_tool_results ¶

edit_extra_data ¶

fork ¶

child ¶

achild async ¶

restore ¶

StateSnapshot¶

StateSnapshot dataclass ¶

history class-attribute instance-attribute ¶

messages class-attribute instance-attribute ¶

pending_calls class-attribute instance-attribute ¶

turn class-attribute instance-attribute ¶

usage class-attribute instance-attribute ¶

extra_data class-attribute instance-attribute ¶

finished class-attribute instance-attribute ¶

answer class-attribute instance-attribute ¶

stopped class-attribute instance-attribute ¶

created_at class-attribute instance-attribute ¶

updated_at class-attribute instance-attribute ¶

id `property` ¶

parent_id `property` ¶

parent_call_id `property` ¶

history `property` ¶

messages `property` ¶

pending_calls `property` ¶

turn `property` ¶

usage `property` ¶

extra_data `property` ¶

finished `property` ¶

answer `property` ¶

stopped `property` ¶

created_at `property` ¶

updated_at `property` ¶

achild `async` ¶

StateSnapshot `dataclass` ¶

history `class-attribute` `instance-attribute` ¶

messages `class-attribute` `instance-attribute` ¶

pending_calls `class-attribute` `instance-attribute` ¶

turn `class-attribute` `instance-attribute` ¶

usage `class-attribute` `instance-attribute` ¶

extra_data `class-attribute` `instance-attribute` ¶

finished `class-attribute` `instance-attribute` ¶

answer `class-attribute` `instance-attribute` ¶

stopped `class-attribute` `instance-attribute` ¶

created_at `class-attribute` `instance-attribute` ¶

updated_at `class-attribute` `instance-attribute` ¶