# Agents & the loop

Configure an agent, manage its lifecycle, and inspect results.

`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.

```python
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](../loop-reliability/#reply-budgets-from-the-model) |
| `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](../prompt-caching/) |
| `tools` | `ToolPolicy()` | Registry-wide tool options applied to the registry the agent adopts. See [Tools](../tools/#registry-wide-tool-policy) |
| `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](../loop-reliability/#a-plan-the-model-keeps-seeing) |
| `max_parallel_subagents` | `4` | How many specialists one agent runs at the same time. `1` runs them one after another. See [Multi-agent](../multi-agent/#specialists-in-parallel) |
| `fallbacks` | `()` | `Fallback` entries to try, in order, when the primary provider fails transiently. See [Retries & fallback models](../retries-and-fallbacks/) |
| `progress` | `ProgressPolicy()` | When the run stream reports a slow model call with `WAITING` events. See [Streaming](../streaming/#show-that-a-slow-call-is-still-working) |

`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](#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](../loop-reliability/#noticing-a-loop) |
| `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](../retries-and-fallbacks/).

`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:

```python
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](../loop-reliability/#recovering-from-a-limit).

## Inject your own components

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

```python
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:

```python
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](../providers/#images-documents-and-audio).

## Inspect the result

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

```python
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](../multi-agent/) 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](../providers/#refusals-paused-turns-and-thinking-tokens)
for how each vendor's reasons map.

```python
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
```