The Agent constructor and every public runtime method (run, resume, the submit family, pause, notes, stream, cancel, retry, close) with exact signatures.

Agent

Agent holds an agent's identity, capabilities, and limits, and exposes the runtime methods that execute it. Construct it once and reuse it across many runs.

from dendrux import Agent
from dendrux.llm.anthropic import AnthropicProvider
 
agent = Agent(
    name="Support",
    provider=AnthropicProvider(model="claude-haiku-4-5"),
    prompt="You are a support agent.",
    tools=[...],
    database_url="sqlite+aiosqlite:///app.db",
)

Agent is an async context manager. Use async with Agent(...) as agent: so MCP sources, the provider, and any agent-owned engine close cleanly. All runtime methods are async.

Constructor

Agent(
    *,
    provider: LLMProvider,
    prompt: str,
    name: str = ...,                       # defaults to the class name
    tools: list[Callable] = [],
    tool_sources: list[MCPToolView | MCPSource | MCPServer | MCPHost] | None = None,
    max_iterations: int = 10,
    max_delegation_depth: int | None = None,
    loop: Loop | None = None,              # defaults to ReActLoop
    output_type: type[BaseModel] | None = None,
    database_url: str | None = None,
    database_options: dict[str, Any] | None = None,
    state_store: StateStore | None = None,
    deny: list[str] | None = None,
    require_approval: list[str] | None = None,
    budget: Budget | None = None,
    pricing: PriceTable | None = None,
    guardrails: list[Guardrail] | None = None,
    skills_dir: str | Path | None = None,
    skills: list[Skill] | None = None,
    deny_skills: list[str] | None = None,
    skill_executor: SkillExecutor | None = None,
    tool_result_transform: ToolResultTransform | None = None,
    before_tool: BeforeToolHook | None = None,
    images: ImagePolicy | None = None,
    blob_store: BlobStore | None = None,
    strict_tenancy: bool = False,
    limits: Limits | None = None,
    tool_threads: int = 8,
    on_governance_change: Literal["warn", "raise"] = "warn",
)

Construction validates its inputs: unknown names in deny/require_approval, deny/require_approval overlap, output_type without SingleCall, tools with SingleCall, and bad values for strict_tenancy, limits, tool_threads or on_governance_change all raise ValueError. A Budget(max_cost_usd=...) for a model with no price raises BudgetConfigurationError here and again at run start.

Limits

Limits is a frozen dataclass of positive integers. There is no None to switch a cap off; raise the number instead.

from dendrux import Agent, Limits
 
agent = Agent(provider=provider, prompt="...", limits=Limits(max_input_bytes=4 * 1024 * 1024))

The note cap is enforced by one conditional insert in the store, so concurrent submit_note calls cannot exceed it. A custom StateStore whose add_note has no max_notes keyword keeps working without the cap.

Transforming tool results

tool_result_transform runs after every successful server-executed tool call — local @tool functions and MCP tools alike — and, since 0.5.0, on every successful result a client submits through submit_tool_results or resume; it runs before the result is serialized. Whatever it returns becomes the result: the persisted ToolInvocation.result, the TOOL message the model sees, and the notifier/stream events all carry the transformed value.

from dendrux import Agent, ToolResultContext
 
counter = 0
 
def tag_tables(ctx: ToolResultContext):
    global counter
    if not isinstance(ctx.result, list):
        return ctx.result            # always return something: ctx.result keeps it unchanged
    counter += 1
    return {"result_id": f"r{counter}", "rows": ctx.result}
 
agent = Agent(provider=provider, prompt="...", tools=[query_sales],
              tool_sources=[warehouse_view], tool_result_transform=tag_tables)

ToolResultContext is frozen and carries tool_name, call_id, params, result (the raw Python return value), target, and meta (the tool's ToolDef.meta; for MCP tools it includes source_name, namespace, and mcp_tool_name). Sync and async callables are both accepted.

• The return value always replaces the result. Return ctx.result to leave it unchanged — None is a real result, not "no change".
• If the transform raises, the tool call fails (success=False) with an error naming tool_result_transform; the model sees it as a tool error.
• It is not applied to failed tool calls or failed client results.
• For a client result, ctx.result is the submitted payload parsed as JSON (plain text arrives as a string) and ctx.target is "client"; branch on ctx.target if the hook is meant for server tools only. The submitted result is validated and capped first, the transform runs, and the output is validated and capped again. If the transform raises or returns something that cannot be stored, the submission is rejected with InvalidToolResultError, the run stays paused and unclaimed, and the client can resubmit.
• It also runs for tools executed on resume after submit_approval.
• It runs on the raw result; guardrails scan the serialized tool message afterwards. MCP results arrive in their adapted form (e.g. structuredContent as {"result": ...}).
• Sync transforms run on the event loop, so keep them cheap.

run

async def run(
    user_input: UserContent,
    *,
    history: list[ChatMessage] | None = None,
    activate_skills: Sequence[str] | None = None,
    tenant_id: str | None = ...,           # omitted = untenanted (None), unless strict_tenancy
    metadata: dict[str, Any] | None = None,
    notifier: LoopNotifier | None = None,
    idempotency_key: str | None = None,
    output_type: type[BaseModel] | None = None,
) -> RunResult

UserContent is str | Sequence[str | Image | ImageRef | ImagePart]: a plain string as before, or a list of text and images in the order the model should see them. Image problems raise InvalidImageError, ImageLimitError, ImageFetchError or UnsupportedModalityError before a run exists. See Images.

Start a new run. Returns a RunResult whose status is terminal (success, error, cancelled, max_iterations) or paused (waiting_*). history seeds prior conversation turns (Chatbot threads); metadata is opaque JSON stored on the run for your queries (top-level keys starting with dendrux. are reserved and raise ValueError; at most Limits.max_metadata_bytes); idempotency_key requires persistence and is unique per tenant, so two tenants can reuse a key (Multi-tenancy). tenant_id="" raises ValueError. activate_skills names skills to deliver in the run's first message instead of waiting for the model to call use_skill; unknown or denied names raise ValueError before any run is created (Skills).

On success, RunResult.answer contains only the final answer turn after tool use, not all assistant text produced during the run. RunResult.commentary is the ordered list of non-empty assistant text from every step that ended in tool calls (narration such as "Let me look that up"); it is [] for SingleCall, for runs with no narration, and for idempotent cached results. Persisted traces carry the same split as TraceEntry.phase. A paused run may instead carry a clarification question.

Truncated answers

A successful RunResult has stop_reason, why the call that produced the final answer stopped (end_turn, max_tokens, stop_sequence, content_filter, refusal, pause_turn, tool_use, other, or None when the provider reported none). If the model hit its output-token limit while writing the answer, the run still ends success, but result.truncated is True and Dendrux logs a warning. Check it before you trust the answer:

result = await agent.run("Summarise this report")
if result.truncated:
    ...  # raise max_tokens, or ask the model to continue

When streaming, the RUN_COMPLETED event has the same event.stop_reason and event.truncated. The run.completed event in the store (and over RunStore.follow) carries stop_reason and truncated. Dendrux does not continue a truncated answer automatically. Anthropic's model_context_window_exceeded counts as max_tokens. See Truncated answers.

With output_type, a final answer that fails validation raises StructuredOutputValidationError instead. It carries usage, model and stop_reason for the rejected call, which is still recorded and priced; the run row's own totals stay at 0, so read the cost from RunStore.get_subtree(run_id).usage. See Structured output.

Two ways to stop a run() in flight behave differently. Cancelling the task awaiting it yourself (including via asyncio.wait_for) atomically finalizes a still-running persisted run as cancelled, emits run.cancelled (reason task_cancelled), and re-raises CancelledError (wait_for raises TimeoutError); no RunResult is returned. Calling agent.cancel_run(run_id) on the same Agent instance in the same process instead interrupts the in-flight model call or tool and makes run() return RunResult(status=CANCELLED) with meta["interrupted"] = True and meta["partial_output"] = False. Existing paused or terminal outcomes are preserved. Stream cancellation has the partial-answer semantics described below. See Cancellation.

A finished run's RunResult.meta["undelivered_notes"] lists inbox note ids it never delivered (present only when there are any); see Inbox notes.

RunResult

@dataclass
class RunResult:
    run_id: str
    status: RunStatus
    answer: str | None
    output: Any                  # the validated output_type model, when one is configured
    steps: list[AgentStep]
    iteration_count: int
    usage: UsageStats
    error: str | None
    meta: dict[str, Any]
    commentary: list[str]
    replay: ReplaySummary | None
    stop_reason: StopReason | None
    error_info: ErrorInfo | None
    pause_id: str | None
    pending: PendingPause | None
 
    @property
    def truncated(self) -> bool
    def to_dict(self) -> dict[str, Any]

to_dict(): the browser envelope

RunResult.to_dict() is the one result shape meant to leave your server. It is an allowlist, not a serializer of the dataclass, and it carries its own version so the shape does not change with the package version:

The PII policy matches the in-process one:

• Client-tool arguments hold real values only where the PII restore policy would restore them for a server tool: always with restore="all" (the default), and only for the tools listed with receives_pii under restore="allowlisted". Other tools show placeholders. Client arguments can still be sensitive, so authorize the actor as well as the tenant and run before you send them.
• Approval and input prompts and arguments are placeholders, taken from the model-sent form that RunStore.get_pending returns. If the verified arguments are not available, or the display would exceed its bounds, params is None and args_hidden is True: resolve that pause on the server and never approve an empty object.
• Never included: pii_mapping, the internal origin map, prompts, steps, traces, meta, tool payloads, provider bodies and image bytes.

Display limit. When a user's input holds a very large number of PII values (around a thousand), a large approval argument can be hidden on the wire. This fails closed: the approver sees args_hidden=True and "arguments hidden; resolve on the server", the pause keeps its id and call ids, and the server-side get_pending view is still available for a fuller review.

The matching inbound parsers are ToolResult.from_dict(data, limits=None) and ToolResult.list_from_dict(items, limits=None, max_results=1000). They accept only name, call_id, payload, success, error, duration_ms and images (image references, never bytes), require finite JSON or bounded plain text, replace an oversized payload with the size_limit failure, reject duplicate call ids, and hold the whole batch to Limits.max_input_bytes. They do not replace your HTTP body limit or the call-id checks that submit_tool_results runs. A browser's tool result is an assertion by the client, not proof the tool ran.

get_result

async def get_result(run_id: str, *, tenant_id: str | None, include_pii: bool = False) -> RunResult

Rebuilds the RunResult of a stored run for a route that did not hold the original object: a reload, a reconnect, or a different worker. tenant_id is required and keyword-only (None means untenanted runs); UNSCOPED is refused. A missing run, a foreign run and a run of another agent all raise the same RunNotFoundError, and a malformed id or tenant raises ValueError before any query. pending and pause_id are filled for a paused run, answer is restored with the run's PII mapping exactly as run() returned it, and output is rebuilt when an output_type is configured. meta["pii_mapping"] is not included unless you pass include_pii=True (a trusted, server-side caller). It checks tenant scope, not the actor: authorize the actor separately.

stream

def stream(
    user_input: UserContent,
    *,
    history: list[ChatMessage] | None = None,
    activate_skills: Sequence[str] | None = None,
    tenant_id: str | None = ...,           # omitted = untenanted (None), unless strict_tenancy
    metadata: dict[str, Any] | None = None,
    notifier: LoopNotifier | None = None,
    idempotency_key: str | None = None,
) -> RunStream

Same inputs as run (including activate_skills, validated before the stream is returned), but returns a RunStream immediately (no await). stream.run_id is available before iteration; with an idempotency key it is provisional until first iteration resolves the key. Iterate it for RunEvents, or .text() for text deltas. Prefer async with agent.stream(...) as s: for deterministic cleanup. To interrupt from another coroutine, call cancel_run(stream.run_id); the stream ends with RUN_CANCELLED carrying the partial answer. Closing the stream early also cancels the run and persists its current partial answer. Guardrails are not supported on the streaming path.

The final RunResult.answer contains the final answer turn only. On cancellation, the partial answer contains only the current text buffer, which resets at every TOOL_RESULT. It can be None even if earlier text was streamed. For example, "Checking…" → tool result → "The answer is 42" yields an answer of "The answer is 42"; cancellation immediately after the tool result can yield None. RunResult.commentary holds the text of the tool steps that completed before the cancellation. To retain the full transcript, accumulate TEXT_DELTA events (or .text()) across all turns, or read persisted assistant messages with RunStore.get_traces(). Traces contain recorded messages, not every delta from an interrupted model call.

Commentary vs final answer

Text a model writes before a tool call ("I'll start by exploring the database…") streams as ordinary TEXT_DELTA events, so concatenating every delta mixes narration into the answer. Each LLM-output and tool event carries iteration (the loop step it belongs to; None on lifecycle events) and some carry phase:

A text block streams before the tool call that makes it commentary, so the phase of a delta is usually unknown when it arrives. Clients buffer instead:

buckets: dict[int, list[str]] = {}
commentary: set[int] = set()
async with agent.stream("...") as stream:
    async for event in stream:
        if event.type == RunEventType.TEXT_DELTA:
            buckets.setdefault(event.iteration, []).append(event.text)
        elif event.phase == "commentary":
            commentary.add(event.iteration)       # that step's text was narration
        elif event.type == RunEventType.RUN_COMPLETED:
            answer = event.run_result.answer       # authoritative final answer
            narration = event.run_result.commentary

Render a bucket provisionally as it streams, restyle it as progress when a commentary event arrives for its step, and treat the remaining bucket as the answer when RUN_COMPLETED carries phase="final_answer". Persist run_result.answer, not the concatenated deltas. After the fact the same split is on TraceEntry.phase (RunStore.get_traces()) and on the phase field of each llm.completed event in the durable event log.

idempotency_key uses the same persistence requirement and request fingerprint as run() (agent name, input, history, context, and output type). A matching terminal run yields one terminal event with its original ID and cached summary result, without calling the provider or replaying text/tool events. Read stream.result.answer after iteration; .text() yields nothing on a cache hit. Cached error and cancelled outcomes use RUN_ERROR and RUN_CANCELLED; success and max_iterations use RUN_COMPLETED with their original status.

Active-run matches and fingerprint conflicts yield RUN_ERROR, with stream.result.meta["error_type"] equal to RunAlreadyActiveError or IdempotencyConflictError. Missing persistence yields RUN_ERROR during setup. These errors do not change the original run or its event log. Before using the ID for cancellation or correlation, consume the first event so the key has been resolved. The cached result is a summary, like run()'s cached result; it does not reconstruct the original steps.

examples/33_cost_cancel_idempotency.py exercises the cache hit and the conflict path against a persisted store.

resume, submit_tool_results, submit_input, submit_approval, submit_resume

The submit family resumes a paused run. Each one matches a pause status (see Pause and resume). All require persistence and the run to exist.

async def submit_tool_results(
    run_id: str, results: list[ToolResult], *, notifier=None,
    tenant_id: TenantScope = ...,    # omitted = UNSCOPED, unless strict_tenancy
    expected_pause_id: str | None = None,
) -> RunResult
async def submit_input(
    run_id: str, text: UserContent, *, notifier=None,
    tenant_id: TenantScope = ...,    # omitted = UNSCOPED, unless strict_tenancy
    expected_pause_id: str | None = None,
) -> RunResult
async def submit_approval(
    run_id: str,
    *,
    approved: bool | None = None,
    decisions: Mapping[str, ApprovalDecision] | None = None,
    rejection_reason: str | None = None,
    notifier=None,
    tenant_id: TenantScope = ...,    # omitted = UNSCOPED, unless strict_tenancy
    expected_pause_id: str | None = None,   # required when approving
) -> RunResult
async def submit_resume(
    run_id: str, *, notifier=None,
    tenant_id: TenantScope = ...,    # omitted = UNSCOPED, unless strict_tenancy
    expected_pause_id: str | None = None,
) -> RunResult
async def resume(
    run_id: str, *, tool_results=None, user_input=None, notifier=None,
    tenant_id: TenantScope = ...,    # omitted = UNSCOPED, unless strict_tenancy
    expected_pause_id: str | None = None,   # required when approving
) -> RunResult

Every method takes a keyword-only tenant_id=, like cancel_run and submit_note. Pass the tenant your request is authenticated for. The run must belong to exactly that tenant (None matches untenanted runs only), otherwise RunNotFoundError is raised, the same error as a missing run, so existence never leaks across tenants. The check runs before anything is written: no tool results, input, approval decisions, claim, status change or event. Omit it (or pass UNSCOPED) to skip the check; with Agent(strict_tenancy=True) omitting it raises TenantScopeRequiredError. See Multi-tenancy.

Each method also takes expected_pause_id=: the pause_id of the pause the caller was shown (RunResult.pause_id, PendingPause.pause_id or the run.paused event). If the run's current pause has a different id, the call raises PauseStatusMismatchError (reason="pause_id") and writes nothing, which stops a stale client from answering a newer pause. Approving requires it: approved=True, a decisions= batch that approves any call, and resume of an approval pause raise InputValidationError (a ValueError, code pause_id_required) without it. For tool results, input and plain continues it is optional, and under strict_tenancy=True omitting it prints a FutureWarning. Control flags such as approved and cascade must be real bool values; "false" or 1 raises ValueError. See Pause ids and stale submits.

await agent.submit_tool_results(
    run_id, results, tenant_id=current_tenant, expected_pause_id=pause_id,
)

submit_tool_results honours ToolResult.images, and submit_input / resume(user_input=) accept images in the text; they are checked and stored before the run is claimed, so a rejected image leaves the run paused. See Images.

submit_approval takes exactly one of approved= (whole batch) or decisions= (per call, ApprovalDecision.approve() / ApprovalDecision.deny(reason) keyed by call id). Client-target calls in the batch are never executed on the server; the run pauses again as waiting_client_tool for them. See Approval.

If the run was created with skills, these methods first compare them with the skills on this agent and raise SkillMissingError or SkillMismatchError (leaving the run paused) when they differ; resume_stream ends with a RUN_ERROR instead. See Resume contract.

Client results are checked before the run is claimed. Each result must name a pending call (matching id and tool name), the batch must cover every pending call once, payloads must be finite JSON or plain text within Limits.max_tool_result_bytes, and text must not contain NUL or lone surrogates. A bad submission raises InvalidToolResultError (or InputValidationError for an unpersistable value) and leaves the run paused and unclaimed. tool_result_transform runs on client results as well as server results. Passing tool_results to resume for an approval pause is a denial path: each result must be a failed result, and Dendrux replaces its text with its own denial wording so a client can never record a success for a call that did not run.

Shared resume errors: RunNotFoundError, RunNotPausedError, PauseStatusMismatchError, RunAlreadyClaimedError, RunAlreadyTerminalError, PersistenceNotConfiguredError. See Errors.

resume_stream

def resume_stream(
    run_id: str, *, tool_results=None, user_input=None, notifier=None,
    tenant_id: TenantScope = ...,    # omitted = UNSCOPED, unless strict_tenancy
    expected_pause_id: str | None = None,
) -> RunStream

Streaming variant of resume: returns a RunStream for the resumed run. Image errors in the submitted tool_results[i].images or user_input are raised from the first iteration, leaving the run paused. A rejected client result ends the attempt with a RUN_ERROR event whose run_result.meta["error_type"] is InvalidToolResultError; the durable run stays waiting with its pause id, so the client can correct the submission and resubmit. Do not read that event as the run having ended.

tenant_id= works as for the other methods, with one difference in how it fails. Without it, setup failures such as a missing run come back as a RUN_ERROR event. With it, a run outside the tenant raises RunNotFoundError from the first iteration, before any event, and a store error from the tenant check (a database outage, say) is raised the same way instead of becoming RUN_ERROR. The run is left untouched. Catch the exception around the async for. If the agent has no persistence, the check is skipped and the usual "requires persistence" RUN_ERROR follows.

cancel_run

async def cancel_run(
    run_id: str,
    *,
    cascade: bool = False,
    tenant_id: TenantScope = ...,    # omitted = UNSCOPED, unless strict_tenancy
) -> RunResult

Request cancellation. Works for paused runs (atomic CAS to cancelled, including paused), running runs (cooperative flag), batch run() calls and streamed runs this instance owns (in-process interrupt: the in-flight model call or tool is abandoned; run() returns CANCELLED with meta["interrupted"], a stream ends with RUN_CANCELLED carrying the partial answer), and terminal runs (no-op, never raises). Since 0.5.0 the same in-process interrupt also applies to a run that a submit_* or resume call is executing: that call returns a CANCELLED RunResult instead of raising CancelledError. Cancelling the task that is awaiting the call still raises. When the durable cancelled status could not be confirmed, meta["dendrux.cancel_persisted"] is False; check it before treating the cancellation as saved. Returns the persisted state at the time of the request; a non-terminal status with meta["cancel_requested"] == True means "requested, not yet stopped".

cascade=True also cancels every non-terminal descendant, deepest first and run_id last, over a point-in-time snapshot; per-run outcomes are in meta["cascade"]. tenant_id= requires the run to belong to exactly that tenant (None = untenanted) or RunNotFoundError is raised, and limits the cascade to the same tenant. A cancel that arrives after the run's last checkpoint does not stop a run that then completes successfully. Raises RunNotFoundError and PersistenceNotConfiguredError. See Cancellation.

request_pause

async def request_pause(
    run_id: str,
    *,
    cascade: bool = True,
    tenant_id: TenantScope = ...,    # omitted = UNSCOPED, unless strict_tenancy
    reason: str | None = None,       # at most 1000 characters
) -> PauseOutcome

Ask a running run to pause durably at its next checkpoint (the top of its next iteration, or between the approved calls of a resumed batch). It never preempts a model call or tool. The run becomes paused, run.paused is written with reason="requested", and the waiting run() or stream returns a paused result. Resume with submit_resume. A waiting_*, paused or terminal run is a no-op; cancellation wins over a pause. cascade=True (the default) flags every non-terminal descendant too, which is flag propagation only, not a resumable call stack: a paused child returns its paused result to the delegating tool, and resuming the child never rewrites what the parent recorded.

PauseOutcome has run_id, status_before, status_after and affected, a list of (run_id, outcome) with outcome requested, already_waiting, already_paused or terminal, deepest descendants first and run_id last. Raises RunNotFoundError, PersistenceNotConfiguredError and ValueError (bad reason). See Pause and resume.

submit_note

async def submit_note(
    run_id: str,
    text: str,
    *,
    metadata: Any = None,
    tenant_id: TenantScope = ...,    # omitted = UNSCOPED, unless strict_tenancy
) -> str

Leave a note for a non-terminal run; returns the note id. The runner delivers pending notes at the top of its next iteration as a user message NOTE FROM USER:\n{text} and writes an inbox.delivered event. Delivery is at-least-once; notes still pending when the run ends are listed in RunResult.meta["undelivered_notes"] and readable with RunStore.get_notes. text is non-empty and at most 32 KiB as UTF-8; metadata must be JSON and at most 8 KiB encoded, and is never shown to the model. Raises ValueError, NoteLimitExceededError, PersistenceNotConfiguredError, RunNotFoundError and RunAlreadyTerminalError. See Inbox notes.

Heartbeat

from dendrux import heartbeat, aheartbeat
 
def heartbeat() -> None            # sync, fire-and-forget, thread-safe
async def aheartbeat() -> None     # awaits the write

Call it from inside a tool that waits a long time (a human, a slow job) to refresh persisted progress and reduce the risk of a stale-run sweep reaping the run. Writes are best-effort and failures are logged; both helpers are no-ops outside a persisted run. heartbeat() works from async tools and from sync tools (which run in a worker thread) and coalesces calls less than a second apart; aheartbeat() must be awaited on the run's event loop and logs, never raises, on failure.

retry

async def retry(run_id: str, *, tenant_id=..., metadata=None, notifier=None, **kwargs) -> RunResult

Start a fresh run that reuses an existing terminal run's input and approximate prior context. Requires persistence. The source run must belong to tenant_id, otherwise RunNotFoundError is raised (the same error for a missing run and a foreign one; it was ReplayRunError or ValueError before 0.5.0). Skills the source run activated explicitly stay activated; if one is no longer attached, retry raises ValueError before creating a run.

close

async def close() -> None

Close directly owned MCP sources, the provider, and any agent-owned engine. Called for you by async with Agent(...). Leases on shared MCP connections (MCPRuntime views, MCPHost) are released, never closed; the application closes its runtime or host at worker shutdown. The shared global engine from the deprecated DENDRUX_DATABASE_URL fallback is not disposed because it is not agent-owned.

Wire answer display bound

RunResult.to_dict() keeps wire version 1 and includes answer_omitted. It is true when the answer cannot be safely projected within the wire display bound and answer is null. It is false for a successfully projected answer, including an empty string. Omission logs the run id with code wire_answer_omitted; trusted in-process results retain their original answer.

Where this fits

• Runs and the lifecycle: the status state machine these methods move through.
• RunStore: read back what a run produced.
• Errors: the exceptions the submit family raises.