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,                 # provider default for the model
        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 five top-level fields and groups everything else into four sub-policies. All are optional; these are the defaults. 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, the provider chooses for the model: 20,000 on Claude 4 models, 8,192 elsewhere. A reply cut off here ends the run with result.truncated true
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

Limits fields:

Field Default Purpose
max_iterations 50 Iteration limit; a reached guardrail returns a failed result. 0 means unlimited
max_context_tokens 150_000 Conversation trim threshold
max_result_chars 12_000 Tool results above this are kept out of the context: the model sees a head-and-tail preview and reads sections back with the built-in read_tool_result tool, which pages by lines and, for long lines, by characters
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

RetryPolicy fields:

Field Default Purpose
attempts 2 Total model calls for one step on 429, 5xx, timeouts, or connection loss; 1 disables retry
backoff_seconds 0.5 Base delay between attempts, doubled each time
call_timeout_seconds None Wall-clock limit around one attempt; unset, it is sized to the reply budget and never below DEFAULT_TIMEOUT_SECONDS (300 s)

ToolPolicy fields:

Field Default Purpose
default_timeout_seconds None Default for tools without their own timeout_seconds; None keeps the registry's default of 300 seconds
dedupe_calls False Identical repeated calls within one run return the first result

The sub-policies are frozen dataclasses; derive a variant with dataclasses.replace. The flat 0.3 names (AgentConfig(max_iterations=20), config.llm_max_attempts, ...) still work as keyword arguments and attributes but emit a DeprecationWarning on every use and are removed in 0.5. AgentConfig.from_dict() reads both shapes without warnings, so saved sessions load unchanged. See Migrating to 0.4.

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)

Inspect the result

RunResult has three 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 RunFailed / RunAwaitingInput / RunTruncated / RunCancelled

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.