Tools · HarnessX

A tool has two parts: a Python handler that performs an operation, and a registered name, description, input schema, and execution policy that the agent uses to call it. Registration makes a handler available; it does not execute it.

Choose a registration style

Style	Example	Best use
Constructor function	`Agent(tools=[fetch_url])`	A small, explicit list of handlers
Constructor built-in name	`Agent(tools=["fetch_url"])`	Load a shipped tool without importing its handler
Constructor bundle	`Agent(tools=["web"])`	Load a built-in group
Later registration	`agent.tools.register_tool(fetch_url)`	Assemble tools after creating the agent
Decorator	`@agent.tools.register(...)`	Define a custom handler beside its registration
Built-in helper	`register_web_tools(agent.tools, ...)`	Configure and select tools in a bundle
Registry loading	`registry.load_builtin("web", ...)`	The same built-in helper options through one entry point
Tool definition	`Agent(tools=[definition])`	Reuse an explicit schema and execution policy

All of these inherit the agent's permission policy when permission is omitted. A normal Agent() defaults to ALLOW. A custom manager can instead default to ASK or DENY. A preconfigured ToolDefinition retains its explicit permission. Registration style does not choose the permission default.

Prefer functions for simple registration and helpers for configured capabilities. Bundle names are filesystem, bash, web, memory, planning, and all; a bundle can contain more tools than one standalone function. The Skill tool and the knowledge tools (search_concepts, read_concept, get_neighbors) are registered by Agent(skills=...) and Agent(knowledge=...), not by a bundle name; see Skills and Knowledge (OKF). The planning tools write_todos and read_todos are registered by default and switched off with AgentConfig(planning=False); see Long-run reliability.

write_todos takes the whole list every time it's called. Each item needs content, a short description of the step, and a status of pending, in_progress or completed. Extra keys on an item are fine. If the list doesn't fit that shape, the model gets a one-line error and nothing changes.

While the built-in write_todos is installed, the system prompt also gets one fixed line asking the model to plan anything with three or more steps, update the list in the same turn as the work, and just answer when it's done. planning=False removes the tools and the line. If you register your own write_todos, yours is used as it is and no line is added.

The current list is attached to the newest message of each request. Earlier copies stay where they were first sent, so the start of the conversation doesn't change between requests and the prompt cache keeps working.

Constructor registration and registration afterward

from harnessx import Agent
from harnessx.builtin.web import fetch_url

agent = Agent(tools=[fetch_url])

The constructor infers the tool name, schema, and description from the function. You can also register afterward, before starting the run:

from harnessx import Agent
from harnessx.builtin.web import fetch_url

agent = Agent()
agent.tools.register_tool(fetch_url)

Both expose fetch_url to subsequent model calls. Finish changing bindings before starting a run; mutating a registry during execution can invalidate an in-flight call. Use async with agent: or await agent.aclose() to close an agent when done.

A constructor list may mix functions, built-in names, and ToolDefinition objects. Passing an existing ToolRegistry attaches that same registry to an ordinary agent; it does not clone the registry. Declare subagent tools separately: each specialist task receives a fresh registry with copied schemas and metadata.

Direct function versus a registration helper

A bundle gives you both: the operation, and a helper that configures and registers it.

from harnessx import Agent
from harnessx.builtin.web import fetch_url, register_web_tools

agent = Agent()
registered = register_web_tools(agent.tools, include=["fetch_url"], max_length=20_000)
assert registered == ["fetch_url"]

Registering fetch_url directly exposes its url and max_length parameters to the model. Going through the helper with a non-default max_length hides that limit, so the model can set only url. Truncation caps the text returned, not the bytes downloaded.

The helper takes a registry, so pass it during setup rather than as an agent tool.

Permission defaults and precedence

Registration	Ordinary agent	Manager default ASK	Manager default DENY
Function, name, bundle, or helper; permission omitted	ALLOW	ASK	DENY
Explicit `permission=ALLOW`	ALLOW	ALLOW	ALLOW
Explicit `permission=ASK`	ASK	ASK	ASK
Explicit `permission=DENY`	DENY	DENY	DENY

A manager override for the tool name beats every row. Otherwise an explicit tool permission beats the manager default. DENY then blocks execution even against an earlier session grant, and a session grant satisfies ASK without another prompt.

ASK needs an approval mechanism: a plain agent denies ASK calls rather than prompting. Permissions & budgets covers managers, callbacks and durable approvals.

Decorators and schema inference

from harnessx import Agent, PermissionLevel

agent = Agent()

@agent.tools.register(permission=PermissionLevel.ALLOW, replay_policy="safe")
def calculate(a: int, b: int) -> int:
    """Multiply two numbers.

    Args:
        a: First number.
        b: Second number.
    """
    return a * b

The description before an Args: or other recognized section becomes the tool description, including multiple lines. An Args: section documents parameters. Type hints drive JSON Schema: primitive types, nullable unions, Literal, nested lists, and dictionaries with string keys are supported. Nullable parameters remain required unless they have a Python default. Unsupported annotations fail during registration; use register_with_schema for a custom contract. replay_policy accepts ReplayPolicy.SAFE, ReplayPolicy.IDEMPOTENT, ReplayPolicy.MANUAL, or the plain strings; ToolDefinition.replay_policy stores the string.

Sync and async handlers both work. Synchronous handlers run outside the event loop. A timeout stops waiting for a synchronous handler; Python cannot forcibly stop its worker thread or reverse an external effect.

Explicit schemas and reusable definitions

Use a schema when type inference cannot express the input contract:

from harnessx import Agent, ToolRegistry

async def label(value: str) -> str:
    return value.upper()

registry = ToolRegistry()
definition = registry.register_with_schema(
    name="label",
    description="Uppercase a short label.",
    input_schema={
        "type": "object",
        "properties": {"value": {"type": "string", "maxLength": 40}},
        "required": ["value"],
        "additionalProperties": False,
    },
    handler=label,
)
agent = Agent(tools=[definition])

name, description, input_schema, and handler are the positional parameters of register_with_schema; permission, concurrent, replay_policy, timeout_seconds, retry, retry_if_result, and replace are keyword-only. Registering a ToolDefinition copies its schema and metadata while retaining the handler. Registration options override only specified metadata; registering an existing definition does not silently reset its permissions or replay policy.

Configured built-in tools

Helpers support include, exclude, permission, and replace, plus options specific to the bundle. Unknown names, filters, and unsupported options fail.

For filesystem inspection, configure a root and only the operations you need:

from pathlib import Path
from harnessx import Agent
from harnessx.builtin import register_filesystem_tools

agent = Agent()
register_filesystem_tools(
    agent.tools,
    include=["read_file", "list_directory"],
    base_path=str(Path.cwd()),
    max_read_bytes=1_000_000,
    max_directory_entries=1_000,
)

The bundle is read_file, write_file, edit_file, delete, glob, grep, list_directory and generate_file. All of them go through the same containment layer when base_path is set.

edit_file replaces an exact string and refuses an ambiguous match rather than editing the wrong occurrence. That refusal is what makes it safer than rewriting a file whole. An agent building a long artifact writes the first chunk, then grows it a section at a time. It never spends its reply budget reproducing bytes it already wrote. delete requires recursive=True for a non-empty tree and will not remove the root itself.

Reads and writes differ in whether they may overlap:

Run concurrently	Run one at a time
`read_file`, `glob`, `grep`, `list_directory`	`write_file`, `edit_file`, `delete`, `generate_file`

Within a turn, concurrent calls run together. The others run one at a time, in the order the model made them, alongside the concurrent ones. So a batch of searches runs in parallel, a write_todos in the same turn doesn't hold them up, and two edits to the same file never interleave. Results go back to the model in the order it made the calls, and tool_call_start events reach the stream in that order too, whichever call finishes first. Serial tools also take turns across an agent and its specialists.

The root must exist. Scoped operations reject symlinks in files and parent directories using descriptor-relative, no-follow access on POSIX. This is an application boundary, not an OS sandbox for arbitrary code: hard links, mounts, and other processes relocating directories require stronger isolation. Without base_path, filesystem tools use host filesystem access under process permissions.

Configured filesystem limits are hidden from model input. Existing files require overwrite=True; writes publish atomically. Writes and downloadable-file creation remain non-concurrent when registered through filesystem helpers, regardless of permission. Direct standalone functions use ordinary registration metadata, so choose concurrent=False explicitly if needed.

Download copies use configured output_dir, otherwise AGENT_OUTPUT_DIR or ./output. Keep the output directory application-owned. Filesystem reads and writes run outside the event loop.

Other options include a sandbox for the bash helper and memory stores for the memory helper. Agent(tools=["bash"], sandbox=sandbox) binds a caller-owned sandbox as the bash built-in's engine, so the model's shell commands run inside it; the registry exposes the binding as agent.tools.sandbox. A sandbox that owns its filesystem, such as OpenShell, also takes the file tools, and base_path does not apply there. The local Sandbox leaves the file tools on the host. Single-tool loading accepts the corresponding bundle options:

from pathlib import Path
from harnessx import ToolRegistry

registry = ToolRegistry()
registry.load_builtin("read_file", base_path=str(Path.cwd()))

Registry-wide tool policy

AgentConfig(tools=ToolPolicy(...)) carries registry-wide options to the registry the agent adopts, including a plain list of functions:

from harnessx import Agent, AgentConfig, ToolPolicy
from harnessx.builtin.web import fetch_url

agent = Agent(
    config=AgentConfig(tools=ToolPolicy(default_timeout_seconds=60, dedupe_calls=True)),
    tools=[fetch_url],
)
assert agent.tools.policy.default_timeout_seconds == 60

Precedence, highest first: a value given to the ToolRegistry(default_timeout_seconds=, dedupe_calls=) constructor, the agent's config.tools, then the built-in default (DEFAULT_TOOL_TIMEOUT_SECONDS, 25 seconds, and no deduplication). Tools registered with their own timeout_seconds are never changed, and neither are the built-ins that declare their own limit (see below). registry.policy shows the effective values. With dedupe_calls=True, identical repeated calls within one run return the first result.

ToolPolicy(retry=ToolRetry(...)) also supplies a default for tools without their own retry policy. An explicit ToolRegistry(retry=...) takes precedence over that agent setting; an explicit per-tool retry stays in place. With no configured policy, the tool uses ToolRetry().

Retry a tool

ToolRetry is separate from the model's RetryPolicy, and a handler signals a repeatable failure by raising TransientToolError. Retry applies only to safe and idempotent tools. See Retries & fallback models.

Replacement, timeouts, and replay

Duplicate tool names fail by default. Use replace=True to deliberately replace a binding during setup. Inspect get_tool(name), list_tools(), or get_tool_params(); the last returns canonical provider-neutral tool schemas.

Ordinary function registrations default to concurrent=True, a timeout of DEFAULT_TOOL_TIMEOUT_SECONDS (25 seconds) unless a ToolPolicy changes the default, and replay_policy="manual". Configure these with register_tool or the decorator. A ToolDefinition that never set its own timeout takes the same default. Built-in helpers do not automatically mark fetching or other operations replay-safe.

A tool that has not answered in 25 seconds is more often stuck than busy, and the model does better being told so than waiting. Give a tool that legitimately takes longer its own limit, such as a heavy SQL query:

agent.tools.register_tool(query_postgres, timeout_seconds=120)

Built-ins that can run long declare their own limit, which ToolPolicy does not change:

Tool	Timeout
`run_bash`	300 seconds; the command's own `timeout` argument ends it first
MCP tools	300 seconds
Subagents	`SubAgent(timeout_seconds=...)`, 300 seconds by default
File tools in a sandbox that owns the filesystem	300 seconds; the first call may start the sandbox
`fetch_url`	60 seconds
`ask_user`	One hour

Other built-ins, such as file tools on the host, planning and memory, use the 25-second default.

In a durable runtime, a manual tool that times out stops the run awaiting input (result.needs_input), because its outcome is unknown. A safe or idempotent tool is retried under its ToolRetry. A direct run answers the timeout instead; see below.

Durable execution uses manual to stop for recovery when an interrupted operation's outcome is unknown. Select safe only when repeating an operation is safe, or idempotent when the integration actually enforces idempotency. Neither setting provides exactly-once external effects on its own.

Timeouts in a direct run

A tool that times out in a direct agent.run() returns an error result to the model: the tool timed out, may have partly run, and its effects should be checked before calling it again. The run goes on. A durable runtime stops for recovery instead, because it can ask whether the call took effect. One failing call in a concurrent batch does not cancel the others.

Tool names the model gets wrong

An exact registered name always wins. A name is repaired only when it differs by formatting and exactly one tool matches:

letter case: Search_Web for search_web
separators: search-web, searchWeb, search.web, or search web
the known prefixes functions., tools., and default_api:

A typo, or an alias that matches two tools, is never dispatched. The model gets an error result that suggests up to three close names and lists the available tools, and must call again with the right one. Spelling similarity only supplies those suggestions; it never selects a tool.

Malformed arguments are handled the same way. If the arguments aren't a JSON object, the model sees the parse error and can fix the call. That includes arguments that get damaged in the stream, either garbled or cut off halfway (a file's content stopping mid-line, say). The tool doesn't run on a partial call. The model gets an error result and sends the call again with the full JSON. Arguments that don't match the tool's schema get a one-line error naming the field that's wrong. See Long-run reliability.

Let the agent ask the user

A model given an ambiguous request picks one reading and commits to it. register_ask_user_tool gives it an ask_user tool, so it can ask one focused question before doing the work. The tool does not know how to reach a person; your application does. Pass ask, a coroutine that receives the question and a list of suggested answers, puts them in front of someone, and returns the answer:

import asyncio
from harnessx import Agent
from harnessx.builtin import register_ask_user_tool

async def ask(question: str, choices: list[str]) -> str | None:
    if choices:
        question += f" ({' / '.join(choices)})"
    answer = await asyncio.to_thread(input, f"{question}\n> ")
    return answer or None

agent = Agent()
register_ask_user_tool(agent.tools, ask)

choices is empty for a free-text question. Return None when nobody can answer: the model is then told to proceed on the most likely reading and to state that assumption at the start of its reply, rather than wait. A web prompt or a chat message works the same way, as long as ask returns the answer.

The tool is registered with ALLOW, concurrent=False, a safe replay policy and a one-hour timeout, since it waits on a person. permission=, timeout_seconds= and replace= change those. It is not part of any bundle.

Calling a function versus executing an agent tool

await fetch_url(url) is an ordinary Python call. It bypasses the registry's schema validation, permission manager, timeout, and durable execution records.

await registry.execute(call, permissions=agent.permissions) validates the input, checks permission, and applies the registered timeout. Without a manager it uses the ordinary ALLOW default for unspecified policies; explicit ASK without a callback and explicit DENY return the error result "Permission denied". It does not provide durable replay or the agent's complete middleware/event pipeline.

agent.run(...), agent.run_stream(...), and AgentRuntime use the shared engine for model-requested tools. They apply execution checks before dispatch, with persisted approvals and recovery when using a durable runtime. Handler exceptions usually become error tool results; runtime/storage errors may fail the run.