HarnessX

A provider-agnostic, library-first Python toolkit for building production-grade LLM agents.

Try harness-web to launch all examples from a browser, review approvals and live output, or use a separate general chat workspace.

Rather than locking developers into rigid state-machine graphs or opaque persona prompts, harnessx gives you a composable set of building blocks: real-time streaming, typed tool registration with schema inference, lazy-loaded skills, multi-agent delegation, 4-tier memory, sandboxed code execution, Model Context Protocol (MCP) tool bridges, native LangSmith tracing, and evaluation suites.

Why HarnessX?

• No Graph Boilerplate: Write simple async Python functions instead of complex state graphs.
• Provider-Agnostic: First-class support for Anthropic (Claude 3.5/3.7/Sonnet/Opus) and OpenAI (GPT-4o/GPT-5).
• Streaming-First: First-class typed event stream (run_stream) for WebSocket and SSE frontends.
• Lazy Instruction Packs (Skills): Load specialized guidelines only when needed via YAML-frontmatter SKILL.md packs.
• Guardrails & Execution Backends: Budget and iteration limits, host subprocess execution, and Docker or macOS Seatbelt access isolation.
• MCP Native: Connect to any Model Context Protocol server (stdio subprocess or remote SSE) in 3 lines of code.
• Full Observability & Evals: Zero-overhead lifecycle hooks, middleware transforms, native LangSmith tracing, and benchmark evaluations.
• Typed Decisions: Optional Jev SDK for Choice routing/classification, Score rubrics, and Noul evidence checks. See the decision SDK guide.

• Setup
• 1. Creating an agent
• 2. Registering tools
• 3. Streaming agents
• 4. Multi-agent orchestration
• 5. Skills
• 6. Memory
• 7. Permissions and guardrails
• 8. Hooks and middleware
• 9. MCP servers
• 10. Session persistence
• 11. Shipping a web app
• 12. Evaluations with LangSmith
• API reference (quick)

Setup

The distribution and Python import are both named harnessx: from harnessx import Agent. Upgrading from datagol-agent-harness? Update the import to harnessx (there is no compatibility alias) and see the changelog.

# From PyPI
pip install harnessx

# With optional extras (LangSmith, OpenAI, Azure OpenAI, Jev, MCP, Docker, or everything)
pip install "harnessx[langsmith]"
pip install "harnessx[openai]"
pip install "harnessx[azure]"
pip install "harnessx[jev]"
pip install "harnessx[all]"

# For local development (editable)
pip install -e ".[all]"

Set your key (a root .env is auto-loaded by the examples package):

ANTHROPIC_API_KEY=sk-ant-...
# or, for OpenAI-backed agents:
OPENAI_API_KEY=sk-...

Run examples as modules from the repo root (not as plain scripts):

python -m examples.simple_chat

See the examples guide for every entry point, optional dependencies, and service requirements. For a first run without API keys, use python -m examples.skills_demo or python -m examples.flight_recorder --output incident.hx.

Agent() and examples.simple_chat use Anthropic by default. To choose Anthropic, OpenAI, Gemini, or OpenRouter, use provider chat: python -m examples.provider_chat --provider openai --model YOUR_MODEL_ID. See the provider setup guide for API keys, optional dependencies, and streaming options.

The standalone decision SDK provides typed Jev assessments alongside agents. Try its offline examples with python -m examples.jev_routing --min-confidence 0.8, python -m examples.jev_classification, or python -m examples.jev_answer_review. Live calls require the jev extra, TYPESAFE_API_KEY, and an explicit --live flag.

For 0.3 API changes, see the migration guide.

1. Creating an agent

Agent is the core agentic loop: LLM reasoning → tool execution → repeat, until the model stops with end_turn.

import asyncio
from harnessx import Agent, AgentConfig

agent = Agent(
    config=AgentConfig(
        model="claude-sonnet-4-6",      # or "gpt-4o", "anthropic/claude-3.7-sonnet"
        provider="anthropic",            # "anthropic", "openai", "gemini", "openrouter", or "azure"
        system_prompt="You are a helpful assistant.",
        max_tokens=8192,
        max_iterations=50,               # loop guardrail
        temperature=None,                # omitted by default for safety
    )
)

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 fields (all optional, these are the defaults):

Prompt caching

Every iteration of the loop resends the system prompt, the tool list, and the conversation so far. Caching is on by default: the engine computes a stable prefix key and marks where the reusable prefix ends, and each provider translates that into its vendor's mechanism. Caching is an optimization, never a correctness dependency, so a provider whose vendor rejects the request runs it uncached.

from harnessx import Agent, AgentConfig, PromptCachePolicy

# Default: on, vendor default lifetime, history breakpoint after the last turn.
agent = Agent(config=AgentConfig(model="claude-sonnet-4-6"))

# Tuned: a one-hour lifetime and per-tenant separation of identical prompts.
agent = Agent(config=AgentConfig(
    provider="gemini",
    model="gemini-3.6-flash",
    prompt_cache=PromptCachePolicy(ttl_seconds=3600, key_salt="tenant-a"),
))

# Off, for every provider.
agent = Agent(config=AgentConfig(prompt_cache=None))

Cache hits show up in the usual usage fields: RunResult.usage.cache_read_input_tokens and cache_creation_input_tokens. Every LLM_REQUEST hook carries data["prefix_key"], so an observer can assert that consecutive iterations share one key. A prefix that drifts, for example a timestamp in the system prompt or tools registered in a different order, silently defeats every vendor's cache. Custom providers receive the hint as an optional cache keyword on create and stream; ignoring it is fine, honoring it means failing open.

Azure OpenAI

Install the extra (pip install "harnessx[azure]"; the openai package covers Azure). Azure reuses the OpenAI translation layer through the AsyncAzureOpenAI client, so completion and streaming both work. On Azure the model you pass is the deployment name, and both an endpoint and an API version are required for provider-created clients. There are two ways to supply credentials — the harness never loads a .env file itself, so your app is responsible for populating the environment or passing values in code.

1. Config-driven, credentials from the environment (AZURE_OPENAI_ENDPOINT, AZURE_OPENAI_API_KEY, OPENAI_API_VERSION):

from harnessx import Agent, AgentConfig

# env: AZURE_OPENAI_ENDPOINT, AZURE_OPENAI_API_KEY, OPENAI_API_VERSION
agent = Agent(config=AgentConfig(provider="azure", model="<your-deployment>"))

2. Explicit construction + injection (pass keys/endpoint in code, or hand in a fully preconfigured client):

from harnessx import Agent, AgentConfig
from harnessx.providers.azure_openai import AzureOpenAIProvider

provider = AzureOpenAIProvider(
    azure_endpoint="https://<resource>.openai.azure.com",
    api_key="<key>",
    api_version="2024-10-21",        # use an API version your resource supports
    azure_deployment="<deployment>", # optional; overrides AgentConfig.model per request
)
agent = Agent(
    config=AgentConfig(provider="azure", model="<deployment>"),
    provider=provider,
)

Token-based auth is supported: omit api_key and pass azure_ad_token or azure_ad_token_provider — they are forwarded straight to AsyncAzureOpenAI. Token counting is approximate for Azure, because a deployment name does not identify the underlying tokenizer.

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=[],                      # declarative SubAgent specialists
    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)

2. Registering tools

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

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 currently include filesystem, bash, web, memory, and all; a bundle can contain more tools than one standalone function.

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

These imports serve different purposes:

from harnessx.builtin.web import fetch_url           # operation
from harnessx.builtin.web import register_web_tools  # registration/configuration

fetch_url is an async Python function. register_web_tools installs the web bundle into a supplied registry and returns the list of registered names. The web bundle currently contains fetch_url.

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

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

This uses the same HTTP-fetch implementation, with configured response truncation. With a non-default max_length, the helper creates a wrapper whose model input is only url; the model cannot override that configured limit. Direct fetch_url registration exposes its url and optional max_length function parameters. With the helper's default max_length=500_000, it currently registers the direct function, so that parameter remains visible. Truncation limits the returned text, not bytes downloaded from the network.

Both approaches inherit permissions. Configuration can still change the handler and schema; equivalent permissions do not imply identical capabilities.

Do not pass register_web_tools as an agent tool: its argument is a registry, not an input for the model. Call the helper during application setup.

Permission defaults and precedence

A manager override for the tool name takes precedence over every row. Otherwise, an explicit tool permission takes precedence over the manager default. After resolving the effective level, DENY blocks execution even if there is a previous session grant. A session grant can satisfy ASK without another callback.

For example, register an explicit policy or override it later:

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

agent = Agent()
agent.tools.register_tool(fetch_url, permission=PermissionLevel.ALLOW)
agent.permissions.set_permission("fetch_url", PermissionLevel.DENY)

ASK needs an approval mechanism

A plain agent does not display a prompt automatically. Direct execution and ordinary runs deny ASK calls when no approval callback is provided.

For a terminal application:

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

agent = Agent(tools=[fetch_url], permissions=CliPermissionManager())

CliPermissionManager defaults to ASK. That includes built-in tools registered without an explicit policy. Alternatively, supply a synchronous or asynchronous callback. This complete callback example approves only one known URL:

from harnessx import Agent, PermissionLevel, PermissionManager, ToolCall, ToolDefinition
from harnessx.builtin.web import fetch_url

async def approve(call: ToolCall, definition: ToolDefinition) -> bool:
    return call.name == "fetch_url" and call.input.get("url") == "https://example.com"

agent = Agent(
    tools=[fetch_url],
    permissions=PermissionManager(
        default_level=PermissionLevel.ASK,
        approval_callback=approve,
    ),
)

A UI can implement the same callback by awaiting the user's decision. Providing only approval_callback leaves the default at ALLOW; also select ASK for the tools that should request approval. Callbacks must return an actual boolean.

Durable AgentRuntime runs persist ASK requests and return awaiting_input. Use the returned execution key with runtime.approve(...), then resume the run; see the recovery documentation. This persisted approval behavior applies to the outer delegation call, not to the non-durable specialist's internal tools. MCP tools retain their explicitly configured per-server permission policy.

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.

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])

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 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. 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()))

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 300-second timeout, and replay_policy="manual". Configure these with register_tool or the decorator. Built-in helpers do not automatically mark fetching or other operations replay-safe.

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.

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

3. Streaming agents

Agent.run_stream() yields typed events and exposes a final RunResult. Use its async context manager to close or cancel an abandoned stream.

import asyncio
from harnessx import AgentConfig, Agent, RunEventType

agent = Agent(config=AgentConfig(system_prompt="You are helpful."))

async def main():
    async with agent, agent.run_stream("Tell me a story") as stream:
        async for event in stream:
            if event.type == RunEventType.TEXT_DELTA:
                print(event.data, end="", flush=True)
            elif event.type == RunEventType.TOOL_CALL_START:
                print(f"\n> calling {event.data.name}({event.data.input})")
            elif event.type == RunEventType.TOOL_RESULT:
                if event.data.is_error:
                    print(f"  tool error: {event.data.content}")
            elif event.type == RunEventType.TURN_COMPLETE:
                print()

asyncio.run(main())

Event types include TEXT_DELTA, TEXT_COMPLETE, TOOL_CALL_START, TOOL_CALL_COMPLETE, TOOL_RESULT, THINKING_DELTA, TURN_COMPLETE, ERROR, ATTEMPT_RESET, and RUN_RESULT. Durable runs also emit approval and recovery events. Text deltas are provisional; discard an interrupted attempt's text when ATTEMPT_RESET arrives.

Usage stats accumulate on the guardrails engine:

agent.guardrails.usage_summary
# {'iterations': 3, 'input_tokens': 5120, 'output_tokens': 640, 'estimated_cost': '$0.0250'}

4. Multi-agent orchestration

Declare specialists when creating an orchestrator. HarnessX handles creating, running, and closing the specialist for each task; you do not write or register delegation wrappers.

import asyncio
from harnessx import Agent, AgentConfig, SubAgent
from harnessx.builtin.web import fetch_url

async def main():
    async with Agent(
        config=AgentConfig(
            system_prompt="Delegate research when useful, then synthesize the findings.",
        ),
        subagents=[
            SubAgent(
                name="researcher",
                description="Read public URLs and summarize their evidence.",
                config=AgentConfig(
                    system_prompt="Read the supplied URLs and cite your sources.",
                    max_iterations=5,
                ),
                tools=[fetch_url],
            ),
        ],
    ) as orchestrator:
        result = await orchestrator.run("Read https://example.com and summarize it.")
        if result.status != "completed":
            raise RuntimeError(str(result.error or result.pending))
        print(result.output)

asyncio.run(main())

Definition and routing

SubAgent is a definition, not a running agent. Required fields:

• name: 1–55 letters, digits, underscores, or hyphens, starting with a letter or underscore.
• description: tells the orchestrator when this specialist is useful.
• config: an explicit AgentConfig; its system prompt instructs the specialist.

Optional fields:

• tools: the same functions, built-in names, definitions, or registry formats accepted by Agent.
• skills: a list of skill paths, loaded separately for each task.
• permission: permission to delegate; omitted means inherit the parent's manager.
• timeout_seconds: whole-task timeout, default 300 seconds, including child model and tool calls.

The SDK exposes delegate_<name>(task: str) to the model internally. The schema and description tell the orchestrator how to request a specialist. Do not also register a tool with that name; collisions and duplicate specialist names fail during construction. No separate orchestrator class is necessary.

The model decides whether to delegate. Naming specialists does not force a fixed sequence; write workflow code when you need a mandatory sequence. Use examples/multi_agent.py for research, code review, and scoped file analysis.

Context, resources, and permissions

Each delegated task gets fresh memory, configuration, and a tool registry. It sees only its own prompt and the assigned task. Include the code, question, URLs, paths, or other context it needs in the task. The child's model/provider comes from its explicit config; it does not inherit the parent's model or injected provider.

Handler functions remain shared application objects: separate conversations do not isolate files, databases, closures, or arbitrary Python code. Child-owned providers and memory close on completion, failure, and cancellation. Skills may add their own tool and prompt catalog; do not also register a Skill tool.

The parent's permission manager is shared, including its defaults, name overrides, session grants, and approval callback. Generated delegation permission is separate from each child tool's permission. Approving delegation does not approve its tools. The same tool-name override applies in parent and child. An ordinary manager defaults to ALLOW; use explicit permissions or another manager to change that policy.

Parent tools, hooks, middleware, extensions, MCP connections, and sandboxes are not automatically copied. The generated call uses the parent's normal tool pipeline; the child uses its own ordinary engine. Use the supported manual-wrapper pattern for advanced child dependencies and their lifecycle management. It remains useful for a specialist with custom tracing extensions or an injected provider.

Results, streaming, and limits

The child returns its completed output to the orchestrator, which continues and synthesizes a response. There is no conversation handoff. Non-completed child statuses and child failures become delegation errors; partial output is not reported as successful specialist work.

Generated delegation tools are non-concurrent and use manual replay policy. The current engine serializes a tool batch containing such a delegation. Both run and run_stream support the same behavior. Streaming reports ordinary tool_call_start, tool_call_complete, and tool_result events for delegation; child text does not appear as parent answer text.

Each specialist has its own iteration and cost limits. Parent RunResult.usage and guardrail totals remain parent-only, not a combined team budget. Set child limits explicitly. A timeout cancels waiting and closes child resources, but cannot reverse external effects or forcibly stop a synchronous tool thread.

Durable execution boundary

AgentRuntime persists the outer delegation as an ordinary manual-replay tool call. Committed results are reused on recovery. An interrupted delegation with an unknown outcome stops for manual recovery rather than automatically starting a new specialist. Live model reruns are not deterministic.

The child uses an ordinary, non-durable run. Child progress, internal approvals, intermediate artifacts, and conversations are not independently persisted by the parent runtime. A child ASK tool uses the approval callback and fails closed without one; it does not create a resumable approval in the parent runtime. Treat output referring to child-local artifacts accordingly; durable child-artifact transfer is not supplied by this API.

Rebind the same subagents definitions when restoring sessions or constructing versioned runtime factories, just as you rebind tools. Session snapshots do not serialize Python handlers. Keep definition changes under the application's normal versioned deployment controls.

Independent child recovery, persistent specialist conversations, nested declarative teams, handoffs, and child-token streaming are outside this initial API.

5. Skills

A skill is a markdown file with YAML frontmatter — a way to ship reusable instruction packs that load lazily: the agent only sees each skill's name + description in its system prompt, and pulls the full body into context by calling the Skill tool when a request matches.

Authoring a skill

skills/commit-message/SKILL.md:

---
name: commit-message
description: Write a conventional commit message for a described change.
---

You write commit messages. Rules:
- Conventional Commits format: type(scope): subject
- Subject in imperative mood, <= 72 chars, no trailing period
- Add a body only when the "why" isn't obvious from the subject

A path can be a folder containing SKILL.md, or a markdown file directly.

Loading skills into an agent

from harnessx import Agent, AgentConfig

agent = Agent(
    config=AgentConfig(
        system_prompt=(
            "You are an engineering assistant. When a request matches a "
            "skill in <available-skills>, call the Skill tool first to load "
            "its instructions, then follow them."
        ),
    ),
    skills=["./skills/commit-message", "./skills/code-review"],
)

The skills= argument accepts a list of paths or a SkillManager:

from harnessx import SkillManager

manager = SkillManager.from_paths(["./skills/commit-message"])
agent = Agent(config=AgentConfig(system_prompt="..."), skills=manager)

for s in agent.skills.list():
    print(s.name, "-", s.description)

Reacting to skill invocation

A SKILL_INVOKED hook fires when the model loads a skill:

from harnessx import HookContext, HookEvent

async def on_skill(ctx: HookContext):
    if ctx.data.get("found"):
        print(f"loaded skill: {ctx.data['skill']} ({ctx.data['body_chars']} chars)")

agent.hooks.on(HookEvent.SKILL_INVOKED, on_skill)

Full demo: examples/skills_agent.py, with sample skills under examples/skills/.

6. Memory

Three layers, all optional:

Conversation memory (default)

Every agent has a ConversationMemory that stores the message list and trims itself when approaching the context limit. You rarely touch it directly, but you can:

agent.memory.get_messages()        # current message list
agent.memory.set_messages(saved)   # restore

Agent memory — markdown notes on disk

AgentMemory persists titled, tagged notes as markdown files. You expose it to the model through tools you write:

from harnessx import AgentMemory, PermissionLevel

memory = AgentMemory(storage_dir=".agent_memory/agent")

@agent.tools.register(permission=PermissionLevel.ALLOW)
async def save_memory(title: str, content: str, tags: str = "") -> str:
    """Save a note to persistent memory.

    Args:
        title: Short title for the memory.
        content: The information to remember.
        tags: Comma-separated tags.
    """
    tag_list = [t.strip() for t in tags.split(",") if t.strip()]
    filename = memory.save(title, content, tags=tag_list)
    return f"Saved '{title}' -> {filename}"

@agent.tools.register(permission=PermissionLevel.ALLOW)
async def search_memory(query: str = "") -> str:
    """Search saved memories by keyword."""
    results = memory.search(query=query)
    return "\n".join(f"- {r['title']} [{r['filename']}]" for r in results) or "None found."

Long-term memory — structured facts

LongTermMemory stores categorized facts with ids:

from harnessx import LongTermMemory

long_term = LongTermMemory(storage_dir=".agent_memory/long_term")
fact_id = long_term.save("User prefers dark mode", category="preference")
long_term.search(query="dark mode")
long_term.update(fact_id, "User prefers light mode")

Tell the agent about these tools in the system prompt and when to use them ("proactively save useful information"). A complete two-layer example is in examples/memory_agent.py.

7. Permissions and guardrails

Permission levels

Tools use ALLOW (execute), ASK (request approval), or DENY (reject). Omitting the permission inherits the manager's default, which is ALLOW. This applies to constructor tools, later registrations, and direct ToolRegistry.execute() calls without a custom manager. Built-in helpers also inherit the manager default; file writes, shell commands, and URL fetching use ALLOW under a default agent. Select ASK explicitly when approval is required. Explicit MCP server policies remain in effect. Per-tool overrides take precedence over explicit registration permissions and the manager default. DENY also overrides a previous session grant.

from harnessx import PermissionLevel

# At registration
@agent.tools.register(permission=PermissionLevel.ALLOW)
def add(a: int, b: int) -> int: ...

# Or override later
agent.permissions.set_permission("run_bash", PermissionLevel.ASK)
agent.permissions.grant_session("read_file")   # pre-approve for this session

Direct execution handles ASK through a sync or async callback. Without a callback it denies the call; the SDK never implicitly reads stdin.

from harnessx import Agent, PermissionLevel, PermissionManager, ToolCall, ToolDefinition

async def approve(call: ToolCall, definition: ToolDefinition) -> bool:
    return await approval_ui.request(call.name, call.input)

agent = Agent(permissions=PermissionManager(
    default_level=PermissionLevel.ASK,
    approval_callback=approve,
))

Setting default_level=PermissionLevel.ASK requests approval for tools without an explicit permission. Supplying only approval_callback leaves the default at ALLOW; the callback handles tools explicitly configured as ASK.

Terminal apps can explicitly use Agent(permissions=CliPermissionManager()) with CliPermissionManager imported from harnessx; this opt-in manager defaults to ASK. Durable runtimes persist approval requests; use runtime.approve(...) and runtime.resume(...).

Guardrails

from harnessx import Agent, AgentConfig

agent = Agent(config=AgentConfig(
    max_iterations=20,
    max_context_tokens=100_000,
    max_cost_dollars=0.50,
    input_cost_per_m=3.00,   # application-supplied prices, not a live price lookup
    output_cost_per_m=15.00,
))
agent.guardrails.usage_summary

Limits are validated at construction. A reached guardrail returns a failed RunResult with error details. max_iterations=0 means unlimited.

8. Middleware, hooks, and extensions

An extension can install middleware with ctx.add_middleware(...). Use standalone middleware for one focused transformation; use an extension to reuse a complete feature or manage its resources and saved state.

from dataclasses import replace
from harnessx import Agent, Extension, ExtensionContext, Middleware, ToolCall

class NormalizeSearch(Middleware):
    async def before_tool_execution(self, call: ToolCall) -> ToolCall:
        query = call.input.get("query")
        if call.name == "search" and isinstance(query, str):
            return replace(call, input={**call.input, "query": query.strip()})
        return call

class SearchNormalization(Extension):
    name = "search_normalization"

    def install(self, ctx: ExtensionContext) -> None:
        ctx.add_middleware(NormalizeSearch())

agent = Agent(extensions=[SearchNormalization()])
# Register your search tool separately, then run inside `async with agent:`.

Read the detailed guides:

• Middleware and hooks: all four stages, ordering, complete examples, streaming behavior, errors, retries, and observation boundaries.
• Extensions: when to use an extension, composing middleware with tools and hooks, lifecycle callbacks, ownership, persistence, and built-ins.

Hook data is event specific. LLM_REQUEST carries message_count, tool_count, and prefix_key, the prompt-cache key for that request (None when caching is disabled). AGENT_END carries the final RunResult.

Mandatory permissions and effect checks belong in the execution path. Hook failures are suppressed, and terminal extension observers are best effort. Result middleware can change what the model receives without erasing raw data already retained by durable execution or recording.

9. MCP servers

MCPManager connects to Model Context Protocol servers (stdio subprocess or remote SSE) and registers their tools into your agent's registry, so MCP tools look exactly like native tools to the model.

from harnessx import Agent, AgentConfig, MCPManager

mcp = MCPManager()
await mcp.connect(
    "filesystem",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
)
# or a remote server:
# await mcp.connect("remote", url="http://localhost:8000/sse")

agent = Agent(config=AgentConfig(system_prompt="..."), mcp=mcp)
mcp.register_tools(agent.tools)

print((await agent.run("What's in /tmp?")).output)
await mcp.disconnect_all()

See examples/mcp_agent.py for the interactive version.

10. Session persistence

Save and restore an agent's conversation:

session_id = await agent.save_session()            # -> writes .agent_sessions/

# later, or in another process:
from harnessx import Agent
async with await Agent.load_session(session_id) as restored:
    print((await restored.run("Where were we?")).output)

Snapshots include configuration, messages, usage, metadata, extension state, and copied spill artifacts. Supply the same extension names on load; an explicitly supplied configuration must match the snapshot. Saving is atomic and rejects unsupported values instead of stringifying them. Saved artifacts survive agent closure until PersistentMemory(directory).delete_session(session_id).

For managed multi-session services (checkpointing, pause/resume, expiry), see durable runtimes and the 0.3 migration guide.

The default backend is SQLite. PostgreSQL automatically provisions its schema from a connection string; Temporal adds distributed workflow recovery and Redis events. For PostgreSQL, pass your configured agent directly: runtime = AgentRuntime(agent, backend=backend). In a replacement process, construct a fresh compatible Agent and call await runtime.resume(session_id). No agent registry is needed for this pattern. AgentRef and AgentRegistry remain available for named definitions and Temporal workers.

Try the PostgreSQL crash-and-resume walkthrough: write a report, let the process exit abruptly, and finish the same run in another process with the tool invocation count still at one. It uses scripted model responses, so only PostgreSQL and the postgres extra are required, not a model API key. Connection details can come from DATABASE_URL or --config examples/postgres.config.json; the config template supports automatic password URL encoding. Use the check command to connect and prepare the runtime schema before starting a run.

For incident debugging, opt in with AgentRuntime(..., recording=True). The flight recorder preserves model/tool attempt boundaries, exports portable incident bundles, and verifies/plays them back offline. Export payloads and artifacts are opt-in. SQLite is locally tested; PostgreSQL requires service qualification, and Temporal recording is not implemented. Try python -m examples.flight_recorder --output /tmp/invoice-incident.hx without an API key.

11. Shipping a web app

The harness is transport-agnostic. The trusted local, single-conversation example in examples/web_app demonstrates:

1. Streaming agent execution translated to Server-Sent Events (SSE) or WebSockets:

import asyncio
import json
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from harnessx import Agent, RunEventType

@asynccontextmanager
async def lifespan(app):
    async with Agent() as agent:
        app.state.agent = agent
        app.state.lock = asyncio.Lock()  # One shared conversation in this demo.
        yield

app = FastAPI(lifespan=lifespan)

@app.post("/api/stream")
async def stream(req: dict):
    async def event_generator():
        async with app.state.lock:
            async with app.state.agent.run_stream(req["message"]) as events:
                async for event in events:
                    if event.type == RunEventType.TEXT_DELTA:
                        yield f"data: {json.dumps({'type': 'text_delta', 'content': event.data})}\n\n"
                    elif event.type == RunEventType.ATTEMPT_RESET:
                        yield 'data: {"type": "attempt_reset"}\n\n'
                    elif event.type == RunEventType.ERROR:
                        yield f"data: {json.dumps({'type': 'error', 'content': str(event.data)})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

2. Register tools as PermissionLevel.ALLOW — in a server context, tools should be pre-approved or gated via an authorization middleware rather than interactive stdin prompts.
3. Built-in UI and MCP integrations — the web app serves a complete browser UI that displays tool calls, token usage, MCP servers, and dynamic skills.

Run the full example from the repo root:

uvicorn examples.web_app.server:app --host 127.0.0.1 --port 8000 --reload
# open http://localhost:8000

12. Evaluations with LangSmith

The harness includes first-class evaluation capabilities powered by the LangSmith evaluation framework (evaluate / aevaluate). You can benchmark tool selection, skill routing, multi-agent delegation, and guardrails either locally without uploading results or in the LangSmith Web UI. Local evaluation still invokes the supplied agent and may incur model/tool costs. The scripted python -m examples.run_evals --offline example needs no model API key.

Running an evaluation

from harnessx import Agent, AgentConfig, PermissionLevel
from examples._calculator import calculate  # Shared helper when running from a checkout.
from harnessx.evals import (
    build_example,
    default_evaluators,
    evaluate_agent,
    tool_selection_evaluator,
    contains_evaluator,
)

# 1. Define your agent factory or instance
def make_agent(inputs):
    agent = Agent(config=AgentConfig(model="claude-sonnet-4-6"))
    agent.tools.register_tool(calculate, permission=PermissionLevel.ALLOW, replay_policy="safe")
    return agent

# 2. Define test cases or use built-in suites ("tool_calling", "skills", "all")
dataset = [
    build_example(
        inputs={"prompt": "What is 144 / 12?"},
        outputs={
            "expected_tools": ["calculate"],
            "contains_all": ["12"],
            "max_allowed_iterations": 3,
        },
    )
]

# 3. Evaluate against the live model, keeping results local.
summary = evaluate_agent(
    agent=make_agent,
    dataset=dataset,
    evaluators=[tool_selection_evaluator, contains_evaluator],
    experiment_prefix="math-agent-benchmark",
    offline=True,   # Disables LangSmith upload; the agent still calls its model.
)

print(f"Pass rate: {summary.pass_rate * 100:.1f}%")

Running evals from the CLI

# Scripted integration demonstration (langsmith extra, no API keys required)
python -m examples.run_evals --offline

# Run live model benchmarks without upload (model API key required)
python -m harnessx.evals.cli --suite tool_calling --offline

# Run against Anthropic and upload results + traces to LangSmith
export LANGSMITH_API_KEY="lsv2_pt_..."
python -m harnessx.evals.cli --suite skills --model claude-sonnet-4-6 --upload

# Run all benchmark suites with concurrency
python -m harnessx.evals.cli --suite all --concurrency 2

When uploaded, the CLI prints a clickable LangSmith URL to inspect row-level scores, side-by-side prompt diffs, and the complete nested execution tree for every turn.

API reference (quick)

Runnable examples

Prerequisites and local verification coverage are listed in examples/README.md.