Agent setup & execution¶

Two paths¶

Convenient: setup_agent()Full control: Agent(...)

from xun import setup_agent

agent = setup_agent(
    name="agent",
    tools=[my_tool],               # your own tool functions
    default_tools=True,            # register all built-in tools + the sub-agent tool
    default_system_prompt=True,
    default_commands=True,
    workdir=".",
)
answer = agent.instruct("...").execute()

Parameter	Type	Default	Purpose
`name`	`str`	`"agent"`	agent name
`tools`	`list[Callable]`	`[]`	tool functions to register additionally
`default_tools`	`bool`	`False`	runs `toolbox.with_defaults().with_subagent_provider()`; the `policy` command group is registered only when `default_tools` and `default_commands` are both true
`default_system_prompt`	`bool`	`True`	injects the built-in system prompt
`default_commands`	`bool`	`True`	registers the built-in session commands
`display`	`DisplayAbstract \\| None`	`None`	display layer; defaults to the console `Display()`
`workdir`	`Path \\| str \\| None`	`None`	working directory; defaults to `Path.cwd()`

The internal order is "configure first, initialise second": initialize() is called last, so the after_initialize hook sees a fully configured agent. The return type is Agent[Agent.T.Init].

Agent is a dataclass whose fields all have defaults and can be overridden as needed:

from xun import Agent, ToolBox

agent = Agent(name="writer", toolbox=ToolBox().with_defaults())
agent = agent.initialize()

Field	Type	Default	Description
`name`	`str`	`agent-<uuid8>`	display name; a sub-agent gets `<parent name>-child-<id8>`
`identifier`	`str`	random UUID	the key the display layer uses to tell agents apart
`display`	`DisplayAbstract`	`NullDisplay()`	silent display; in this case `get_choice` raises `NotImplementedError` (when `auto_confirm` is true the default choice is returned straight away, without reaching the display layer)
`conversation`	`Conversation`	new one	message history
`toolbox`	`ToolBox`	empty	you must call `with_defaults()` or `register()` yourself
`command`	`CommandRegistry`	empty	session commands
`workspace`	`Workspace`	`workdir=Path.cwd()`	working directory + lazily created temporary directory
`cancel_event`	`ChainedEvent`	new one	can be attached beneath a parent agent's token
`config`	`AgentConfig`	`load_config().clone()`	deep copy of the global configuration; `AgentConfig.default()` does not read `config.json` (nor does it load `.env`), but still substitutes `${XUN_*}` placeholders from `os.environ`, and a missing placeholder is an error (only `XUN_OPENAI_MODEL` and `XUN_AUTO_CONFIRM` have built-in fallback values)
`api_call_semaphore`	`Semaphore`	3	maximum number of concurrent model calls; parent and child share the same semaphore
`state`	`dict[str, Any]`	`{}`	free attachment point (currently the sub-agent depth and the policy objects of the built-in tools)
`hooks`	`Hooks`	new one	hook registry
`compactor`	`CompactorAbstract`	`AutoCompactor()`	installed into `before_execution_step` at initialisation time
`extension_loader`	`ExtensionLoader`	process-wide `default_loader`	shared by reference, so sub-agents reuse the same scan

Lifecycle¶

stateDiagram-v2
  [*] --> Uninit: Agent(...)
  Uninit --> Init: initialize() or with
  Init --> Init: turn
  Init --> Final: finalize()
  Uninit --> Final: finalize()
  Final --> [*]

Here "one turn" stands for repeated system / instruct / execute calls; a with block runs initialize() on entry and finalize() on exit.

initialize() performs, in order: apply extensions → auto-detect the model name (when config.model.name is empty) → auto-confirm warning → bind the display layer → prepare the workspace → install the compactor → trigger after_initialize. Calling it on an already initialised agent returns it unchanged, but calling it on a finalized one raises RuntimeError.

with Agent(toolbox=ToolBox().with_defaults()) as agent:   # initialize on entry
    agent.instruct("...").execute()
                                                          # finalize on exit

finalize() triggers before_finalize and unbinds the display layer. If an agent is garbage-collected before being finalised, the fallback registered with weakref.finalize performs the remaining cleanup.

Injecting messages¶

Method	Signature	Description
`system`	`system(content: str) -> self`	overwrite the system message (message 0 of the session); chainable
`instruct`	`instruct(instruction, images=None, _emit_event=True) -> self`	append a user message; `images` accepts paths, URLs or PIL images; `_emit_event=False` means nothing is broadcast to the UI (used internally by sub-agents)

Both are available on Agent[T.Alive], return self, and leave the state unchanged.

Execution¶

# the two overloads (for the real return types see below)
def execute(self, schema: type[T], max_iterations: int = 512, context: Any = None) -> T: ...
def execute(self, schema: None = None, max_iterations: int = 512, context: Any = None) -> str: ...

Because execute carries @except_safe, the actual return type is Result[str, ErrorInfo] or Result[T, ErrorInfo]:

`Result` API	Description
`is_ok()` / `is_err()`	test the outcome
`unwrap()` / `unwrap_err()`	get the value (calling `unwrap()` on an `Err` raises)
`.value`	read the raw value directly
`value_json()` / `value_str()`	JSON object / string form (used when tool results are written to disk)

max_iterations defaults to 512 and can be overridden with the internal environment variable _XUN_DEFAULT_MAX_ITER; exhausting it fails: an ErrorEvent is broadcast first, then RuntimeError is raised, and Err(ErrorInfo("Maximum tool call iterations exceeded.", ...)) is returned in the end.
context is wrapped into a ToolCallContext and injected into tools that declare that parameter; inside the tool, ctx.value retrieves the original value; see the tool system.
Passing a schema (a pydantic model) requests structured output: both response_format (strict=False) and a schema embedded in the prompt are used, as a double safeguard; when parsing, json_repair repairs the text first and then model_validate runs. Structured output requires the last message of the conversation to be a user message: the schema description is appended to the last user message, so instruct() has to be called before every execute(schema=...), otherwise a ValueError is raised (also converted into an Err).

KeyboardInterrupt and CancelledError are not swallowed; they propagate to the caller.

Inside the loop¶

Trigger before_execution (its arguments can be modified in place: schema, max_iterations, context_value; if a hook sets takeover_result, the whole loop is skipped and that string is returned directly — see Hooks & session commands).
Each turn: check_cancel() → before_execution_step (automatic compaction happens here) → stream the model call (limited by api_call_semaphore, with a 900 second per-request timeout because time-to-first-token can be long, stream_options={"include_usage": True}).
Text/reasoning events flow out through model_text_delta and model_reasoning_delta (hooks may rewrite the content).
If there are tool calls: before_tool_call (the list is editable) → arguments repaired with json_repair and written back into the assistant message → executed one by one → after_tool_call → written into the conversation → after_execution_step (where defer_tool_image injects its image).
A tool exception does not interrupt the run; it is turned into an Err result and handed back to the model to deal with. Model-call exceptions — a usage-only empty response counts as one — are retried at most 3 times, each time asking Retry? (with auto-confirm enabled the answer defaults to yes); a provider that reports no token usage ends the run with an error.
The run ends as soon as a turn makes no tool calls; that turn's text is the result.

tools and tool_choice="auto" are sent only when toolbox.list_tools_json(config.model.capabilities) is non-empty (filtered by model capability, so a tool whose required_capabilities are not covered is not sent); temperature and reasoning_effort are sent only when they are not None.

Cancellation¶

API	Description
`agent.cancel()`	set the cancellation token; returns whether this was a "cancel while running"; when idle it returns `False` and leaves the state unchanged
`agent.check_cancel()`	self-check inside long-running tools; raises `CancelledError` as soon as it hits
`agent.is_running`	whether the agent is currently running
`with agent.cancellable_execution():`	manually open a run scope (`run_start` / `run_end` are triggered from here; nestable)

Cancellation is cooperative: the loop checks the token at the start of every turn, on every streaming delta and before every tool call; cancelling a parent agent cascades to all of its sub-agents (ChainedEvent.parent). The token is cleared when the scope exits, so an agent that has been cancelled can still be reused. See Sub-agents & cancellation for the cascade details.

Deriving a sub-agent¶

child = Agent.inherit(
    parent,
    share_workspace=True,      # share the workdir and the same lazy temporary directory
    share_display=True,        # False makes the sub-agent use a NullDisplay
    copy_toolbox=True,         # shallow-copy the tool table
    copy_command=True,
    copy_conversation=False,   # sub-agents start with a blank context by default
    inherit_cancel_event=True, # cascading cancellation
)

See Sub-agents & cancellation for the parameter-by-parameter description, the shallow-copy semantics and the cascading cancellation.