Agents & the loop · HarnessX

Agent is the core agentic loop: LLM reasoning → tool execution → repeat, until the model finishes, an approval is required, or a runtime limit is reached.

import asyncio
from harnessx import Agent, AgentConfig, Limits

agent = Agent(
    config=AgentConfig(
        model="claude-sonnet-4-6",      # Choose a model available to your account.
        provider="anthropic",            # a built-in name or one passed to register_provider()
        max_tokens=None,                 # resolve the reply budget from the model's limits
        system_prompt="You are a helpful assistant.",
        temperature=None,                # omitted by default for safety
        limits=Limits(max_iterations=50),  # loop guardrail
    )
)

async def main():
    async with agent:
        result = await agent.run("What is 17 + 25?")
        if result.error:
            raise RuntimeError(result.error["message"])
        print(result.output)

asyncio.run(main())

AgentConfig keeps eight top-level fields and groups everything else into five sub-policies. All are optional. Only model, provider, and max_tokens may be passed positionally.

Field	Default	Purpose
`model`	`"claude-sonnet-4-6"`	Model id passed to the provider
`provider`	`"anthropic"`	`"anthropic"`, `"openai"`, `"gemini"`, `"openrouter"`, `"azure"`, or a name added with `register_provider()`
`max_tokens`	`None`	Reply token budget. Unset, it comes from the model's own limits: its output limit, capped at 32K and fitted to the context left. See Long-run reliability
`system_prompt`	`"You are a helpful assistant."`	System prompt
`temperature`	`None`	Sampling temperature (omitted by default for safety)
`limits`	`Limits()`	Budgets the harness enforces; fields below
`retry`	`RetryPolicy()`	Transient-failure handling for model calls; fields below
`prompt_cache`	`PromptCachePolicy()`	Caching of the stable prompt prefix; `None` disables it. See Prompt caching
`tools`	`ToolPolicy()`	Registry-wide tool options applied to the registry the agent adopts. See Tools
`planning`	`True`	Registers `write_todos` and `read_todos`, adds a line to the system prompt asking the model to plan work of three or more steps, and puts the current list in every request. See Long-run reliability
`max_parallel_subagents`	`4`	How many specialists one agent runs at the same time. `1` runs them one after another. See Multi-agent
`fallbacks`	`()`	`Fallback` entries to try, in order, when the primary provider fails transiently. See Retries & fallback models
`progress`	`ProgressPolicy()`	When the run stream reports a slow model call with `WAITING` events. See Streaming

Limits fields:

Field	Default	Purpose
`max_iterations`	`50`	Iteration limit. `0` means unlimited
`max_context_tokens`	`None`	Context size condensing works against. `None` uses the model's own window
`max_result_chars`	`12_000`	Tool results above this are kept out of the context. See Results too large to keep
`max_cost_dollars`	`None`	Cost ceiling, enforced when prices are supplied
`input_cost_per_m`	`None`	Application-supplied input price per million tokens
`output_cost_per_m`	`None`	Application-supplied output price per million tokens
`cache_read_cost_multiplier`	`0.1`	What a prompt token read from the cache costs, as a share of `input_cost_per_m`. Anthropic charges a tenth; use `0.5` for OpenAI
`cache_write_cost_multiplier`	`1.25`	What a prompt token written to the cache costs, as a share of `input_cost_per_m`. Use `1.0` for OpenAI
`loop_guard`	`LoopGuard()`	Warns the model when it keeps making the same tool calls and getting the same results. On by default; it adds a note to the tool result and never fails the run. See Long-run reliability
`max_truncation_recoveries`	`3`	How many times one run recovers from a reply cut off at its budget, each time with a doubled budget. `0` ends the run at the first cut
`final_answer_on_limit`	`True`	At `max_iterations`, asks the model once more, with no tools, for an answer. The run completes with `result.limited` true. `False` fails it with `MaxIterationsError`

RetryPolicy governs transient model failures and failover. Its fields, and the Fallback chain that goes with it, are on Retries & fallback models.

ToolPolicy fields:

Field	Default	Purpose
`default_timeout_seconds`	`None`	Default for tools without their own `timeout_seconds`; `None` keeps the registry's default of 25 seconds. Built-ins with their own limit, such as `run_bash`, keep it
`dedupe_calls`	`False`	Identical repeated calls within one run return the first result
`retry`	`None`	Default `ToolRetry` for tools without their own policy; automatic retry requires a safe or idempotent replay policy

Results too large to keep

A tool result longer than Limits.max_result_chars doesn't go into the conversation as it is. What happens depends on what it looks like.

If it's JSON with a list of records, the kind search APIs and database tools return, every record's text is cut to the same length until the whole result fits. The model still gets every record and most of what each one says. Text that was cut ends with "[... N more characters]", and the full result is saved so the model can read the rest. A search 16% over the limit comes through at about 86%, instead of a one-line summary per record.

Anything else, or a list too long to fit even with every record cut short, is saved to a file and the model gets a preview. The note above the preview says how much of the result it shows, asks the model to read what it needs before relying on it, and gives the exact call to do that. A list of records is previewed one line per record (title, link or id, and its first real sentence, skipping page navigation).

The model reads the rest with the built-in read_tool_result tool:

Argument	Default	Purpose
`result_id`	required	The tool-use id named in the notice
`offset`, `limit`	`0`, `2000`	Lines to return, for a result with many lines
`char_offset`, `max_chars`	`0`, `0`	Character position and size of one read. `0` means as much as fits under `max_result_chars`, so a result that only just spilled comes back whole
`pattern`	`""`	A case-insensitive regular expression. Returns the passages that match, with their character positions, instead of reading in order

A single read never returns more than fits under max_result_chars, so reading a saved result back never sends it to disk again.

The sub-policies are frozen dataclasses; derive a variant with dataclasses.replace. Invalid values are rejected when the config is constructed, not when a run starts.

To recover from a cut-off reply only once and fail at the step limit:

from harnessx import AgentConfig, Limits

config = AgentConfig(limits=Limits(
    max_iterations=30,
    max_truncation_recoveries=1,
    final_answer_on_limit=False,
))

See Long-run reliability.

Inject your own components

You can also inject your own pieces. Everything is a constructor parameter:

from harnessx import (
    Agent, AgentConfig, ConversationMemory, HookManager,
    MiddlewarePipeline, PermissionManager, ToolRegistry,
)

agent = Agent(
    config=AgentConfig(system_prompt="..."),
    tools=ToolRegistry(),               # your own registry
    subagents=[],                      # SubAgent specialist definitions
    memory=ConversationMemory(),        # your own memory
    permissions=PermissionManager(),    # your own permission policy
    hooks=HookManager(),                # lifecycle hooks
    middleware=MiddlewarePipeline(),    # request/response transforms
)

Conversation loop

agent.run() keeps the conversation in memory, so calling it repeatedly is a multi-turn chat:

async def chat(agent):
    while True:
        text = input("You: ").strip()
        if text in ("", "quit"):
            break
        print("Agent:", (await agent.run(text)).output)

agent.run() also takes a list of content blocks, to send an image, PDF or audio clip with the text. See Model providers.

Inspect the result

RunResult has predicates and a raiser, so a caller rarely compares status by hand:

result = await agent.run("Summarize the report")
if result.ok:             # status is COMPLETED
    print(result.output)
elif result.failed:       # status is FAILED; result.error has type and message
    print(result.error["message"])
elif result.needs_input:  # status is AWAITING_INPUT; result.pending lists PendingTool objects
    print([pending.call.name for pending in result.pending])

result.raise_for_status()  # returns result, or raises a RunError subclass

ok means the run completed with a whole answer. A reply cut off at the token budget completes but is not ok: result.truncated is true, stop_reason is "max_tokens", and raise_for_status() raises RunTruncated.

A failed run raises RunFailed, an approval wait RunAwaitingInput, and a cancelled run RunCancelled. Each exception carries .result.

One agent runs one turn at a time. agent.busy is True while a run is in progress and agent.closed is True once aclose() has started; starting a run in either state raises RuntimeStateError. Use async with agent: to close it.

Use Agent(subagents=[SubAgent(...)]) to declare specialists at construction. Each delegated task gets a fresh conversation and returns its result to the orchestrator. See multi-agent orchestration for configuration and recovery boundaries.

Refusals and the step limit

Each completed result that is not ok has its own predicate and exception:

Predicate	`stop_reason`	`raise_for_status()` raises
`result.truncated`	`max_tokens`, after recovery ran out	`RunTruncated`
`result.refused`	`refusal` or `safety`: the model declined or a filter stopped the reply	`RunRefused`
`result.limited`	`max_iterations`: the output is the model's account of where it got to	`RunLimitReached`

StopReason.REFUSAL marks a model that declines, and StopReason.PAUSE_TURN a server that pauses a long turn; the loop sends a paused turn back to resume it. See Model providers for how each vendor's reasons map.

async def answer(agent, prompt):
    result = await agent.run(prompt)
    if result.refused:
        return "The model declined: " + result.output
    if result.limited:
        return "Stopped at the step limit: " + result.output
    return result.raise_for_status().output