dex-agents
dex-agents (import dex_agents) is the Python core of the Dex agent
framework adapters. It provides one durable agent session Flow,
AgentSessionFlow, and the AgentFrameworkRunner contract that each
framework adapter implements. The plan is
agent-framework-adapters.md.
One Flow execution is one agent session; the Flow ID is the session ID. The
session runs in Step mode: every model call is one RunModel Step, and
every tool call, READ included, is its own Step. The tool call Steps of one
model response run in parallel, and ResumeTurn waits for their results
before the next model call. Each call therefore has its own retry policy,
timeout, failure route, history entry, and Dex Web node, and a crash repeats
at most the one call that was running.
Installation
Applications usually install an adapter, which brings in this package. To
build a runner of your own, install the core directly. It needs Python 3.11 or
later and requires the released dex-python-sdk==1.38.0:
pip install dex-agents
dex_agents.testing ships in the same wheel, so an adapter's integration
tests can start a private engine and Worker processes. So does
dex_agents.ui_stream, the browser streams.
dex_agents.mcp offers the tools of MCP servers as agent tools; install it
with pip install "dex-agents[mcp]", which adds the MCP SDK (mcp>=2.3,<3).
See MCP tools.
For OpenTelemetry spans, install the otel extra, which adds only
opentelemetry-api; see Tracing.
The package has these modules. Optional dependencies are extras, so no module needs a new package name:
| Module | Install | Contents |
|---|---|---|
dex_agents |
dex-agents |
The session Flow, the runner contract, built-in tools, AgentSessionClient, and tracing options |
dex_agents.stepwise |
dex-agents |
Stepwise mode: AgentDefinition, function tools, tool sources, the ModelClient protocol, ScriptedModelClient, and the runner |
dex_agents.litellm |
dex-agents[litellm] |
LiteLLMModelClient, a ModelClient on LiteLLM |
dex_agents.mcp |
dex-agents[mcp] |
McpToolSource, MCP servers' tools as agent tools |
dex_agents.ui_stream |
dex-agents |
Server-Sent Event frames and snapshot JSON for UIs |
dex_agents.testing |
dex-agents |
Private engines, Worker processes, gates, Step history, and ScriptedModelClient (the same class as dex_agents.stepwise's) |
Spans need the otel extra. Importing dex_agents.litellm without its extra
raises ImportError that names dex-agents[litellm].
Develop the package inside this repository with uv 0.12.2. The adapters in
this repository resolve it from ../core by path through
[tool.uv.sources]; their published wheels require it from PyPI instead:
cd agents/python/core
uv sync --locked
DEXCLI=/path/to/dexcli ./ci.sh
An adapter requires dex-agents==<same version>. All agents packages share
one version line; see the
compatibility table.
Running an agent
from dex import AsyncClient, AsyncWorker, BlobCacheConfig, ClientOptions, Registry
from dex import WorkerOptions, open_blob_cache
from dex_agents import AgentSessionClient, AgentSessionFlow, AgentSessionOptions
from dex_agents import ToolEffect, ToolPolicy
flow = AgentSessionFlow(
MyFrameworkRunner(agent),
flow_type="TriageAgent",
options=AgentSessionOptions(
tool_policies={"search_tickets": ToolPolicy(ToolEffect.READ)},
),
)
registry = Registry([flow], allow_async_handlers=True)
cache = open_blob_cache(BlobCacheConfig(".dex-cache", 64 * 1024 * 1024))
worker = AsyncWorker(registry, cache, WorkerOptions(server_address="localhost:8801"))
client = AsyncClient(registry, cache, ClientOptions(server_address="localhost:8801"))
sessions = AgentSessionClient(client, flow)
await sessions.start_session("session-42", "Find my open tickets")
snapshot = await sessions.get_agent_snapshot("session-42")
Register one AgentSessionFlow per agent, each with its own flow_type. All
Steps are async, so the Registry needs allow_async_handlers=True and an
AsyncWorker serves it. Tool calls run as parallel Steps, so give the Worker
enough task slots (PullWorkerOptions.max_concurrent_tasks) for the largest
batch of tool calls a model makes. A session never completes by itself; it
waits for the next user message after every turn. Stop it with
AsyncClient.stop_flow.
Stepwise mode
Stepwise mode is the built-in agent loop, for agents defined in Dex terms
without a framework. StepwiseAgentRunner implements the runner contract
below, so the session Flow, its Steps, approvals, and events are the ones
every adapter uses. Each RunModel makes exactly one model call through a
ModelClient, and every tool call is its own Step.
from dex import Registry
from dex_agents import BuiltInTool, ToolEffect
from dex_agents.litellm import LiteLLMModelClient
from dex_agents.stepwise import AgentDefinition, ToolCallContext, function_tool
from dex_agents.stepwise import stepwise_agent_session_flow
@function_tool(effect=ToolEffect.READ)
async def lookup_order(order_id: str) -> dict[str, str]:
"""Look up an order by its ID."""
return await orders.get(order_id)
@function_tool(effect=ToolEffect.NON_IDEMPOTENT_WRITE)
async def refund_order(order_id: str, amount_cents: int, context: ToolCallContext) -> str:
"""Refund part of an order."""
return await payments.refund(order_id, amount_cents, idempotency_key=context.idempotency_key)
support_agent = AgentDefinition(
name="support",
instructions="You help customers with their orders.",
model=LiteLLMModelClient("openai/gpt-5-mini"),
tools=[lookup_order, refund_order],
built_in_tools=frozenset({BuiltInTool.ASK_USER, BuiltInTool.DURABLE_WAIT}),
)
flow = stepwise_agent_session_flow(support_agent, flow_type="SupportAgent")
registry = Registry([flow], allow_async_handlers=True)
AgentDefinition fields: name, instructions (a string, or a function of
InstructionContext evaluated in every RunModel), model, model_settings,
tools, tool_sources (see Tool sources), tool_policies,
default_tool_effect, built_in_tools,
max_model_calls_per_turn (100; after that many calls in one turn the next
call offers no tools, and a tool call the model makes anyway gets the result
The tool call was not run: this turn reached its limit of <n> model calls.
while the turn ends), interrupt_deadline, event_stream_capacity_bytes, and
plan_mode_enabled (see Plans and plan mode).
A function tool may not share a name with an enabled built-in tool:
AgentDefinition refuses it, and so does run_model_step, before any model
request, when only the session options enable that built-in.
stepwise_agent_session_flow derives the AgentSessionOptions from it.
@function_tool(effect=..., requires_approval=None, name=None, description=None, parameters_schema=None) turns a plain sync or async
function into a tool. The name defaults to the function name, the description
to the docstring, and the JSON Schema is derived from the type hints with the
standard library: str, int, float, bool, None, list, dict,
Literal, Enum, unions, dataclasses, TypedDict, and
Annotated[T, "description"]. A parameter typed ToolCallContext receives the
idempotency key, attempt, Flow ID, turn ID, tool call ID, event writer,
heartbeat, and the answers to the tool's questions. A sync function runs in a
worker thread. Arguments that break the schema, and an unknown tool name,
raise ToolExecutionError, which the model sees: arguments that are not JSON
give invalid arguments for <tool>: arguments are not valid JSON. A str result is the
output; any other result is serialized as JSON.
In RunModel, the runner builds the model input from committed state only:
the instructions, every stored item, the plan note from build_plan_note
before the turn's items when a plan exists or the turn plans, and the results
of the previous model response from the final records. Every stored tool call is paired with a
result; a missing one becomes the outcome-unknown text, so an orphaned call
never reaches a provider. It stores items as they appear
(ToolCallsRequested.new_messages), so FrameworkRunState holds only the
turn's model call count.
Tool sources
AgentDefinition.tool_sources takes tools that are listed at run time, such
as an MCP server's tools. Any object with a source_name and list_tools,
deferred_tool_call, execute_deferred_tool_call, and poll_tool_task is a
ToolSource; McpToolSource is one:
from dex_agents.mcp import McpStdioServerConfig, McpToolSource
files = McpToolSource(
[McpStdioServerConfig(name="files", command="mcp-server-files", args=["/srv/reports"])],
resource_and_prompt_tools_enabled=True,
)
agent = AgentDefinition(
name="reports",
instructions="You manage reports.",
model=LiteLLMModelClient("openai/gpt-5-mini"),
tools=[lookup_order],
tool_sources=[files],
built_in_tools=frozenset({BuiltInTool.ASK_USER}),
)
In every RunModel of a chat or executing turn, the runner awaits each
source's list_tools, for at most tool_source_listing_timeout (10 seconds),
and offers the tools after the function tools and before the built-in tools.
A source whose listing raises or takes longer offers no tools in that model
call, with a warning log, so a down server never fails a turn. A planning
turn, and a call after the model call limit, list nothing.
A call of a listed tool becomes the source's deferred_tool_call, so the
source's policy, effect class, and tool_source_reference apply: approvals,
CallTool or CallSingleAttemptTool, the idempotency key, and tool tasks
work as for any tool. The runner passes the definition it offered as
listed_tool, so a listing refreshed during the model call cannot change it.
It stores the source's reference as {"stepwise_tool_source": <source_name>, "reference": ...}, so the tool Step and PollToolTask reach the source that
built the call, with its own reference, on any Worker, without a listing, and
whatever the order of tool_sources. Source names must be unique and stable
across deploys: McpToolSource defaults its source_name to "mcp", which
adding or removing a server keeps, and takes source_name=. Two
McpToolSources in one agent must therefore name at least one of them. A stored call whose source name this Worker does not have raises
LookupError, so its effect class decides the outcome. A name that no listing
offered, such as a tool of a server that was down, fails as unknown tool <name> without an approval (unclassified, no approval): it never reaches a
source.
A listed name that repeats a function tool, an enabled built-in tool, or
another source's tool fails the RunModel attempt with ValueError before any
model request: tool <name> uses the name of an enabled built-in tool, tool <name> of a tool source uses the name of a function tool, or two tool sources list a tool named <name>. A source may define declared_tool_names(), names
it always lists, and AgentDefinition checks those when it is built; two
McpToolSources with the resource and prompt tools are refused there. A
source may also define cached_tools(), its last listing without I/O, which
the synchronous estimate_input_tokens counts; otherwise it counts the
runner's last listing of that source.
Model clients
class ModelClient(Protocol):
async def call_model(self, request: ModelRequest, output: ModelOutputWriter) -> ModelResponse: ...
def estimate_input_tokens(self, request: ModelRequest) -> int: ...
Every client makes one request per call with no retries of its own, sends the
whole input every time with no provider-side conversation state, and returns
each output item provider-exact. ModelOutputItem.item is the provider's JSON
text: opaque fields such as encrypted reasoning content and thinking
signatures are kept byte for byte and replayed unchanged to the same provider
and API. An item from another provider or API replays as its text and tool
calls, and its reasoning is dropped. User text and tool results are stored in
a provider-neutral form.
LiteLLMModelClient(model, api=None, api_base=None, api_key_provider=None, request_options=None) uses the Responses API for OpenAI models, with
store=False and include=["reasoning.encrypted_content"], and streaming
Chat Completions for every other provider, keeping Anthropic thinking_blocks
with their signature. It streams text and reasoning summaries to
text_delta and reasoning_delta events and reports usage. Credentials come
from the Worker's environment, as LiteLLM reads them, or from
api_key_provider; never from Flow input or Attributes.
ScriptedModelClient(respond) from dex_agents.stepwise answers each call
from a function of the ModelRequest, so an agent runs without a model key: a
local demo mode, a tutorial, or a test. respond returns a
ScriptedModelReply with text, tool_calls (ScriptedToolCall), and
reasoning, sync or async. ScriptedModelClient.user_texts(request) and
ScriptedModelClient.tool_results(request) read the request. requests keeps
the latest max_recorded_requests requests, 100 by default. Keep the script
deterministic, because a crashed RunModel reruns it. dex_agents.testing
exports the same class.
from dex_agents.stepwise import ModelRequest, ScriptedModelClient, ScriptedModelReply
def reply_as_demo_model(request: ModelRequest) -> ScriptedModelReply:
results = ScriptedModelClient.tool_results(request)
if results:
return ScriptedModelReply(text="; ".join(result.output for result in results))
return ScriptedModelReply(text=f"Demo reply: {ScriptedModelClient.user_texts(request)[-1]}")
model = ScriptedModelClient(reply_as_demo_model) if demo else LiteLLMModelClient(model_name)
Built-in tools: questions and durable waits
Built-in tools are session features, so they work under every runner. Enable
them in AgentSessionOptions.built_in_tools. The runner offers the enabled
definitions from BUILT_IN_TOOL_DEFINITIONS (filter by
turn.enabled_built_in_tools) to its model, and RunModel routes their calls
to their own Steps with no approval and the read effect. write_todos is
described in Plans and plan mode. An enabled name is
reserved; a disabled one is an ordinary tool name. Invalid arguments record
failed with invalid <tool> arguments: <reason> at once. null for an
optional argument means the argument is absent, because providers in
strict-schema mode send it.
ask_user{question, choices?, allows_free_text?}raises an interrupt of kindquestionwith the tool call ID as its interrupt ID and waits inAwaitQuestionAnswer.answer_question(interrupt_id, answer)answers it; the model sees{"status":"answered","answer":"<text>"}as the tool result and continues the same turn. A second answer, a blank answer, or an answer outsidechoiceswhen free text is not allowed returnsFalse.resolve_interrupt(..., approved=False)declines it (The user declined to answer.), andinterrupt_deadlineapplies (question deadline passed).durable_wait{duration_seconds, reason}recordswaiting_for_timerwithwait_untiland waits inAwaitTimeron a Timer. No model call, coroutine, or Worker holds the wait. The model then sees{"status":"completed","waited_seconds":<n>,"reason":"<reason>"}. Durations are whole seconds from 1 tomax_durable_wait(365 days by default).
A tool can ask a question too: execute_deferred_tool_call returns
ToolInputRequested(question, choices, allows_free_text, continuation) instead
of a result, only when the call has taken no effect yet. The tool Step stores
the question on its record and moves to AwaitQuestionAnswer, whose Execute
raises the question interrupt <tool_call_id>-input-<n> under the
PendingInterrupts lock, so parallel tool calls can ask at once. If another
call of the turn already uses that ID, for its record or a pending interrupt,
the asking call records failed instead. After the
answer, the session runs the same tool Step type again with DeferredToolExecution.input_answers and
tool_call.input_continuation. A function tool returns it from the function
and reads the answers from ToolCallContext.
Plans and plan mode
The write_todos built-in replaces the session's plan, the AgentPlan
Attribute, as a whole. Its arguments are {todos: [{content, status}]}, with
a status of pending, in_progress, or completed. RunModel records the
call in its own commit, publishes its result, and still moves to ResumeTurn;
no tool Step runs. Each call that succeeds writes revision + 1: no tasks is
cleared, a planning turn writes a draft, every task completed is
completed, and anything else is active. The model sees
{"status":"updated","revision":<n>,"task_count":<k>} or
{"status":"cleared","revision":<n>}, and the session writes plan_updated,
which it also writes when an approval marks a draft active.
A response with more than one write_todos call fails each of them with
write_todos was called more than once in one response; invalid arguments fail
with invalid todos: <reason>.
Plan mode reviews a plan before it runs. Enable it with
AgentSessionOptions(plan_mode_enabled=True), which needs write_todos and a
runner that is a PlanModeRunner (supports_plan_mode is True). The
Stepwise runner is one; AgentDefinition(plan_mode_enabled=True) sets the
option. AgentDefinition.planning_tool_choice is "auto" by default: a
planning turn offers only write_todos, the plan note tells the model to write
the plan with it, and a turn that answers in text, such as with a clarifying
question, ends without a draft. "required" forces write_todos on the first
model call of a planning turn while no draft exists. Claude Opus 5.5, Claude
Sonnet 5.5, and Anthropic extended thinking refuse a forced tool choice, so
AgentDefinition raises ValueError for "required" when the model client's
supports_forced_tool_choice(settings) returns False, as
LiteLLMModelClient's does for them. For a model LiteLLM routes to Anthropic,
or to Claude on Bedrock or Vertex AI, it counts a thinking option that is not
disabled, or a reasoning_effort other than "none", as extended thinking.
Gemini and OpenAI models accept a forced tool choice with thinking or reasoning.
sessions = AgentSessionClient(client, flow)
await sessions.send_user_message("session-42", "Plan the migration", plan_mode=True)
snapshot = await sessions.get_agent_snapshot("session-42")
if snapshot.plan is not None and snapshot.plan.status is PlanStatus.DRAFT:
await sessions.execute_plan("session-42", snapshot.plan.revision)
- A message with
plan_mode, or any message while adraftplan exists, starts a turn whoseAgentState.interaction_modeisplanning. The runner reads it fromAgentTurnContext.interaction_modeand offers onlywrite_todos.RunModelrejects any other call of a planning turn withThe tool call was not run: a planning turn can only call write_todos.. Feedback on a draft is an ordinary message that revises it. send_user_messagereturnsFalsefor aplan_modemessage when plan mode is off.execute_plan(ExecutePlanRequest(revision))returnsTrueonly while the session waits for a user message, the plan has that revision and isdraftoractive, no execution of it is pending, and nothing is queued, steered, or pending. It records the approval inPendingPlanExecutionand publishes it to thePlanExecutionsChannelMap instance of that revision. It locksPendingPlanExecution, so a concurrent call getsRpcLockConflictError(AgentSessionClientretries it) and thenFalse.AwaitUserMessagewaits on the approval of the current revision only, after queued messages, so an approval of a revision that a later turn replaced never runs. It marks the planactive, writesplan_execution_started, and starts anexecutingturn with every tool, whose user message is[Dex] The user approved plan revision <n>. Execute it and keep the task statuses current with write_todos..build_plan_note(plan, interaction_mode)returns the note that shows the model the plan. Stepwise adds it as user text before the turn's items in every model call, so the model always sees the latest plan.
Steering and turn cancellation
A user message is queued or steered. send_user_message queues it: it waits
until the running turn ends, and the user can still delete or steer it.
steer_user_message steers the running turn with new text or with a queued
message, and cancel_turn stops it with no new turn.
sessions = AgentSessionClient(client, flow)
await sessions.send_user_message("session-42", "Also check order 8")
snapshot = await sessions.get_agent_snapshot("session-42")
queued = snapshot.queued_messages[0]
await sessions.steer_user_message("session-42", message_id=queued.message_id)
await sessions.steer_user_message("session-42", text="Use the staging account instead")
await sessions.delete_queued_message("session-42", snapshot.queued_messages[1].message_id)
await sessions.cancel_turn("session-42", snapshot.state.active_turn_id)
A steered turn ends at its next safe Step boundary. Steering never interrupts
a model call or a tool call that has started: a running model call finishes,
and a final answer it returns completes the turn as usual; a tool that has
started runs to the end, and so does a tool task it started, questions
included. cancel_turn stops the turn now instead: it cancels a running model
call or compaction, whose answer is dropped, and every wait for an approval, an
answer, a durable wait's Timer, or a tool task's next poll. A tool call Step
that has started still runs to the end, and the turn ends after it. A started
tool task is abandoned: its call records outcome_unknown with
cancellation_reason turn_cancelled, and a runner that implements
ToolTaskCanceller is asked to cancel the task, best-effort; a result the
task returns later is dropped. A call that has not started records
rejected with cancellation_reason steered or turn_cancelled, and the
model later sees
{"status":"cancelled","reason":"superseded_by_steered_user_message"} or
{"status":"cancelled","reason":"turn_cancelled_by_user"} as its result. That
includes approvals, questions, and durable waits that were still waiting. A
call that is settled first keeps its own result: an ask_user answer that
commits first, a durable wait whose Timer fires first, and a built-in call with
invalid arguments, which fails at once. A UI can briefly show an approval or a
question of a turn that is already steered; answering it then returns False
or has no effect. ApplyTurnControl then ends the turn. When the runner
implements build_unfinished_turn_messages, it stores the turn's exact
transcript; otherwise it stores a note that lists the calls that ran, as after
a failed turn. A steered turn is followed by a new turn whose user messages are
every steered message, in order. That turn is a planning turn when any of
its messages has plan_mode or a draft plan exists, as for a queued message;
a queued plan_mode message moved by its ID keeps plan_mode. A write_todos
call settles in RunModel's own commit, like a built-in call with invalid
arguments, so one that a model call made while the user steered still replaces
the plan.
While no turn is active, there is nothing to steer: steer_user_message with
text queues the message like send_user_message, so it starts the next turn
at once.
steer_user_messageanddelete_queued_messagetake the Channel message ID fromAgentSnapshot.queued_messages. Both returnFalsefor a message that is no longer queued. When the session consumes the message while the RPC runs, the RPC raisesChannelMessageNotFoundErrorand changes nothing; reload the snapshot and let the user choose again.cancel_turnreturnsTruewhen the request is recorded. The turn may still complete first, when its model call committed the final answer before the RPC committed. A second call for the same turn returnsTrueand changes nothing.steer_user_messageanddelete_queued_messagelock nothing.cancel_turnlocksTurnCancellationRequestandPendingInterrupts, which only short Steps lock, so concurrent cancels apply in order and none of them waits for a model call;AgentSessionClient.cancel_turnretries the rareRpcLockConflictError.resolve_interruptoranswer_questionracing a steer either commits first or returnsFalse. An approval or a tool's answer that commits first moves its call on, and the tool Step then records it cancelled before the tool starts; anask_useranswer that commits first is the call's result.AgentSnapshot.steered_messageslists the steered messages not applied yet.- A turn that a Worker without steering started or advanced cannot be steered
or cancelled:
AgentState.accepts_turn_controlisFalse, and both RPCs returnFalse. It runs to its end as before. Call the RPCs only after every Worker of the Flow type runs this release and every turn that was active during the rollout has ended, because such a turn can hold a wait that an earlier Worker began, which a steer never wakes.
Message storage and retention
Upgrade rule: stop every 0.1.0 Worker of a Flow type before any 0.2.0 Worker of it starts. Once a 0.2.0 Worker stores a message in a 0.1.0 session, it moves the older messages out of
AgentMessages. A 0.1.0 Worker that then runs a Step of the session fails the turn, because it cannot read the moved messages.
The newest messages are the window: one AgentMessages instance per
message, so storing a message writes only that message. When the window holds
more than message_window_size messages (64), its oldest messages move into
immutable AgentMessageChunks instances in the same commit. A chunk holds at
most message_chunk_size messages (32) and message_chunk_max_bytes of
encoded JSON (1 MiB, at most 8 MiB), so it stays well under the Server's
values.maxValueBytes. A larger message skips the window and goes straight
into its own chunk in the Step that stores it, so a message too large for the
Server fails that Step's turn at once instead of a later rollover. One commit
moves at most 4 chunks and 8 MiB, so the window can briefly hold more than its
size. The AgentMessageIndex Attribute lists each kept chunk's first and last
sequence, where the window starts, and the oldest stored message; it changes
only when chunks are written or deleted. TurnReplySequence records where the
latest turn's replies start. Every Step that stores messages loads the whole window.
With compaction, RunModel loads only the chunks after the summary, through
per-movement load options. Without compaction, the model input is every
message, so it loads every chunk.
With compaction, retention keeps Flow state bounded however long the session
runs. It deletes the oldest chunk while every message in it is at or before
ContextSummary.last_sequence and at least retained_message_limit messages
(1,000) are stored after it, so the newest retained_message_limit messages
and every message after the summary always stay. A deleted message is gone
from list_agent_messages and get_agent_snapshot; paging ends at
AgentMessageIndex.first_sequence, and every page's
deleted_before_sequence says where the deleted messages end. An application
that needs the whole transcript copies new messages out with
list_agent_messages before retained_message_limit more are stored after
them, such as after each turn_completed, or sets
retained_message_limit=None. A runner without ContextCompactionRunner on
a session with deleted messages logs a warning on each model call, because
its input starts after them with no summary.
from dataclasses import replace
options = replace(
stepwise_agent_session_options(definition),
message_window_size=64,
message_chunk_size=32,
message_chunk_max_bytes=1024 * 1024,
retained_message_limit=5_000,
)
A session that dex-agents 0.1.0 stored has every message in AgentMessages
and no index. This release reads it as one large window, and each Step that
stores messages moves at most 4 chunks, so the older messages leave the
window over several commits.
Compaction
Compaction keeps the model input under the context window. Set
AgentSessionOptions.compaction (or AgentDefinition.compaction for
Stepwise) to a CompactionPolicy:
from dex_agents import AgentSessionOptions, CompactionPolicy
options = AgentSessionOptions(
compaction=CompactionPolicy(context_window_tokens=400_000), # trigger 0.85, keep 0.3
)
Before each model call, RunModel asks the runner for
estimate_input_tokens(turn). At trigger_ratio of the window it asks for
select_compaction_cutoff(turn, retained_token_budget), the last message to
summarize; when that sequence is past the current summary, RunModel sets
status compacting_context and moves to CompactContext without a model
call. CompactContext has RunModel's timeouts, retries, and failure route.
It calls summarize_context(ContextSummaryRequest) once, stores the
ContextSummary Attribute (generation, first_sequence 1, last_sequence,
text, created_at), records the call's usage, writes compaction_started and
compaction_finished, and moves back to RunModel. A crash reruns only the
summary call. From then on the runner receives the messages after the summary
in turn.messages and the summary in turn.context_summary. A second estimate
that is still high finds no new cutoff, so the pair cannot loop.
Summarized messages stay stored until retention deletes them; see
Message storage and retention. The runner must implement
ContextCompactionRunner, and AgentSessionFlow raises TypeError for a
compaction policy otherwise. StepwiseAgentRunner implements it: its input
starts with CONTEXT_SUMMARY_NOTE_HEADER
([Dex] Summary of the earlier conversation:) and the summary, it cuts only
between model responses whose tool calls all have stored results, so a turn
with many tool calls compacts mid-turn without separating a call from its
result, and it summarizes with CONTEXT_SUMMARY_INSTRUCTIONS and no tools.
ContextSummaryRequest.input_token_budget carries the policy's
context_window_tokens. When the summary request plus the definition's
max_output_tokens, if set, would exceed it, the runner leaves out the oldest
messages, up to a cut that keeps each model response and each tool call with
its result whole, and says so with
build_context_summary_omission_note(omitted_count)
([Dex] <n> earlier messages did not fit in this request and are left out.).
Each CompactContext attempt emits a chat span for the summary call, as
RunModel does; return ContextSummaryResult.model_call_metadata to name it.
A framework adapter cannot measure or rebuild the framework's own request, so
the core gives adapters one shared rule. estimate_stored_input_tokens(turn, fixed_input_tokens=...) counts each stored framework_item with
estimate_text_tokens, three UTF-8 bytes per token, which errs high for
English text. select_turn_boundary_cutoff(turn, retained_token_budget) cuts
only between turns, never inside the running turn, and returns None while
turn.framework_run_state is set, because a resumed call continues from that
state. build_context_summary_prompt(request) writes the range as one plain
text prompt, starting with CONTEXT_SUMMARY_TRANSCRIPT_HEADER, that fits
input_token_budget. CompactionPolicy.reserved_input_tokens adds what an
adapter's estimate cannot see, such as a framework's system prompt. The
adapters of this repository use them; see each adapter's README.
dex_agents.testing.ContextWindowSessionScenario runs the same session past a
small context window for any adapter.
Session settings
Session settings let one Flow type serve sessions that differ, such as by model, without a Worker pool per model. They are one JSON object that the application sets when it starts the session and can replace later:
models = {name: LiteLLMModelClient(name) for name in ("openai/gpt-5-mini", "openai/gpt-5")}
definition = AgentDefinition(
name="support",
instructions="You help customers with their orders.",
model=models["openai/gpt-5-mini"],
# Choose only from clients the Worker built; the settings come from the application.
resolve_model=lambda settings: models.get(
str(settings.get("model")), models["openai/gpt-5-mini"]
),
)
await sessions.start_session("session-42", "Hello", session_settings={"model": "openai/gpt-5"})
await sessions.update_session_settings("session-42", {"model": "openai/gpt-5-mini"})
StartAgentSessionstoresAgentSessionStartInput.session_settingsin theSessionSettingsAttribute.update_session_settingsreplaces it; nothing is merged, and{}clears it. It returnsFalsebefore the session initialized.- A change takes effect when the next turn starts, the same boundary as
plan_mode. The Step that starts a turn copies the settings intoTurnSessionSettings, tagged with the turn, and the turn's model calls, retries, and compaction read only that copy, so a running turn keeps the settings it started with, even across a Worker crash. A copy that an earlier turn left reads as no settings. - A runner reads
AgentTurnContext.session_settings, an empty mapping when there are none, such as in a session started on 0.1.0.ContextSummaryRequest.session_settingscarries the same value. Stepwise'sAgentDefinition.resolve_modelmaps it to theModelClientof each model call, including summary calls and token estimates, andInstructionContext.session_settingspasses it to an instructions function. get_agent_snapshotreturns the next turn's settings asAgentSnapshot.session_settings, not the running turn's: right after an update, the running turn may still use the earlier settings.- Only model calls see the settings. Tool calls do not receive them, and the
compaction policy's
context_window_tokensis one value for every resolved model. - Never put a secret, or anything a browser must not see, in settings.
They are Flow state that
get_agent_snapshotreturns, anddex_agents.ui_stream's browser snapshot JSON includes them as stored. Put a reference, such as a model or credential name, and let the Worker resolve it. An exception thatresolve_modelraises becomes the turn'slast_turn_error, which snapshots also show, so keep the settings out of its message. - The compact JSON text, UTF-8 with non-ASCII unescaped, is at most
MAX_SESSION_SETTINGS_BYTES(16 KiB).validate_session_settingsandsession_settings_sizecheck it, andAgentSessionClientchecks before it sends. The cores write some floats and U+2028/U+2029 differently, so the session's own checks acceptSESSION_SETTINGS_SIZE_TOLERANCE_BYTES(1 KiB) more:update_session_settingsreturnsFalsebeyond that, or forNone, andStartAgentSessionfails the session at once, with no retry, with a reason that starts withINVALID_START_SESSION_SETTINGS_REASON. - Start sessions with settings, and call
update_session_settings, only after every Worker of the Flow type runs 0.2.0. A 0.1.0 Worker drops the start settings and starts turns without them.
The runner contract
An adapter implements three methods. The session calls them from Steps, so each call may run again after a crash or retry. Keep the runner stateless between calls and derive everything from the arguments.
class AgentFrameworkRunner(Protocol):
def build_user_message(self, text: str) -> FrameworkMessage: ...
async def run_model_step(self, turn: AgentTurnContext) -> AgentTurnOutcome: ...
async def execute_deferred_tool_call(self, call: DeferredToolExecution) -> DeferredToolOutcome: ...
class UnfinishedTurnMessageBuilder(Protocol): # optional
def build_unfinished_turn_messages(self, turn: AgentTurnContext) -> Sequence[FrameworkMessage]: ...
DeferredToolOutcome is ToolExecutionResult | ToolInputRequested.
AgentFrameworkRunner is generic in what execute_deferred_tool_call returns,
and a bare AgentFrameworkRunner means AgentFrameworkRunner[ToolExecutionResult],
as in 0.1.0. A runner whose tools ask the user implements
AgentFrameworkRunner[DeferredToolOutcome].
AgentTurnOutcome is TurnCompleted | ToolCallsRequested.
build_user_message(text)
Return FrameworkMessage(role=MessageRole.USER, framework_item=..., text=text).
The session stores it as the turn's first message before the first
RunModel. RecordTurnFailure also calls it for the turn-failure note. Any
other role fails the Step.
run_model_step(turn)
The framework makes at most one model call per run_model_step, and
stops before it runs any tool it can defer. AgentTurnContext carries
committed state:
| Field | Meaning |
|---|---|
flow_id, turn_id, attempt |
Session ID, "turn-<n>", and the one-based RunModel attempt. |
messages |
Every stored AgentMessage in sequence order. The last user message belongs to this turn. |
framework_run_state |
None for the turn's first model call; otherwise the value from this turn's last ToolCallsRequested. is_resuming tests it. |
tool_calls |
Every AgentToolCall record of this turn by tool call ID. When RunModel runs, every record is final. recorded_tool_call(id) reads one. |
resolve_tool_policy(name, framework_effect=None) |
The policy for a tool: explicit policy, else framework metadata, else default_tool_effect. |
events |
AgentEventWriter bound to the turn, Step execution, and attempt: write_text_delta, write_reasoning_delta, write_tool_call_started, write_tool_call_finished, write_event. Safe to call from any thread; writes after the call returns are dropped. |
heartbeat() |
Records liveness. The session already heartbeats every 20 seconds in the background. |
enabled_built_in_tools |
The built-in tools to offer to the model, from AgentSessionOptions.built_in_tools. |
A runner must:
- Without run state, rebuild framework input from each message's
framework_item. With run state, restore the framework from it and give every tool call the result from its final record instead of running the tool.describe_tool_call_outcome(record)renders the model-visible text. A framework with a native rejection API may passrecord.rejection_messagethere instead. - Let the framework make one model call.
- If the response asks for tools, return
ToolCallsRequested(framework_run_state, tool_calls, usage)with oneDeferredToolCall(tool_call_id, tool_name, arguments, policy)per call, in model order. Get eachpolicyfromturn.resolve_tool_policy(name, framework_effect); passframework_effectwhen the framework exposes metadata, such as MCPreadOnlyHint(READ) oridempotentHint(IDEMPOTENT_WRITE).framework_run_statemust be non-empty and must continue the turn after the tools. Tool call IDs must be non-empty, contain no/, and be unique within one turn; IDs may repeat in later turns, because the idempotency key includes the turn ID. - If the response is the final answer, return
TurnCompleted(new_messages, usage).new_messagesholds the turn's items the session has not stored yet, in order: model output, tool requests, tool results, and the final answer. The session appends them in one commit. A runner may instead store items as they appear withToolCallsRequested.new_messages, which the session appends in the commit that records the tool calls; such a runner implementsbuild_unfinished_turn_messages, whichRecordTurnFailurecalls to store the results of the turn's tool calls before its note. If it raises or returns anything butFrameworkMessagevalues, the session logs a warning and stores only the note. A runner that setsnew_messageswithout the method fails its turn after that one model call, with no retry. - Report
usagefor this model call only. The session adds it to the session total. Optionally returnmodel_call_metadata=ModelCallMetadata( request_model, response_model, provider_name, response_id, finish_reasons)in either outcome. The session never stores it; it names thechatspan (see Tracing). Leave a fieldNonewhen the framework does not expose it. - Never run a tool inside
run_model_step, unless the framework cannot defer it. Document such tools as at-least-once: aRunModelretry runs them again.
Any exception fails the attempt. RunModel retries up to three attempts
within 30 minutes, rerunning only this one model call, then routes to
RecordTurnFailure. Each attempt has a 15 minute timeout. run_model_limit in
AgentSessionOptions changes it, with the attempts and the retry budget.
When a failed turn already has final tool call records, RecordTurnFailure
stores one note before it deletes them, built with build_user_message, so
the next turn's model does not repeat a write. build_turn_failure_note
renders it, one line per record in deferral_order:
[Dex] The previous turn failed after these tool calls ran:
- <tool_name> <arguments> -> <status>: <describe_tool_call_outcome(record)>
Do not repeat these tool calls unless the user asks again.
execute_deferred_tool_call(call)
DeferredToolExecution carries flow_id, the approved tool_call record,
idempotency_key ("<flow_id>/<turn_id>/<tool_call_id>", the same on every
attempt), attempt, events, heartbeat, heartbeat_checkpoint,
save_heartbeat_checkpoint, and input_answers. Run the real tool once and
return ToolExecutionResult(output), ToolInputRequested when the tool
needs an answer before it takes any effect, or ToolTaskStarted when the tool
started remote work that the session polls (see Tool tasks).
Tool calls of one model response run in parallel Steps, so this method may run
concurrently with itself.
save_heartbeat_checkpoint(text) records a heartbeat that carries text, and
every later heartbeat of the attempt, the session's background heartbeats
included, carries it too. A retried attempt reads it as heartbeat_checkpoint
and can resume work the earlier attempt started, such as polling a job,
instead of starting it again. A DeferredToolCall may also carry
tool_source_reference, opaque text stored on the call's AgentToolCall
record (wire field tool_source_reference), so the tool's source can resolve
the tool in the tool Step without listing it again.
Raise ToolExecutionError(message) for a failure the model should see; the
session records failed with that message and does not retry. Any other
exception is a Step failure. read and idempotent_write calls retry up to
three attempts, and then RecordToolFailure records failed for a read and
outcome_unknown for a write, because the write may have taken effect.
Single-attempt calls do not retry; RecordToolOutcomeUnknown records
outcome_unknown. Both read the recovery error's attempt_progress. When it
is NOT_STARTED, no attempt reached a Worker and the tool never ran, so both
record failed with an error that starts with
TOOL_CALL_NOT_RUN_ERROR_PREFIX (the tool did not run, so it is safe to call again), then ": " and the recovery detail. The model may call the tool
again. A call whose tool already asked a question ran before, so it keeps the
status of its effect.
Before calling the runner, the tool Step writes tool_call_started and awaits
a heartbeat. A Worker that disconnects after it received the task fails the
attempt, and the call records outcome_unknown. The heartbeat matters because
Dex treats a stream that half-closes before any output as one whose task never
ran: the attempt returns to pending with the same attempt number, uses no
retry, and is handed out again. After output, a half-close is a disconnect
that fails the attempt with NOT_COMPLETED. So a tool that started is never
returned as not run at the same attempt number, and never reported later as
NOT_STARTED.
Tool tasks
Some tools start remote work and get a handle back, such as an MCP task. A
tool Step must not hold a connection while that work runs, so
execute_deferred_tool_call returns
ToolTaskStarted(task_id, poll_interval_seconds), and the runner implements
ToolTaskPoller.poll_tool_task(poll):
class ReportRunner: # also implements AgentFrameworkRunner[DeferredToolOutcome]
async def execute_deferred_tool_call(self, call):
saved = call.heartbeat_checkpoint
if saved is not None: # an earlier attempt started the job
return ToolTaskStarted(task_id=saved, poll_interval_seconds=5)
job_id = await reports.start(call.idempotency_key)
await call.save_heartbeat_checkpoint(job_id)
return ToolTaskStarted(task_id=job_id, poll_interval_seconds=5)
async def poll_tool_task(self, poll: ToolTaskPoll) -> ToolTaskPollOutcome:
job = await reports.get(poll.tool_call.task_id)
answered = poll.tool_call.input_continuation # ID of the question answered last
resolved = poll.pending_answer # its resolution, on the poll right after it
if resolved is not None and job.question_id == answered:
# Keyed by the question ID, so a retried poll never answers a later question.
if resolved.resolution is QuestionResolution.ANSWERED:
job = await reports.answer(job.id, answered, resolved.answer)
else: # DECLINED or DEADLINE_PASSED
job = await reports.skip(job.id, answered)
if job.question_id is not None and job.question_id != answered:
return ToolInputRequested(question=job.question, choices=job.choices,
continuation=job.question_id)
if not job.done:
# Keep the answered ID only while the job still shows that question.
kept = answered if job.question_id == answered else None
return ToolTaskPending(poll_interval_seconds=10, continuation=kept)
return ToolExecutionResult(output=job.summary)
The tool Step records waiting_for_task with task_id and
task_poll_interval_seconds, writes tool_task_started, and ends.
PollToolTask waits on a Timer for the interval, rounded up to whole seconds
from 1 to a day, and calls poll_tool_task once per execution:
ToolExecutionResultrecordssucceeded, andToolExecutionErrorrecordsfailed.ToolTaskPending(poll_interval_seconds, continuation=None)stores the next interval, storescontinuationasinput_continuation, and polls again;Noneclears it.ToolInputRequestedasks the user throughAwaitQuestionAnswer, with the interrupt ID<tool_call_id>-input-<n>. The task has started, so the question never ends the call: an answer, a decline, and a passed deadline each go back to the task as aQuestionAnswerwhoseresolutionisANSWERED,DECLINED, orDEADLINE_PASSED.PollToolTaskpolls at once and hands it over aspoll.pending_answer, only on that first poll; a retried attempt of it gets it again. That poll reads the question's continuation intool_call.input_continuation, and later polls read what the lastToolTaskPendingreturned. Keep the answered question's continuation only while the task may still show that question, so that a later identical question reads as new. Deliver idempotently, keyed by a question ID when the task has one. When a retried attempt cannot tell whether an earlier one delivered, return the question again instead: the user answers twice rather than a stale answer reaching a later question. When the task cannot take a decline, cancel it best-effort and raise an exception other thanToolExecutionError; the call is then recorded by its effect.- Any other exception fails the poll attempt. A poll gets three attempts; then
RecordToolFailurerecordsfailedfor areadcall andoutcome_unknownfor a write, andRecordToolOutcomeUnknownrecordsoutcome_unknownfor a single-attempt call, because the task may still run.
No Step holds a connection while the work runs, and a Worker restart only
delays the next poll. A crash during a poll repeats only that poll, so a poll
must only read the task, or deliver an answer idempotently. A CallTool
attempt that dies after the remote work started runs again: save the handle
with save_heartbeat_checkpoint before you return it, and return the saved
handle when heartbeat_checkpoint holds one. Dex may drop a heartbeat
checkpoint, so an attempt that dies before its checkpoint is saved and before
the tool Step commits can still start the work twice; pass the idempotency key
on so the remote system can drop the duplicate. PollToolTask owns the record
until it is final and takes no lock. Steering waits for a started task, its
questions included, and a call with task_id is never recorded rejected.
cancel_turn abandons the task instead, so Stop never waits for it: it
cancels PollToolTask, drops whatever a running poll returns, and records the
call outcome_unknown with cancellation_reason turn_cancelled in
CancelToolTask. Implement ToolTaskCanceller.cancel_tool_task to stop the
remote work best-effort; make it idempotent by task_id, because a crash or a
failed attempt calls it again. When it keeps failing, the call is still
recorded and the turn ends:
async def cancel_tool_task(self, cancellation: ToolTaskCancellation) -> None:
await reports.cancel(cancellation.tool_call.task_id)
DeferredToolExecution.supports_tool_tasks tells a tool whether the runner polls tasks. A runner without poll_tool_task
that returns ToolTaskStarted anyway gets failed for a read call and
outcome_unknown otherwise, with MISSING_TOOL_TASK_POLLER.
A long task polls many times, and each poll is a Step in the session's history, so poll at the slowest interval the task allows.
Effects and policies
| Effect | Step | Retry | Crash mid-call | Default approval |
|---|---|---|---|---|
read |
CallTool |
Up to 3 attempts; then failed |
Runs again | No |
idempotent_write |
CallTool |
Up to 3 attempts, key <flowId>/<turnId>/<toolCallId>; then outcome_unknown |
Runs again with the same key | No |
non_idempotent_write |
CallSingleAttemptTool |
One attempt; then outcome_unknown |
Outcome unknown | Yes |
unclassified |
CallSingleAttemptTool |
One attempt; then outcome_unknown |
Outcome unknown | Yes |
ToolPolicy(effect, requires_approval=None) overrides the defaults per tool
name; None uses the effect's default approval, which is_approval_required
reports. AgentSessionOptions also sets default_tool_effect (default
unclassified), the AgentEvents Stream capacity (default 10 MiB), and an
optional interrupt_deadline, built_in_tools, max_durable_wait, and
plan_mode_enabled. A tool
whose policy requires approval runs only after AwaitToolApproval approves it.
Finding sessions that need a person
AgentSessionClient.start_session makes a session searchable by the Steps it
waits in, so one query finds every session waiting for a person, across agents:
await sessions.start_session("session-42", "Refund order 7")
waiting = await client.search_flows('ActiveStepTypes = "AwaitToolApproval"', 50)
AwaitQuestionAnswer finds questions the same way. A session leaves the search
when it moves on. Set active_step_search_mode in the start options to choose
another mode, such as ActiveStepSearchMode.DISABLED. AsyncClient.start_flow
does not turn this on; pass AGENT_SESSION_ACTIVE_STEP_SEARCH_MODE in its
config_override, or call update_flow_config for a running session. See
Find sessions that need a person.
Spend limits
ModelSpendLimit is an application's spend limit, usually per account or API key
across many sessions. The session asks it before each model call and reports what
the call used. Counters and limits belong to the application. Without one, the
session makes no checks. See the interface's documentation and
Spend limits.
Flow shape
StartAgentSession -> AwaitUserMessage -> RunModel
AwaitUserMessage -> steer -> ApplyTurnControl
RunModel -> steered or cancelled -> ApplyTurnControl
cancel_turn -> cancels RunModel, CompactContext, and the waiting Steps -> FinishTurnCancellation(turn_id)
FinishTurnCancellation -> ApplyTurnControl, or CancelToolTask(id) per abandoned task
RunModel -> TurnCompleted -> AwaitUserMessage
RunModel -> CompactContext(through_sequence) -> RunModel with a compaction policy
RunModel -> ToolCallsRequested -> go_to_many(
AwaitToolApproval(id) -> CallTool(id) or CallSingleAttemptTool(id) per call needing approval
CallTool(id) or CallSingleAttemptTool(id) per other call
AwaitQuestionAnswer(id) per ask_user call
AwaitTimer(id) per durable_wait call
ResumeTurn(expected_result_count))
CallTool(id) -> ToolInputRequested -> AwaitQuestionAnswer(id) -> CallTool(id)
CallTool(id) -> ToolTaskStarted -> PollToolTask(id) -> PollToolTask(id) ...
PollToolTask(id) -> ToolInputRequested -> AwaitQuestionAnswer(id) -> PollToolTask(id)
ResumeTurn -> RunModel, or ApplyTurnControl when steered or cancelled
ApplyTurnControl -> RunModel with steered messages, else AwaitUserMessage
| Step (input) | What it does | Next |
|---|---|---|
StartAgentSession (AgentSessionStartInput) |
Start Step. Initializes state, usage, interrupts, and SessionSettings. |
RunModel with a non-blank initial_message, else AwaitUserMessage |
AwaitUserMessage (none) |
Waits for any of, in this order, TurnControlRequests, QueuedUserMessages.for_one(), and, while the plan is draft or active, PlanExecutions.for_one(<revision>). Locks TurnCancellationRequest in Execute, so no turn starts while cancel_turn decides. A steer: moves to ApplyTurnControl. A message: stores it and starts a chat or planning turn. An approval: clears PendingPlanExecution, marks the plan active, and starts an executing turn. |
ApplyTurnControl, RunModel, or itself for a stale approval |
RunModel (none) |
Ends the turn and moves to ApplyTurnControl without a model call when the turn is steered or cancelled, or else to CompactContext when a compaction policy's threshold is crossed. Otherwise calls run_model_step. 15 min timeout, 2 min heartbeat timeout, 3 attempts in 30 min, failure route RecordTurnFailure. On ToolCallsRequested, stores the run state, one AgentToolCalls record per call, and the interrupts; records write_todos calls, and every call but write_todos of a planning turn, final in this commit. |
AwaitUserMessage, ApplyTurnControl, CompactContext, or go_to_many with one Step per tool call plus ResumeTurn |
CompactContext (CompactContextInput) |
Calls summarize_context once and stores ContextSummary. RunModel's timeouts, attempts, and failure route. |
RunModel |
AwaitToolApproval (tool call ID) |
Waits for ToolCallCancellations.for_one(id) or InterruptResolutions.for_one(id), plus a Timer at the deadline, rounded up to whole seconds, while the interrupt is pending. Approved: marks the record approved. Rejected or expired: records rejected and publishes ToolCallResults. A Timer that fires before the deadline waits again. A cancellation records rejected with its reason. |
CallTool or CallSingleAttemptTool, or ends |
AwaitQuestionAnswer (tool call ID) |
Waits for ToolCallCancellations.for_one(id) or InterruptResolutions.for_one(interrupt_id), plus a deadline Timer while the question is pending, or a zero Timer while a tool's question is not raised yet. Locks PendingInterrupts in Execute, where it raises a tool's question. Records the answer, decline, or deadline and publishes ToolCallResults, or returns a tool question's call to its tool Step, or a task's question to PollToolTask. Steering never cancels a task's question; cancel_turn cancels the wait and abandons the task. |
Ends, the tool Step, PollToolTask, FinishTurnCancellation, or itself |
AwaitTimer (tool call ID) |
Waits on a Timer for the whole seconds until wait_until, or ToolCallCancellations.for_one(id). Records succeeded, or rejected when cancelled, and publishes ToolCallResults. |
Ends |
CallTool (tool call ID) |
Runs one read or idempotent_write call, unless the turn is steered or cancelled before tool_call_started. 3 attempts, failure route RecordToolFailure. Records succeeded, or failed for ToolExecutionError, and publishes ToolCallResults. A ToolInputRequested result raises a question instead, and a ToolTaskStarted result records waiting_for_task and writes tool_task_started. |
Ends, AwaitQuestionAnswer, or PollToolTask |
CallSingleAttemptTool (tool call ID) |
Runs one non_idempotent_write or unclassified call. 1 attempt, failure route RecordToolOutcomeUnknown. |
Ends, AwaitQuestionAnswer, or PollToolTask |
PollToolTask (tool call ID) |
Waits on a Timer for task_poll_interval_seconds, or a zero Timer while task_answer_pending, then calls poll_tool_task. 3 attempts, failure route RecordToolFailure for read and idempotent_write calls and RecordToolOutcomeUnknown otherwise. No lock: steering waits for the task's result, and cancel_turn cancels the poll and drops its result. Records the result and publishes ToolCallResults, stores the next interval and continuation, or records the task's question. |
Ends, itself, AwaitQuestionAnswer, or FinishTurnCancellation |
RecordToolFailure (tool call ID) |
Records failed for a read and outcome_unknown for a write, with the recovery detail in error, and publishes ToolCallResults. A call whose tool Step never started (attempt_progress NOT_STARTED) records failed with TOOL_CALL_NOT_RUN_ERROR_PREFIX instead. |
Ends |
RecordToolOutcomeUnknown (tool call ID) |
Records outcome_unknown with the recovery detail in error, and publishes ToolCallResults. A call whose tool Step never started records failed with TOOL_CALL_NOT_RUN_ERROR_PREFIX instead. |
Ends |
ResumeTurn (expected result count) |
Waits for ToolCallResults.for_n(count) or TurnControlRequests. Locks PendingInterrupts in Execute. A steer wake cancels the pending interrupts of the turn's approvals and of questions no task asked, and wakes each such approval, question, and durable wait through ToolCallCancellations, then waits again. |
RunModel, ApplyTurnControl when steered or cancelled, or itself |
ApplyTurnControl (none) |
Consumes every steered message and stale wake. Ends an active steered or cancelled turn with the runner's unfinished-turn messages or a note, and writes turn_steered or turn_cancelled. |
RunModel with the steered messages, else AwaitUserMessage |
RecordTurnFailure (the failing Step's input, ignored) |
Records last_turn_error. Appends the runner's unfinished-turn messages when it implements build_unfinished_turn_messages. When the turn has final tool call records, appends one turn-failure note. Clears the turn's records and run state. |
AwaitUserMessage |
FinishTurnCancellation (turn ID) |
Locks TurnCancellationRequest and PendingInterrupts. When cancel_turn cancelled the model call or compaction, moves to ApplyTurnControl. While tool calls run, records each call whose wait was cancelled as rejected, hands each started task to CancelToolTask, and ends; ResumeTurn then ends the turn. |
Ends, ApplyTurnControl, CancelToolTask, or a waiting Step whose answer committed first |
CancelToolTask (tool call ID) |
Calls cancel_tool_task when the runner implements ToolTaskCanceller, then records outcome_unknown and publishes ToolCallResults. 3 attempts in 2 min, 30 s timeout, failure route RecordToolOutcomeUnknown. |
Ends |
Every Step sets wait_for and execute durability to SYNC explicitly, so a
FlowConfig(step_durability=ASYNC) never replays a Step result. Each tool
call Step and its failure route receive the tool call ID as input, so parallel
Steps never share state. Every tool call publishes exactly one ToolCallResults
message, when its record becomes final, and a tool Step whose record is already
final ends without running or publishing. ResumeTurn goes back to waiting if
a consumed result left a record of the turn unfinished. AwaitToolApproval
and the tool Steps load only their own AgentToolCalls instance, through
per-movement StepOptions.
Persistence, shared by every agent Flow type and exported from dex_agents:
| Name | Kind | Value |
|---|---|---|
AgentState |
Attribute AGENT_STATE |
AgentState |
AgentMessages |
AttributeMap AGENT_MESSAGES, key message_sequence_key(n) (12-digit zero-padded) |
AgentMessage; only the window, from AgentMessageIndex.window_first_sequence |
AgentMessageChunks |
AttributeMap AGENT_MESSAGE_CHUNKS, key message_sequence_key(first sequence) |
AgentMessageChunk, immutable, consecutive messages listed in AgentMessageIndex.chunks |
AgentMessageIndex |
Attribute AGENT_MESSAGE_INDEX |
AgentMessageIndex; unset until a Step first writes a chunk |
TurnReplySequence |
Attribute TURN_REPLY_SEQUENCE |
TurnReplySequence, written by the Step that starts a turn |
FrameworkRunState |
Attribute FRAMEWORK_RUN_STATE |
string, set only while a turn waits for tool calls |
AgentToolCalls |
AttributeMap AGENT_TOOL_CALLS, key = tool call ID |
AgentToolCall, deleted when the turn ends |
PendingInterrupts |
Attribute PENDING_INTERRUPTS |
list[AgentInterrupt] |
AgentUsage |
Attribute AGENT_USAGE |
AgentUsage |
QueuedUserMessages |
Channel QUEUED_USER_MESSAGES |
AgentUserMessage |
InterruptResolutions |
ChannelMap INTERRUPT_RESOLUTIONS, key = interrupt ID |
AgentInterruptResolution |
ToolCallResults |
Channel TOOL_CALL_RESULTS, internal |
ToolCallCompleted |
SteeredUserMessages |
Channel STEERED_USER_MESSAGES |
AgentUserMessage |
TurnControlRequests |
Channel TURN_CONTROL_REQUESTS, internal |
TurnControlRequest |
TurnCancellationRequest |
Attribute TURN_CANCELLATION_REQUEST |
TurnCancellationRequest, written only by cancel_turn |
ToolCallCancellations |
ChannelMap TOOL_CALL_CANCELLATIONS, internal, key = tool call ID |
ToolCallCancellation |
AgentTraceContext |
Attribute AGENT_TRACE_CONTEXT |
string, the W3C traceparent of the current activation's root span when its context is valid, sampled or not; deleted otherwise |
AgentPlan |
Attribute AGENT_PLAN |
AgentPlan, unset until write_todos first succeeds |
PlanExecutions |
ChannelMap PLAN_EXECUTIONS, key plan_execution_key(revision) (decimal) |
PlanExecutionRequest |
PendingPlanExecution |
Attribute PENDING_PLAN_EXECUTION |
PlanExecutionRequest that execute_plan accepted and AwaitUserMessage has not consumed |
ContextSummary |
Attribute CONTEXT_SUMMARY |
ContextSummary, set once the session compacts |
AgentEvents |
Stream flow.agent_events, one per Flow type |
AgentEvent |
AgentState.status is waiting_for_user_message, running_model,
running_tools, or compacting_context, and AgentState.interaction_mode is chat, planning, or
executing. An AgentToolCall record is awaiting_approval, approved,
awaiting_answer, waiting_for_timer, or waiting_for_task while it waits,
and rejected, succeeded, failed, or outcome_unknown when final; runners
only see final records, and poll_tool_task a waiting_for_task one.
RPCs:
| RPC | Input | Output | Notes |
|---|---|---|---|
send_user_message |
AgentUserMessage |
bool |
Transactional; False for blank text, and for a plan_mode message when plan mode is off. |
execute_plan |
ExecutePlanRequest |
bool |
Locks PendingPlanExecution; True only while the session waits for a user message, the plan has this revision and is draft or active, no execution of it is pending, and nothing is queued, steered, or pending. AgentSessionClient retries lock conflicts. |
steer_user_message |
SteerUserMessageRequest |
bool |
Transactional, locks nothing. Moves a queued message by message_id, or steers with new text; while no turn is active, queues text. False for both or neither, blank text, a message no longer queued, or a turn that does not accept steering; ChannelMessageNotFoundError when it was consumed while the RPC ran. |
delete_queued_message |
DeleteQueuedMessageRequest |
bool |
Transactional, locks nothing. False for a message no longer queued. |
cancel_turn |
CancelTurnRequest |
bool |
Locks TurnCancellationRequest and PendingInterrupts. Cancels the turn's model call, compaction, and waiting Steps, and moves to FinishTurnCancellation. False when no turn is active, turn_id names another turn, or the turn does not accept cancellation. |
resolve_interrupt |
ResolveInterruptRequest |
bool |
Locks PendingInterrupts; False when not pending or past its deadline; removes the interrupt and publishes the answer in one commit. On a question, approved=False declines it and approved=True returns False. A concurrent Step can raise RpcLockConflictError; AgentSessionClient retries it for 5 seconds. |
answer_question |
AnswerQuestionRequest |
bool |
Locks PendingInterrupts; False when the interrupt is not a pending question, is past its deadline, or the answer is blank or not an allowed choice; removes it and publishes the answer in one commit. AgentSessionClient.answer_question retries lock conflicts like resolve_interrupt. |
get_agent_snapshot |
none | AgentSnapshot |
Loads the window and every kept chunk. State, last 200 stored messages, pending interrupts, the active turn's tool calls, usage, the queued message count, the plan, and the queued and steered messages with their IDs. Serves retained closed sessions. |
list_agent_messages |
ListAgentMessagesRequest |
AgentMessagesPage |
Loads the window and every kept chunk. Oldest first; limit clamped to 1..200; the last page starts at AgentMessageIndex.first_sequence, and deleted_before_sequence is set once retention deleted earlier messages. |
Events on AgentEvents (AgentEventType): turn_started {"sequence"},
turn_completed {"message_count"}, turn_failed {"error"},
turn_steered {"steered_message_count", "message_count"},
turn_cancelled {"message_count"}, text_delta
and reasoning_delta {"text"}, tool_call_started {"tool_call_id", "tool_name"}, tool_call_finished {"tool_call_id", "tool_name", "status"}
(once per tool call, including rejections), tool_call_deferred
{"tool_call_id", "tool_name", "effect", "requires_approval"},
interrupt_raised {"interrupt_id", "tool_call_id", "tool_name", "deadline", "kind", "question", "choices", "allows_free_text"}, interrupt_resolved
{"interrupt_id", "approved", "rejection_message", "kind", "answered", "cancellation_reason"} (steered or turn_cancelled when steering or
cancellation removed the interrupt), timer_started {"tool_call_id", "wait_until", "reason"}, plan_updated {"revision", "status", "task_count"} (after write_todos, or when an approval marks a draft
active), plan_execution_started {"revision"}, tool_task_started
{"tool_call_id", "tool_name", "task_id"}, usage_recorded
{"recorded", "session"}, compaction_started {"through_sequence"}, and
compaction_finished {"generation", "through_sequence"}.
Every event carries turn_id, step_execution_id (shared by all attempts of
one Step execution), and the Step attempt. A UI groups live text by
(turn_id, step_execution_id, attempt): within one step_execution_id it
keeps only the highest attempt seen and drops the partial text of lower
attempts. Different step_execution_id values of a turn are separate model
call and tool call Steps, and each keeps its text. After turn_completed,
turn_failed, turn_steered, or turn_cancelled, replace the turn's live
text with the stored messages. Streams
are best effort; rebuild from get_agent_snapshot.
MCP tools (dex_agents.mcp)
dex_agents.mcp turns the tools of MCP servers into tools of an agent
session. Every MCP tool call runs in its own tool Step, with an effect class
from the tool's annotations, so a crash never repeats a write silently.
Install the MCP SDK with the mcp extra; the core itself never imports it:
pip install "dex-agents[mcp]"
Configure the servers on the Worker, never from Flow input, a user message, or a browser: a stdio server's command runs with the Worker's privileges.
from dex_agents.mcp import McpStdioServerConfig, McpStreamableHttpServerConfig, McpToolSource
tools = McpToolSource([
McpStdioServerConfig(name="files", command="mcp-server-files", args=["/srv/reports"]),
McpStreamableHttpServerConfig(
name="tickets",
url="https://tickets.example.com/mcp",
headers={"Authorization": f"Bearer {os.environ['TICKETS_TOKEN']}"},
),
])
A Stepwise agent takes the source in AgentDefinition.tool_sources; see
Tool sources. Another runner offers the tools and defers
every call; the tool Step runs it:
class MyRunner:
async def run_model_step(self, turn: AgentTurnContext) -> AgentTurnOutcome:
offered = await tools.list_tools() # McpTool: name, description, input_schema
response = await call_my_model(turn.messages, offered)
if not response.tool_calls:
return TurnCompleted(new_messages=..., usage=...)
return ToolCallsRequested(
framework_run_state=serialize(response),
tool_calls=[
tools.deferred_tool_call(
turn, tool_call_id=call.id, tool_name=call.name, arguments=call.arguments
)
for call in response.tool_calls
],
usage=...,
)
async def execute_deferred_tool_call(self, call: DeferredToolExecution) -> DeferredToolOutcome:
handlers = {**self._local_handlers, **await tools.deferred_tool_handlers()}
return await handlers[call.tool_call.tool_name](call)
async def poll_tool_task(self, poll: ToolTaskPoll) -> ToolTaskPollOutcome:
return await tools.poll_tool_task(poll)
list_tools lists each server on its own and caches its tools per Worker.
The first listing of a server is awaited; after listing_ttl (5 minutes) a
server is listed again in the background while its earlier tools are
returned. A server that cannot be listed is left out with a warning log and
is not tried again for listing_failure_backoff (30 seconds), then in the
background, so one bad server never hides the others' tools or slows every
model call. One listing per server runs at a time. list_tools(refresh=True)
lists every server again now. deferred_tool_call stores the
server and the tool's MCP name on the call's record, so the tool Step calls
that server directly, without a listing; call_tool finds the server through
the listing. Tool names are <server>__<tool>, with characters other than
letters, digits, _, and - replaced by _ and names over 64 characters
shortened with a hash. When two tools get the same name, the first in server
order is offered and the other is left out with a warning log. Key
ToolPolicy entries by that name.
Resources and prompts
With resource_and_prompt_tools_enabled=True, list_tools also offers three
READ tools after the servers' tools, when at least one server is
configured. Each takes a configured server's name as server, an enum of
the names, and opens a new connection to that server with the source's
transport, headers, and connection_timeout:
| Tool | Arguments | Output |
|---|---|---|
mcp_list_resources_and_prompts |
server |
{"prompts": [...], "resource_templates": [...], "resources": [...]}, every page |
mcp_read_resource |
server, uri |
{"contents": [...]} |
mcp_get_prompt |
server, name, optional arguments (strings by name) |
{"description": ..., "messages": [...]} |
Outputs are the server's items as it sent them, in JSON with sorted keys, so
both cores show the same text. A server that does not offer resources or
prompts lists none. A connection failure is retried connection_attempts
times. An MCP error answer, a server that stays unreachable, and an
unconfigured server raise ToolExecutionError. Enable the tools on one
source of an agent only. Resource subscriptions and list-change notifications are
not followed. The names are MCP_LIST_RESOURCES_AND_PROMPTS_TOOL_NAME,
MCP_READ_RESOURCE_TOOL_NAME, and MCP_GET_PROMPT_TOOL_NAME; key a
ToolPolicy by them to require approval.
Effects
| MCP annotations | Effect | Tool Step | Approval by default |
|---|---|---|---|
readOnlyHint: true |
read |
CallTool |
No |
idempotentHint: true |
idempotent_write |
CallTool |
No |
| anything else, or none | default_tool_effect, unclassified by default |
CallSingleAttemptTool |
Yes |
idempotentHint makes a tool idempotent_write whatever destructiveHint
says, in both cores. A ToolPolicy in AgentSessionOptions.tool_policies wins
over the annotations. trusts_tool_annotations defaults to True, so a server
with a wrong readOnlyHint on a write gets that write run without approval and
retried up to three times. Set trusts_tool_annotations=False on a server
whose hints you do not trust; its tools then keep the session default unless a
policy names them.
What a call does
Every request opens a new connection: a new process for a stdio server, a new
HTTP connection for a Streamable HTTP server. Connecting and each handshake
request are bounded by connection_timeout (30 seconds by default), and a
tools/call by request_timeout (10 minutes). The source speaks MCP 2026-07-28,
which has no protocol session, and falls back to the initialize handshake for
older servers, so a server restart between listing and calling is harmless.
Every tools/call carries the session's idempotency key,
<flowId>/<turnId>/<toolCallId>, in _meta under
IDEMPOTENCY_KEY_META_KEY (io.superdurable.dex/idempotencyKey). The MCP SDK
adds traceparent to _meta when OpenTelemetry has an active trace.
| What happens | Recorded | Re-issued? |
|---|---|---|
| A result | succeeded: text blocks joined with newlines, other content as JSON |
- |
An isError result, or a JSON-RPC error response other than -32603 |
failed with the text (ToolExecutionError) |
No |
| The call was never sent: its server is not configured, or 3 connection attempts failed | failed |
No |
The stream broke, the server exited, or the request timed out after it was sent; an internal error (-32603), an HTTP error status, or a body that is not JSON-RPC (-32700), which a gateway can return after the server ran the call; a server that keeps asking to repeat the call |
McpToolCallInterruptedError, a Step failure |
read and idempotent_write: CallTool sends a new request, same idempotency key, up to 3 attempts. Other effects: outcome_unknown, never sent again |
| The Worker dies during the call | as for any tool Step | Same as above |
MCP 2026-07-28 removed stream resumption, so a broken stream is never resumed; the effect class alone decides whether the call is sent again.
notifications/progress become mcp.tool_progress events on AgentEvents
({"tool_call_id", "tool_name", "progress", "total", "message"}).
Long-running MCP tasks
The source declares the tasks extension, io.modelcontextprotocol/tasks, on
every request. A server may then answer tools/call with a task handle. MCP
2026-07-28 has no protocol session and lets the server choose the task TTL, so
any connection can poll the task.
A Streamable HTTP server's task becomes a tool task: the tool
Step saves the task ID as its heartbeat checkpoint, returns ToolTaskStarted,
and ends. PollToolTask then calls poll_tool_task on a Timer, which polls
tasks/get once and writes an mcp.task_status event when the status
changed. The next interval is
the server's pollIntervalMs, rounded up to whole seconds, or, when that is
shorter, double the previous interval, up to a minute, so a task that runs for
hours adds a few Steps a minute at most. A completed task is the tool's
result; a failed or cancelled task records failed. No Step holds a
connection while the task runs, a Worker restart only delays the next poll,
and a task can run for days. A CallTool attempt that dies after tools/call
returned the task reads the checkpoint on its retry and returns the same task
without calling the tool again, once the checkpoint reached Dex; a lost
checkpoint leaves only the idempotency key to drop a second task. A runner
without poll_tool_task gets every task polled inside the tool Step, as
below. A server that no longer knows the task, or
cannot be reached for three poll attempts, records failed for a read tool
and outcome_unknown otherwise; the source never calls the tool again from a
poll. cancel_turn abandons the task, and cancel_tool_task sends
tasks/cancel; forward it from your runner's cancel_tool_task, as the
Stepwise runner does. Any answer from the server ends the cancellation, so a
retried one is harmless.
A stdio task lives in its server process, which ends with the request. The
tool Step polls it on the call's own connection, with a heartbeat per poll,
for at most task_wait_timeout (10 minutes by default) per attempt, holding a
Worker slot. When the wait passes, or the Worker dies, CallTool retries a
read or idempotent_write call and the retry calls the tool again; other
effects record outcome_unknown. call_tool, the framework adapter path,
polls every task this way.
Input requests
A task in input_required that asks one form elicitation/create whose
schema has a single string field, with or without enum, or a single
boolean field becomes a session question: poll_tool_task returns
ToolInputRequested with the elicitation's message, the enum values or
yes and no as choices, and the request key and a fingerprint of the
request in its continuation. AwaitQuestionAnswer raises the question
interrupt <tool_call_id>-input-<n>. The first poll after the question is
resolved, which runs at once, sends the resolution with tasks/update: an
accept input response with the answer, decline for a declined question,
or cancel for a passed deadline. It sends it only while the task still
lists that same request, so it never answers a different later request
under the same key, and only on its first attempt, after a heartbeat. With
output sent, a half-close of the stream fails the attempt instead of returning
it as not run, so once the send may have happened, Dex retries that attempt
rather than handing it out again under the same attempt number.
A retried attempt cannot tell whether the first one sent it, and the task may
since have asked the identical request again, so it asks the user again
instead: the user may answer twice, but an answer never reaches a later
question.
Right after tasks/update, the task may still list the answered request until
it takes the answer in. For 10 seconds after the send, polls run a second apart
and read that request as working; the record's continuation carries the send
time. Once a poll sees the task leave input_required, or after those 10
seconds, an identical request is a new question, so a task that asks the same
thing once per batch is asked each time. The task's own result then ends the
call. When tasks/update fails for a decline, or the task is cancelled after
one, the source cancels it best-effort and fails the poll, so the call is
recorded failed for a read tool and outcome_unknown otherwise.
Any other input request records failed with the MCP server asked for input that this agent session cannot provide yet (...), naming each request, such as
an answer to 'Which account?', and the waiting task is cancelled. A stdio
task's input request fails this way too, because its process ends with the
request. An input_required answer to tools/call itself, outside a task,
records failed the same way; mapping it to a question is a follow-up. A
state-only input_required answer, which only asks for the same request
again, is retried at once.
With a framework adapter
The runner above is the core's runner-agnostic path. A framework adapter uses
the source when it accepts tools defined by a JSON Schema: wrap each McpTool
as a framework tool whose function calls tools.call_tool(name, arguments, idempotency_key=...) with the key from the adapter's tool context, and pass
{**tools.tool_policies(), **your_policies} as the adapter's tool_policies.
No adapter package changes for this.
| Adapter | Schema-only tool | Key inside the tool |
|---|---|---|
dex-agents-openai-agents |
FunctionTool(params_json_schema=..., on_invoke_tool=...) |
get_current_deferred_tool_execution(), the whole call |
dex-agents-pydantic-ai |
Tool.from_schema(...) |
deferred_tool_idempotency_key(ctx) |
dex-agents-langgraph |
StructuredTool with a JSON Schema args_schema |
get_tool_call_idempotency_key(config) |
dex-agents-google-adk |
a BaseTool with its own FunctionDeclaration |
idempotency_key_for_tool_call(tool_context) |
dex-agents-claude-agent-sdk |
claude_agent_sdk.tool(name, description, schema) |
current_deferred_tool_execution(); write calls only, reads run inside the harness |
Pydantic AI, LangGraph, and Google ADK also defer their own MCP client tools
into tool Steps; use the source there for the shared idempotency key and
annotation handling. The OpenAI Agents SDK runs its own MCP servers inline in
RunModel, so a wrapper is its only path to tool Steps.
Browser streams
dex_agents.ui_stream lets a browser render a live session from its
AgentEvents Stream and recover after a reconnect. It needs no store besides
Dex and no web framework. An application mounts two endpoints:
- Snapshot.
read_agent_event_stream_snapshot(client, flow, session_id)returns theget_agent_snapshotresult plus the Dex resume token to open the event stream from. Serve itsto_json():{"snapshot": ..., "resume_token": ...}in the wire format. While a turn is active, the token replays that turn's events, so the browser can rebuild its live text. The snapshot leaves out the fields only the session's Steps use:framework_itemon messages, the framework's replay data, which can hold the agent's instructions, andtool_source_reference,trace_context,input_continuation,input_answers, andtask_answer_pendingon tool calls. Read them withget_agent_snapshoton the server when needed. It includessession_settingsas stored, so keep anything a browser must not see out of session settings, or remove the key. - Events.
await stream_agent_event_frames(client, flow, session_id, resume_token=...)checks the token with one short read, then returns a generator of Server-Sent Event frames. Each frame'sidis the event's Dex resume token and itsdatais theAgentEventwire JSON, withpayloadas JSON text. A reconnectingEventSourcesends the lastidasLast-Event-ID; pass it asresume_tokenand the stream continues with the next event, with no repeated or missing event while the Stream retains them. After each empty long poll it yieldsKEEP_ALIVE_FRAME, an SSE comment. A token that Dex rejects, such as one of another session, raisesInvalidAgentEventResumeTokenErrorbefore the response starts; answer 400, which stops theEventSource, and let the browser load the snapshot again.
The _sync variants, read_agent_event_stream_snapshot_sync and
stream_agent_event_frames_sync, take a synchronous Client for WSGI
servers. Send AGENT_EVENT_STREAM_HEADERS with the event response. The JSON
matches the TypeScript core's @superdurable/dex-agents/ui-stream, which also
offers an AI SDK UI message stream for useChat.
Authorize every route. The session ID and the resume token come from the browser, so each route must check that the caller may read the session it names before it calls a helper.
FastAPI or Starlette, with an AsyncClient:
from typing import Any
from fastapi import FastAPI, Request, Response
from fastapi.responses import StreamingResponse
from dex_agents.ui_stream import (
AGENT_EVENT_STREAM_HEADERS,
InvalidAgentEventResumeTokenError,
read_agent_event_stream_snapshot,
stream_agent_event_frames,
)
app = FastAPI()
@app.get("/sessions/{session_id}/snapshot")
async def get_session_snapshot(session_id: str, request: Request) -> dict[str, Any]:
await authorize_session(request, session_id)
return (await read_agent_event_stream_snapshot(client, flow, session_id)).to_json()
@app.get("/sessions/{session_id}/events")
async def get_session_events(session_id: str, request: Request) -> Response:
await authorize_session(request, session_id)
resume_token = request.headers.get("last-event-id") or request.query_params.get(
"resume_token", ""
)
try:
frames = await stream_agent_event_frames(
client, flow, session_id, resume_token=resume_token
)
except InvalidAgentEventResumeTokenError:
return Response(status_code=400)
return StreamingResponse(frames, headers=dict(AGENT_EVENT_STREAM_HEADERS))
Starlette without FastAPI uses the same calls in Route handlers, returning
JSONResponse(value.to_json()) and StreamingResponse(frames, headers=...).
Quart, with an AsyncClient. Turn off the response timeout, because the
stream stays open:
from quart import Quart, make_response, request
app = Quart(__name__)
@app.get("/sessions/<session_id>/snapshot")
async def get_session_snapshot(session_id: str):
await authorize_session(request, session_id)
return (await read_agent_event_stream_snapshot(client, flow, session_id)).to_json()
@app.get("/sessions/<session_id>/events")
async def get_session_events(session_id: str):
await authorize_session(request, session_id)
resume_token = request.headers.get("Last-Event-ID") or request.args.get("resume_token", "")
try:
frames = await stream_agent_event_frames(
client, flow, session_id, resume_token=resume_token
)
except InvalidAgentEventResumeTokenError:
return "", 400
response = await make_response(frames, dict(AGENT_EVENT_STREAM_HEADERS))
response.timeout = None
return response
Flask, with a synchronous Client. Each open stream holds one worker thread,
so serve it from a threaded or greenlet server:
from flask import Flask, Response, request
from dex_agents.ui_stream import (
AGENT_EVENT_STREAM_HEADERS,
InvalidAgentEventResumeTokenError,
read_agent_event_stream_snapshot_sync,
stream_agent_event_frames_sync,
)
app = Flask(__name__)
@app.get("/sessions/<session_id>/snapshot")
def get_session_snapshot(session_id: str):
authorize_session(request, session_id)
return read_agent_event_stream_snapshot_sync(sync_client, flow, session_id).to_json()
@app.get("/sessions/<session_id>/events")
def get_session_events(session_id: str) -> Response:
authorize_session(request, session_id)
resume_token = request.headers.get("Last-Event-ID") or request.args.get("resume_token", "")
try:
frames = stream_agent_event_frames_sync(
sync_client, flow, session_id, resume_token=resume_token
)
except InvalidAgentEventResumeTokenError:
return Response(status=400)
return Response(frames, headers=dict(AGENT_EVENT_STREAM_HEADERS))
In the browser, load the snapshot, then open the event stream from its token:
const { snapshot, resume_token } = await (await fetch(`/sessions/${sessionId}/snapshot`)).json();
render(snapshot);
const events = new EventSource(
`/sessions/${sessionId}/events?resume_token=${encodeURIComponent(resume_token)}`,
);
events.onmessage = (message) => applyAgentEvent(JSON.parse(message.data));
On a reconnect the browser repeats the URL and adds Last-Event-ID, which the
routes above prefer to the query parameter. The browser applies the event
rules in Flow shape: within one step_execution_id it keeps
only the highest attempt, so a RunModel retried after a crash shows only
its new attempt's text. After turn_completed, turn_failed, turn_steered,
or turn_cancelled, and after any long gap, it reloads the snapshot. Approve or reject a pending interrupt from
the snapshot with AgentSessionClient.resolve_interrupt.
Limits:
- Stream events live in the memory of the active Dex Server process, so a
restart or failover loses them, and trimming can remove old events. With the
Stream Store disabled (
streamStore.backend: disabled), the snapshot still works, with an empty token, and the event stream yields only keep-alive comments; poll the snapshot instead. - The event stream never ends by itself; the web framework closes it when the
browser disconnects. An async generator stops when the framework cancels
it; a synchronous one first finishes its current long poll, up to
poll_timeout. - Stopping in the browser only closes the response. To stop the turn, the
application's stop route calls
cancel_turn.
Tracing
The session emits OpenTelemetry GenAI spans through the OpenTelemetry API. The
otel extra installs only opentelemetry-api; the application brings the SDK
and an exporter:
pip install "dex-agents[otel]" opentelemetry-sdk opentelemetry-exporter-otlp
Tracing is on by default and uses the global tracer provider, so an
application that configures the SDK gets spans without further setup. Without
the API, the session never imports OpenTelemetry. With the API but no SDK, the
spans are no-ops and the session stores nothing extra. AgentTracingOptions,
passed as AgentSessionOptions.tracing, changes the defaults:
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from dex_agents import AgentSessionFlow, AgentSessionOptions, AgentTracingOptions
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
flow = AgentSessionFlow(
MyFrameworkRunner(agent),
flow_type="TriageAgent",
options=AgentSessionOptions(
tracing=AgentTracingOptions(tracer_provider=provider, agent_name="Triage"),
),
)
| Option | Default | Meaning |
|---|---|---|
is_enabled |
True |
False emits nothing and never imports OpenTelemetry. |
tracer_provider |
None |
The provider to emit with; None uses the global provider. |
agent_name |
the Flow type | gen_ai.agent.name and the invoke_agent span name. |
should_capture_message_content |
False |
Put message text, tool arguments, and tool results on spans. |
semantic_conventions_version |
V1_42_0_DEV |
The GenAI semantic conventions the spans follow. |
One trace per activation
An activation is a burst of Steps between durable waits. It starts when a Step
consumes an input from outside the session: StartAgentSession,
AwaitUserMessage, or ApplyTurnControl with a user message, or
AwaitToolApproval with an interrupt answer or an expired deadline. It ends when every Step it started has
ended or waits again. Each activation is one trace:
| Span | Kind | Emitted by | Parent |
|---|---|---|---|
invoke_agent {agent name} |
INTERNAL |
Each attempt of the Step that starts the activation | None. It is the trace root, with a span link to the previous activation's root. |
chat, or chat {model} when the runner returns ModelCallMetadata |
CLIENT; INTERNAL when the runner reports model_calls other than 1 |
Each RunModel and CompactContext attempt, when it ends |
The activation's root |
execute_tool {tool name} |
INTERNAL |
Each CallTool, CallSingleAttemptTool, and PollToolTask attempt |
The root of the activation that deferred or approved the call |
A turn without approvals is one trace. Each approval answer starts another, linked to the one before, and so does each new turn. Every span ends inside its Step attempt, so no span stays open across a durable wait, and every backend sees a closed root.
The invoke_agent root ends with the Step that starts the activation, so its
duration is near zero, and the activation's later spans start after it ended.
That is on purpose. A span must end in the process that started it, and the
OpenTelemetry API cannot give a span an ID before it starts, so a root that
covered the whole activation would have to stay open across Steps, Workers,
and possibly days. Tracing backends drop or hide such traces. A closed root
with later children is a valid trace that every backend shows.
While a tool call runs, its execute_tool span is the current span. While a
model call runs, the activation's root is current, because the chat span
starts when the attempt ends: its kind depends on the model_calls the runner
reports. Framework and HTTP client instrumentation therefore nests under the
activation.
The Steps of one activation can run in different Worker processes. The root
therefore stores its W3C traceparent, about 55 bytes, in the
AgentTraceContext Attribute in the same commit as its Step. RunModel reads
it to parent its chat span and copies it onto each tool call record it
creates, as AgentToolCall.trace_context. An approval replaces the record's
copy with its own root. Each tool Step parents its spans from its record, so
overlapping approvals never swap tool spans between traces. The next
activation's root reads the Attribute for its link.
The Attribute is written whenever the root has a valid span context, sampled
or not. A sampled-out root stores its unsampled context, so under a
parent-based sampler the activation's other spans are dropped with it instead
of starting traces of their own. When a root has no valid context, because
its Worker has tracing disabled or has no OpenTelemetry API or SDK, that Step
deletes the stored value. The activation's later spans then start traces of
their own, instead of joining the previous activation's trace. The schema
always registers AgentTraceContext, so Workers with and without tracing can
serve one Flow type.
Every attempt is its own span with dex.step.attempt, and all attempts of one
Step execution share dex.step.execution_id, so a retry shows as a second span
of the same Step. A failed attempt has status ERROR, error.type set to the
exception class name, and the exception as an event. A ToolExecutionError
sets error.type to ToolExecutionError. An attempt whose Worker died exports
no span. ResumeTurn, RecordToolFailure, RecordToolOutcomeUnknown, and
RecordTurnFailure emit no span: the failed attempts before them already have
error spans.
Attributes
| Span | Attribute | Value |
|---|---|---|
| All | gen_ai.conversation.id, session.id, dex.flow.id |
The Flow ID, which is the session ID |
| All | dex.run.id |
The Flow run ID |
| All | dex.step.type, dex.step.execution_id, dex.step.attempt |
The Step that emitted the span |
| All | dex.agent.turn.id |
turn-<n> |
| All | gen_ai.operation.name |
invoke_agent, chat, or execute_tool |
invoke_agent, execute_tool |
gen_ai.agent.name |
agent_name |
invoke_agent |
dex.agent.activation.trigger |
user_message, interrupt_resolution, or interrupt_deadline |
invoke_agent |
dex.agent.user_message.key |
message_sequence_key of the user message |
invoke_agent |
dex.agent.interrupt.id, dex.agent.interrupt.approved |
The interrupt and its decision |
chat |
gen_ai.request.model, gen_ai.response.model, gen_ai.provider.name, gen_ai.response.id, gen_ai.response.finish_reasons |
From ModelCallMetadata, when the runner returns it |
chat |
gen_ai.usage.input_tokens, gen_ai.usage.output_tokens |
From usage, when model_calls is 1 and the input or output count is not zero |
chat |
gen_ai.usage.cache_read.input_tokens, gen_ai.usage.reasoning.output_tokens |
cached_input_tokens and reasoning_tokens, under the same condition, when not zero |
chat |
dex.agent.input_messages.last_key |
message_sequence_key of the newest stored message the call received; for CompactContext, the last message it summarized |
chat |
dex.agent.output_messages.keys |
message_sequence_key values of what this call produced: the TurnCompleted messages after the turn's last tool result |
chat |
dex.agent.requested_tool_call.ids |
AgentToolCalls keys that a ToolCallsRequested created |
execute_tool |
gen_ai.tool.name, gen_ai.tool.call.id |
The tool and its AgentToolCalls key |
execute_tool |
dex.agent.tool.effect, dex.agent.tool_call.status |
The effect and the final status |
The chat span is named chat {request_model} when the runner returns
ModelCallMetadata.request_model, and chat otherwise. Each framework
adapter in this repository returns ModelCallMetadata with what its framework
reports for the call and leaves the rest unset. Its README lists the fields it
sets and where they come from.
A runner that reports model_calls other than 1, such as a whole-turn harness
that makes several model calls in one RunModel, gets an INTERNAL chat
span without usage attributes. Instrumentation of the provider client
typically traces each of those calls with its own usage, so a summed count on
the session's span would count the tokens twice. Usage still reaches the
AgentUsage Attribute.
A single model call can still be counted twice. When instrumentation of the
provider client, such as an OpenAI or Anthropic instrumentation library, also
traces the call, both its span and the session's chat span carry usage. Sum
usage from one instrumentation scope in the tracing backend, or remove the
session's usage attributes, for example with an OpenTelemetry Collector
transform processor:
processors:
transform/drop_dex_agents_usage:
trace_statements:
- context: span
statements:
- delete_matching_keys(attributes, "^gen_ai\\.usage\\.") where instrumentation_scope.name == "dex_agents"
Tracing failures
Tracing never fails or retries a Step. Every OpenTelemetry call is guarded:
starting, annotating, and ending spans, changing the current context, and
serializing content. An exception there, such as one from a span processor or
exporter, is logged on the dex_agents.tracing logger and that span is
skipped. The first failure in a process logs a warning with its traceback;
later ones log at debug level. The Step's own result is unchanged, so a broken
processor can never send a write to outcome_unknown or rerun a tool.
Message content
Spans carry content by reference by default: the Flow ID, Step execution ID,
message sequence keys, and tool call IDs above, because the content is already
durable in Dex. Read it with get_agent_snapshot, list_agent_messages, or
AsyncClient.get_attributes. With should_capture_message_content=True, spans
also carry content as JSON text in the GenAI message format:
invoke_agent:gen_ai.input.messages, the user message.chat:gen_ai.input.messages, the stored messages plus the turn's tool calls and their recorded results, andgen_ai.output.messages, what this model call produced. ForCompactContext, the messages it summarized and the summary text.execute_tool:gen_ai.tool.call.argumentsandgen_ai.tool.call.result.
Content comes from the AgentMessage projections (text and the tool_*
fields), not from framework_item. System instructions and tool definitions
are not captured. Content can be sensitive and large; capture it only where the
tracing backend may store it.
The spans set no gen_ai.input.messages_ref or gen_ai.output.messages_ref.
Those attributes take an absolute URI, and Dex has no URI for a stored
message. Use the dex.agent.* keys with the Flow ID instead.
Semantic conventions version
The GenAI semantic conventions are still marked Development, so the session
pins a version and sets its schema URL on the tracer.
GenAISemanticConventionsVersion.V1_42_0_DEV, the default, follows
open-telemetry/semantic-conventions-genai
at commit dc3571562603552dc555c667dfa327a638d94ab2
(2026-11-01). Its schema URL is
https://opentelemetry.io/schemas/gen-ai-dev/1.42.0-dev; the repository has no
release tag yet.
V1_36_0 follows OpenTelemetry semantic conventions 1.36.0 for backends that
still read the older names: chat spans carry gen_ai.system instead of
gen_ai.provider.name, omit the cache and reasoning token counts, and never
carry content, because 1.36.0 recorded content as events.
Testing an adapter
dex_agents.testing runs integration tests against a private dexcli dev --engine zd engine. It reads DEXCLI and expects dex-server-zd next to it
(or DEX_SERVER_ZD_BINARY). The engine runs with
dev.durabilityMode: processCrashSafe, which skips fsync and survives a
process kill, through a -zd-config file that also works with dexcli
releases that predate --durability. A
server_zd_config file of your own replaces it, so set dev.durabilityMode
there too. The Worker runs in a child process in its own process group, so a
test can SIGKILL it at an exact moment.
# tests/my_worker.py: the Worker child.
from dex_agents.testing import run_agent_worker
from tests.my_agent import build_flow
if __name__ == "__main__":
run_agent_worker([build_flow()])
# tests/test_approvals.py: the test process builds the same Flow for its Client.
from pathlib import Path
from dex_agents.testing import EXECUTE_COMPLETED_EVENT, AgentTestStack, wait_for_agent_snapshot
from tests.my_agent import build_flow
PACKAGE_DIR = Path(__file__).resolve().parents[1]
async def test_write_survives_a_worker_kill(tmp_path: Path) -> None:
flow = build_flow()
async with AgentTestStack(tmp_path, [flow], "tests.my_worker",
worker_cwd=PACKAGE_DIR,
worker_env={"MY_STATE_DIR": str(tmp_path)}) as stack:
sessions = stack.sessions(flow)
worker = await stack.start_worker()
await sessions.start_session("session-1", "refund order 7")
interrupt = await wait_for_agent_snapshot(
sessions, "session-1",
lambda snapshot: next(iter(snapshot.pending_interrupts), None),
description="a pending approval")
worker.kill() # SIGKILL while the session waits
await stack.start_worker() # replacement Worker
assert await sessions.resolve_interrupt(
"session-1", interrupt.interrupt_id, approved=True)
history = await stack.read_step_history("session-1")
executed = [entry.step_type for entry in history
if entry.event_type == EXECUTE_COMPLETED_EVENT]
Run tests with pytest-asyncio in auto mode, as pyproject.toml here does.
Inside the Worker child, a tool marks a FileGate(state_dir, name) and awaits
wait_for_release(); the test awaits wait_until_reached() and kills the
Worker at that exact point. Two gates reached at once prove that two tool
Steps run in parallel. InvocationLog(path) counts real tool and model calls
across Worker processes. read_step_history(server_address, flow_id) lists
completed Steps (StepHistoryEntry with event_type, step_type,
step_execution_id, final_attempt) through dexcli flow history.
wait_until(description, probe) polls with a deadline instead of sleeping.
The core's own tests in tests/ use a scripted runner and show every pattern.
Each package's ci.sh installs locked dependencies, runs lint and type
checks, and runs the integration tests with DEXCLI set. Agents CI
discovers every agents/*/*/ci.sh.
The 2,000-turn compaction session behind the agent deployment guide's storage
numbers runs for minutes. pytest skips it unless
DEX_AGENTS_LONG_TESTS=1, and ci.sh runs an 800-turn guard with a retention
limit of 100 instead. The
Agents Long Tests Nightly workflow runs it every night. Run it by hand with:
cd agents/python/core
DEX_AGENTS_LONG_TESTS=1 DEXCLI=/path/to/dexcli uv run --frozen pytest -s tests/test_compaction.py -k 2000_turn
Design decisions
The plan records each of these decisions too.
- Step mode: one
RunModelStep per model call and one Step per tool call, READ included. No framework loop runs several model calls in one Step. - Tool call Steps and their failure routes take the tool call ID as input.
There is no shared "current tool call" in
AgentState. RecordToolFailurerecordsfailedfor a read andoutcome_unknownfor a write: it is reached only through crashes, timeouts, or exceptions other thanToolExecutionError, so a write may have taken effect.- A failed tool Step whose recovery error reports
attempt_progressNOT_STARTEDnever ran the tool, so its call recordsfailedwithTOOL_CALL_NOT_RUN_ERROR_PREFIXrather than a new status. The model sees an ordinary failure and can call the tool again. AgentToolCall.deferral_orderkeeps model order for display and for the turn-failure note; AttributeMap keys alone do not keep order.- By default nothing from a turn is stored until it completes:
TurnCompletedcarries all of the turn's messages andToolCallsRequestedkeeps them in the framework run state. A runner may store them incrementally withToolCallsRequested.new_messages; Stepwise does. - Built-in tools are routed by the session, not the runner, so they work under every adapter that offers their definitions.
- A failed turn that ran tools leaves a note message instead of appending nothing, so its writes are not repeated by the next turn.
- Every Step sets SYNC durability explicitly.
- Message sequences start at 1. A blank
initial_messagestarts no turn.last_turn_erroris cleared when a later turn completes. resolve_interruptdropsrejection_messageon approval. When it resolves an interrupt just as its deadline Timer fires,AwaitToolApprovalwaits again and applies the answer.AgentSnapshot.stateisNoneuntilStartAgentSessioncommits.- Status, role, effect, and record status values are
StrEnumtypes in Python; the wire carries the plain strings.
Limits
- Without compaction,
RunModelloads every stored message, and nothing is deleted.get_agent_snapshot,list_agent_messages, and aRecordTurnFailureof a runner that builds unfinished-turn messages load every kept chunk; retention bounds that load. - Each model call and tool call adds a Step round trip.
- Steering never stops a model call or a tool call that has started; it
applies at the next safe Step boundary.
cancel_turnstops a model call but never a tool call Step that has started. - Interrupt deadlines use the Worker clock.
- One
text_deltacall writes one Stream message; batch very small chunks. - Tool call IDs that are empty or contain
/fail the turn. - The tool Steps use SYNC durability, and Dex never hands one attempt number to a second Worker, so a single-attempt tool does not run twice after a Worker crash. Cancellation is cooperative, though: a cancelled run may keep running while the failure route starts.
attempt_progressNOT_COMPLETEDis conservative, so a call whose tool never ran can still recordoutcome_unknown, such as when Dex Server crashed after it recorded the attempt start but before the task reached the Worker.- A tool the framework cannot defer runs inside
RunModeland repeats on aRunModelretry. - From server-zd 0.1.21, turn records cleared at turn end no longer count
against the Server's
limits.maxFlowStateBytesrun budget. Older servers count them as superseded values until the history window prunes them. dex_agents.mcppolls a stdio server's MCP task inside its tool Step for at mosttask_wait_timeout, holding a Worker slot, because the task ends with the server process.- Each
PollToolTaskpoll is a Step in the session's history; a task that runs for days adds up to a Step a minute while the server's interval is shorter. - An MCP task that asks the identical request again within 10 seconds of an
answer, without leaving
input_requiredin between, waits until those 10 seconds pass before the user is asked. A server slower than 10 seconds to take an answer in has its question asked twice. - A retried
PollToolTaskattempt asks the user again rather than send an answer that its first attempt may have sent. - An MCP
input_requiredanswer totools/calloutside a task recordsfailed; mapping it to a session question interrupt is a follow-up. dex_agents.mcpstarts a new stdio process, or opens a new HTTP connection, for every request. An unreachable server is tried again, with its connection attempts, on everylist_toolscall until it answers.- MCP annotations are hints from the server and are trusted by default. A
wrong
readOnlyHinton a write makes that write run without approval and retry; settrusts_tool_annotations=Falseor aToolPolicyfor servers you do not control. - Tracing covers the session's Steps only. The Dex SDK emits no spans, so Dex Server and Worker internals do not appear in the traces.
- The adapters report only the model information their framework exposes, so
some
chatspans carry no provider, response ID, or finish reasons. Each adapter's README lists what it sets.
Metadata
Release files for dex-agents 0.3.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | Actions |
|---|---|---|---|---|
| dex_agents-0.3.0.tar.gz | 257.2 kB | Without provenance | Details for dex_agents-0.3.0.tar.gz |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Actions Reset |
|---|---|---|---|---|
| dex_agents-0.3.0-py3-none-any.whl | Python 3 | none | any | Details for dex_agents-0.3.0-py3-none-any.whl |
Total release size: 508.6 kB
Release files / dex_agents-0.3.0.tar.gz
| Download URL | dex_agents-0.3.0.tar.gz |
|---|---|
| Size | 257.2 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
eb35ce29dba643124aa496139369e77981215744d1a02e6d93a6228e50eacf98
|
|
BLAKE2b-256 checksum How to use checksums |
d6f55cbc784d7d7ae2e027349d46155ca14f01f7dc5037c14c569a1e2204c0ff
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.12.2 {"installer":{"name":"uv","version":"0.12.2","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"26.04","id":"resolute","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Release files / dex_agents-0.3.0-py3-none-any.whl
| Download URL | dex_agents-0.3.0-py3-none-any.whl |
|---|---|
| Size | 251.3 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
197c71b105316644bdebd28cc341532ddc1e959124684b32deb1e4b873a195b4
|
|
BLAKE2b-256 checksum How to use checksums |
63c9dc335068e5270ca20b24ff015dfecc7867a4e157b6387e0a53700b5a1d69
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.12.2 {"installer":{"name":"uv","version":"0.12.2","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"26.04","id":"resolute","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Security for dex-agents 0.3.0
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.