soev_agents.runtime

One agent: the whole contract between it and the server that runs it.

An agent says which class each fresh input's payload must parse into, strictly, and builds the graph for a parsed turn. ReActAgent in soev_agents.toolkit is a ready agent. Write your own class with these attributes and build for any other graph.

An async function that returns the resources for an agent's build.

The catalog calls it each time that it builds a graph, never for a read, a cancel or a fork. It gets the agent, so one function can serve agents that need different resources.

Parse an input's payload, strictly, and return the input with the elements that the payload holds.

The catalog calls it for each fresh input, so that the thread records the elements of the input. Code that builds graphs without the catalog calls it too.

Raises:

• InputDeniedError: The payload does not parse into input_type.

The nodes that run a turn, and the edges between them.

A fresh input starts at the entry point. After each node, the turn follows the edge from that node. A node without an edge ends the turn, as does a Request, which waits for a response. A step that raises stops the run with a Failure. If the failure is transient, the node runs again when the thread runs again.

Build a graph once for each turn, in Agent.build. The runner does not change it.

from soev_agents.domain import ModelOutput, Trace
from soev_agents.runtime import Graph, LlmStep, ToolStep

# builder is a ContextBuilder, model is a ChatModel, and tools are the Tool objects of the turn.
def after_model(view: Trace) -> str | None:
    output = view.latest(ModelOutput)
    return "tools" if output is not None and output.calls_tools else None

graph = Graph()
graph.add_node("model", LlmStep(builder=builder, model=model))
graph.add_node("tools", ToolStep(tools={tool.name: tool for tool in tools}))
graph.set_entry_point("model")
graph.add_conditional_edge("model", after_model)
graph.add_edge("tools", "model")

Make an empty graph, without nodes, entry point or budget.

A named node of a graph and the step that it runs.

Graph.add_node makes nodes. Read them back with Graph.node.

Where a turn goes after a node: fixed targets, or a router that picks one.

Graph.add_edge and Graph.add_conditional_edge make edges. Read them back with Graph.edge.

A function that picks the next node from the trace.

The runner calls it with the view of the stream after the node of the edge recorded its events. It returns the name of the next node, or None to end the turn. A router is a pure function of the trace: if it raises or names a node that the graph does not have, the runner records a Failure and does not try again.

A limit on the model calls that call tools in one turn of a stream.

When a model output that calls tools brings the count past cap, the runner records BudgetExceeded instead of the next node. The turn then resumes at resume_at, at most tolerance times. After the last of these, the next model call gets no tools, so the model can only write its final output. One more trip ends the turn.

The model call loses its tools only when no event other than a Compaction comes between the trip and the call.

Another graph that a node runs as subagents, one for each task that hand_offs finds in the parent's view.

A node whose step is a Delegation starts one subagent for each task, each in a stream of its own. When all subagents ended, the node records what hand_back returns for each of them in the parent stream, and the parent continues on the node's edge.

An async function that reads the trace and returns the events that its node records.

The runner calls it with the view of the node's stream, and records what it returns with the node's name as origin. It must return at least one event, no Input, and a Request only as the last event. If it raises, the runner records a Failure and the run stops. A TransientError makes that failure transient, so the node runs again when the thread runs again.

A step that makes one model call with the context that builder builds, and records the output.

The model's output records:

• the model that replied: the model of the call, unless the adapter names another, as a fallback does
• the characters that the model was sent, and the hash of the ChatRequest
• its tool calls, with the numbers that they name translated to real ids and call ids

When the context has a citation strategy, the step removes the markers from the output's text and resolves each number to the source that it named. Streamed text reaches a listener without markers, and each citation follows its text.

A step that runs the tool calls of the latest ModelOutput and records one result for each call.

Each call's result is what its tool returns. When the tool is unknown or the arguments do not parse, the result is a ToolOutput with an error instead, and the tool does not run.

Other results, for which the tool does not run:

• A call that is identical to an earlier call of the turn, by tool name and arguments, gets an error that says so, when the earlier call succeeded or is in the same model output. repeats lets each tool run it again, and a tool's own repeats replaces the step's.
• A call to a tool that its turn_limit says is spent gets the limit's error.

The calls run in parallel. A call to a tool with a limit waits for the calls before it in the same model output and sees their results, so the results are the same in any order of completion. An argument of type Recorded gets the elements of its type that the turn recorded. A tool that raises TransientError, InvalidQueryError or ResourceUnavailableError gets that error as its result. Any other exception stops the step, and the run records a Failure.

Raises:

• MutableToolError: When the step is made: a tool, or an object that tools share, could keep state between calls.
• ToolDefinitionError: When the step is made: a tool's run takes more than the call's arguments, or a tool has parameters that still need a value (unbound).

A step that stops the turn to wait for a response from outside, such as an approval.

It records one Request. The thread then waits. The next input is the response, and the turn continues on the edge of this node, not at the entry point.

Everything that one model call receives, and what its output is read with.

A ContextBuilder makes it from the trace, and LlmStep sends it.

A function that builds the context of one model call from the trace alone.

It is synchronous, so it cannot call a model, a store or any other resource: the same code and the same trace always give the same ChatRequest. MessageBuilder in soev_agents.toolkit is the usual builder.

How one model is told to cite and how its markers are read back.

soev_agents.toolkit has ready strategies, such as BRACKETS for markers like [1].

A node that changes the world exactly once: it records its intent before the effect.

When the node runs, intend returns the intent and the runner records it. Then effector performs it, and the runner records the result, linked to the intent. If the process stops between the two, the next run finds the intent without a result and calls recovery instead of a second effect. If the caller cancels the turn during the effect, cancellation decides. A response to a Request of the step returns the turn to the same node.

The adapter that changes the world for an EffectStep: it sends the mail, or makes the refund.

An async function that resolves an intent without a result, after a crash during its effect.

The runner calls it when the effect step's node runs again and the stream's last intent has no result. It gets the view, the intent and the effector, and it can ask the world whether the effect happened. It must return exactly one event: the EffectResult of the intent, or a Request that asks a person. A response to that request brings the node back to its recovery, with the response in the view.

An async function that decides about an effect in progress when the caller cancels the turn.

It gets the view, the intent, the effector and the asyncio task that performs the effect. It can await that asyncio task, or ask the world. It returns the one EffectResult of the intent, one Request that asks a person, or nothing. Nothing leaves the intent without a result for the step's recovery. The runner cancels the asyncio task when the function returns. Use abandon to give the effect up at once.

Give up the effect in progress: the EffectCancellation that awaits nothing and records nothing.

intent stays without a result, so the effect step's recovery resolves it when the thread runs again.

Return the decorator that makes an async function the tool that the model calls as name.

The parameters before * are what the model writes, and the tool's argument schema comes from their types and defaults. A parameter after * has the type Recorded, or keeps its default, or gets its value later from FunctionTool.bind. The model reads description, never the function's docstring.

from soev_agents.domain import ToolOutput
from soev_agents.runtime import tool

# SearchIndex is your own search client class, and resources are the resources that `Agent.build` gets.
@tool(name="search", description="Search the documents. Use short keyword queries.")
async def search(query: str, *, index: SearchIndex) -> ToolOutput:
    ...

agent_tools = [search.bind(index=resources.index)]

The decorator that tool(...) returns. Applied to a function, it makes the FunctionTool.

A tool that the tool decorator made of an async function.

run calls the function with the call's arguments and the Recorded values. Calling the tool itself calls the function, with its own signature and result type, so a test can call it directly.

What the model can call: a name, a description, the arguments that the model writes, and run.

run gets only the call's arguments, parsed into arguments, with each element id already real. What the turn recorded so far reaches a tool as a Recorded argument, never as the trace. A tool is a frozen dataclass whose fields hold adapters, code constants and values of the turn's input. ToolStep refuses any other tool, so that no tool keeps state between calls. The tool decorator makes a tool of an async function.

A tool can also have these attributes, which ToolStep reads:

• turn_limit: a Limit for each turn, for example at_most(3, calls="search", error="You searched three times already.").
• repeats: whether a call that is identical to an earlier successful call of the turn runs again. It replaces the step's repeats for this tool.
• unbound: the parameters that still need a value. ToolStep refuses a tool that names any, so a missing value fails when the graph is built, not when the model calls the tool.

Return what the model is shown of tool, with the element id fields that its calls record.

The schema of the arguments has no titles and no class docstrings: the model reads the property names and the field descriptions.

Raises:

• ValueError: An ElementId in tool.arguments sits where no call could be translated.

A mark on a tool argument that holds the id of an element of one of accepts, subclasses included.

Use it as Annotated[str, ElementId(Document)], alone, in a list or optional. The model writes the number that it was shown, and LlmStep replaces the number with the real id when it records the call. The mark itself changes no validation, so code can build arguments from real ids. The JSON schema shows a plain string.

A field with an element type instead, such as Annotated[MailReference, ElementId()], gets the element first recorded under that id. ElementId() without classes is only for a field with an element type. It accepts the element classes of that type.

Make the mark.

A mark on a tool argument that holds the handle of a tool result, such as "r12".

Use it as Annotated[str, ResultHandle()], alone, in a list or optional. The model writes the handle that it was shown, and LlmStep replaces it with the call id of the tool call that has that result, as it does for ElementId.

The type of a tool parameter that gets the elements of one type that the turn recorded so far.

Write Recorded[Document] as the type of a parameter after *, or of a field of the tool's arguments. Before each call, ToolStep fills it with every Document of the turn, the input's included, in recorded order. The model never sees the parameter, and a value that the model writes for it is replaced. This is how a tool reads what the turn built up without the trace.

A limit of one tool for each turn.

It gets the trace that a call would see, and returns the error that the model reads when the tool is spent, or None while the tool can run. Give a tool a limit as its turn_limit. at_most makes the usual limit.

Return a limit that makes error a call's result, and does not run the tool, once the turn reached n.

The limit counts the calls with a result of the tool named calls, or the distinct elements of the type elements. An element recorded again counts once. Give the limit to a tool as limit in tool, or as its turn_limit.

Raises:

• ValueError: n is below 1, or not exactly one of calls and elements is given.

A limit of one tool for each turn, which at_most makes.

Exactly one of calls and elements is set.

A tool is defined so that it could reach state, or so that the model could not call it.

Make the error.

A tool, or an object that tools share, could keep state between calls.

The object is not a frozen dataclass, or one of its fields holds a mutable value: a list, dict, set, bytearray or asyncio primitive. ToolStep raises it when it is made, and FunctionTool.bind raises it for a mutable value.

A tool returned something other than one ToolOutputBase, so its call has no single result.

A tool showed an element whose ref names nothing recorded, in its output or earlier.

The model would see a handle it can do nothing with, so a tool that shows a part records its whole beside it.

A catalog of agents, by name, over one store: the object that opens, runs and reads threads.

open makes a thread under an agent's name and records Opened with the first input. thread reads that Opened back and returns the Thread, with a graph that its agent built for the turn that the caller is about to run, or without a graph for a caller that only reads.

A fresh input is parsed by its agent's input class and recorded with the elements that it holds, so a thread's numbering never depends on the agent's code.

from soev_agents.adapters import InMemoryTraceStore
from soev_agents.domain import Input
from soev_agents.runtime import AgentCatalog
from soev_agents.toolkit import ReActAgent, ReActConfig

# chat_model: a ChatModel adapter, such as an OpenAIChatModel.
config = ReActConfig(model=chat_model, prompt="Answer briefly.")
catalog = AgentCatalog(InMemoryTraceStore(), {"assistant": ReActAgent(config)}, version="1.4.0")
thread, _ = await catalog.open("assistant", Input(payload="Hello", model="my-model"))
trace = await thread.resume()

Make a catalog.

One thread of a store: it reads the thread, takes inputs, runs its graph, cancels and forks.

Get a thread from AgentCatalog.open or AgentCatalog.thread. A thread from the catalog takes no fresh input other than the input that its graph was built for. To start the next turn, get a new thread with AgentCatalog.thread(thread_id, input=...). A thread from the catalog without a graph reads, takes responses to requests, cancels and forks, but cannot run.

The usual calls:

• send records an input and runs the turn to its end. submit only records the input.
• stream runs the turn and yields each event as it is recorded. resume runs it to its end.
• status reads the state of the current turn, and trace reads all events.
• cancel ends the current turn, fork copies the thread up to a position, and delete removes it.

Make a handle on a thread. Most callers get a thread from an AgentCatalog instead.

The running asyncio tasks of this process, one per thread.

A caller joins the running asyncio task instead of a second one, so one process never runs a thread twice. Across processes each run holds the thread's lease in the store, under this registry's owner, for lease_ttl seconds at a time.

Make one Runs for each process and give it to the AgentCatalog. Call close when the process stops.

Make an empty registry.

The state of a thread's current turn, read from its trace and its lease.

The state of a thread's current turn, read from its trace and its lease alone.

• "running": a process holds the thread's lease.
• "waiting": a request waits for a response.
• "failed": a step failed, and the run stopped on the Failure.
• "interrupted": the run stopped without an end of the turn, for example in a crash. Resume the thread.
• "cancelled": the caller cancelled the turn.
• "finished": the turn ended, or the thread has no turn yet.

An input was refused, and nothing was recorded.

The payload does not fit the agent's input class, a response does not fit the schema of its request, the request that it responds to is not waiting, or Agent.build refused the turn. The default HTTP hooks respond with 422.

The thread's turn has not ended, so it takes no fresh input.

The turn is running or was interrupted, a request waits for its response, or an effect intent has no result. A response to a waiting request is accepted. Resume or cancel the turn first. The default HTTP hooks respond with 409.

A thread with this id already has events, so it cannot be opened again.

The catalog has no agent of this name, or the thread names no agent.

No thread has events under this id.

The turn is finished, failed or cancelled already, or the named input did not open the current turn.

A log position is outside the thread, for example the position of a fork.

The thread runs another agent than the one that the caller named.

Return whether the caller cancelled the current turn: its root stream holds a Cancelled.

A step that does long work can check it, but the runner already stops the turn after a cancel.

Return the root input that opened the current turn, None for a thread without one.

It reads the store backwards only as far as that input, so a server can build the turn's graph before it loads the whole thread.

Return the positions in events of the inputs that open a turn, each judged in its own stream.

An input opens a turn when it responds to no request: the event before it in its stream is not a Request.

Return the current turn's requests that wait for a response, in all streams of the turn.

After a cancel, only the requests that ask for the outcome of an effect with an unknown result still wait.

Number every element in trace by first appearance, a ref before the element naming it.

A real id keeps its number wherever it appears again, and the numbers do not change as the thread grows.

Each tool result with a call id gets a handle, r1 first, in recorded order.

Each real id's number, and the element that first showed it.

The model never sees a real id: it sees the number, and it cites and passes that number to tools. numbering makes a numbering from the trace.

The event and element classes stored events are read back into, each by its type tag.

CORE_EVENT_TYPES holds the types of the framework. Add an agent's own types with including, or use event_types_of.

Make the set of types.

Raises:

• ValueError: Two event classes share a type tag.

The event and element types of the framework, which every thread records.

A thread of an agent with its own types is read with event_types_of([agent]), which adds them to these.

Return the core event types plus every class that agents declare their threads record.

Use it to read stored threads of these agents, for example in replay or a store migration.

Raises:

• ValueError: Two event classes share a type tag.

Return event as a StoredEvent: its type tag, its bookkeeping fields and its payload as JSON.

A store keeps each event as a StoredEvent. EventTypes.decode reads it back. Use it to write a trace directly into a store, for example in a test or an import.

Raises:

• TypeError: The event's class declares no type tag.

A stored event has a type tag that is not among the types that the thread is read with.

Add the event class to the agent's event_types, or read the thread with the types that event_types_of returns for its agents.

An event would not read back as it was recorded.

Its type, or the type of an element that it holds, is not among the types that the thread is read with, or one of its fields does not survive JSON. A thread raises it before it records the event, and the runner records a Failure instead of an event from a step.

Rebuild every model call in events, one thread's events in log order.

graph_for builds the agent's graph for the turn opened by an input, as the server did when the turn ran; it is called once per turn, and its models are asked to prepare but never to reply. Nothing is executed, so the result is what today's code renders, which differs from what was sent if the agent changed since.

Use it to see what each model call of a recorded thread received, or what a change of a prompt or a context policy does to recorded threads. Build the graphs with refusing adapters, so that a build that calls a resource fails at once.

from soev_agents.domain import Input
from soev_agents.runtime import Graph, accepted, refusing, replay

# agent is the Agent of the thread, and thread is a Thread with recorded turns.
def graph_for(input: Input) -> Graph:
    return agent.build(refusing("resources"), accepted(agent.input, input)[1], input.model)

events = [entry.event async for entry in thread.watch(follow=False)]
calls = await replay(events, graph_for)

One model call of a thread, as replay rebuilt it.

Return an adapter that refuses every call with ResourceUnavailableError, whatever the method.

Give it to build in place of each resource when replay builds a graph, so a build that calls an adapter fails at once.