Metadata-Version: 2.4
Name: agent-harnessed
Version: 0.3.0
Summary: A runtime environment that coordinates context, memory, prompts, tools, guardrails, output formatting, and a bounded reasoning loop around an LLM.
Project-URL: Homepage, https://github.com/NaveedMunsif/agent-harness
Project-URL: Repository, https://github.com/NaveedMunsif/agent-harness
Project-URL: Issues, https://github.com/NaveedMunsif/agent-harness/issues
Author: naveed
License: MIT
License-File: LICENSE
Keywords: agent,async,guardrails,harness,llm,tools
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: pydantic>=2.0
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Provides-Extra: examples
Requires-Dist: anthropic>=0.40; extra == 'examples'
Description-Content-Type: text/markdown

# agent-harness

[![PyPI](https://img.shields.io/pypi/v/agent-harnessed.svg)](https://pypi.org/project/agent-harnessed/)
[![Python](https://img.shields.io/pypi/pyversions/agent-harnessed.svg)](https://pypi.org/project/agent-harnessed/)
[![License: MIT](https://img.shields.io/pypi/l/agent-harnessed.svg)](https://github.com/NaveedMunsif/agent-harness/blob/main/LICENSE)

```bash
pip install agent-harnessed
```

A clean, async-first Python library that provides the **runtime environment around an LLM**.

```
Agent = LLM + Context + Memory + Tools + Control Flow + Guardrails + State
```

The LLM reasons and proposes actions. The harness decides what it sees, validates and
executes what it asks for, controls how long the loop runs, formats the result for its
delivery channel, and hands back the final answer.

The model client is **not** part of this library. You inject an async
`(CompiledPrompt) -> LLMTurnStep` callable, so `agent-harness` depends on `pydantic>=2.0`
and nothing else — no LLM SDK, no templating engine, no vector store.

## Install

The distribution is published as `agent-harnessed`; the import name is `agent_harness`:

```python
from agent_harness import LoopController, ToolGateway
```

New here? [**GETTING_STARTED.md**](https://github.com/NaveedMunsif/agent-harness/blob/main/GETTING_STARTED.md)
walks from an empty folder to a real agent calling your own tool, in about ten minutes.

For local development:

```bash
pip install -e ".[dev]"
```

## Architecture

| Module | Responsibility | Key types |
| --- | --- | --- |
| `state` | Task continuity. Explicit pivot-vs-continuation, never inferred from chat history. | `TaskFrame` |
| `memory` | Typed recall: durable facts, what happened, how to do things. Backends are injected. | `MemoryKind`, `MemoryRecord`, `MemoryBackend`, `InMemoryBackend`, `MemoryStore` |
| `context` | Decides what the model gets to see this turn, and owns the pivot decision. | `ContextEngine`, `ScopedContext` |
| `prompt` | Versioned prompt assembly — a compiled artifact, not scattered f-strings. | `PromptCompiler`, `CompiledPrompt` |
| `tools` | The only path from a proposal to a real side effect: allowlist, authorize, rate-limit, redact. | `Tool`, `ToolGateway`, `ToolProposal`, `ToolResult` |
| `guardrails` | Composable checks with a typed payload per stage. | `GuardrailPipeline`, `Guardrail`, `GuardrailStage`, `GuardrailViolation` |
| `output` | Channel-aware rendering of the final answer. Injected like the model client. | `RenderChannel`, `OutputFormatter`, `PlainTextFormatter` |
| `loop` | The bounded reasoning loop. Owns every exit decision and all memory write-back. | `LoopController`, `LoopLimits`, `LoopResult`, `LLMTurnStep`, `TurnStepType` |

### The loop has exactly three exits — and the LLM controls none of them

| `stopped_reason` | `final_text` | OUTPUT guardrail | Formatter |
| --- | --- | --- | --- |
| `final_answer` | the answer | ✅ runs | ✅ applied |
| `clarification_needed` | the question | ❌ skipped | ❌ skipped |
| `max_turns_exceeded` | `None` | ❌ skipped | ❌ skipped |
| `max_tool_calls_exceeded` | `None` | ❌ skipped | ❌ skipped |
| `max_seconds_exceeded` | `None` | ❌ skipped | ❌ skipped |
| `repeated_call_detected` | `None` | ❌ skipped | ❌ skipped |
| `guardrail_violation:{input\|context\|tool\|output}` | `None` | — | ❌ skipped |

A model can emit `final` on every turn; the loop still ends only when `LoopController`
says so. The OUTPUT guardrail is skipped for a clarification because **a question is not
a claim** — there is nothing to ground it against.

`repeated_call_detected` is the one exit the loop tries to avoid reaching. A model that
proposes the same call twice is usually *stuck* rather than looping — it wanted
something the result did not contain and looked again. So the first repeat appends a
correction to episodic context (*"you already called this; its result is above"*) and
the turn continues; no tool runs and no budget is spent. Only a second identical
proposal, after that correction, ends the turn. The alternative — stopping on the first
repeat — hands the caller `final_text=None` when the answer was already in hand.

### Guardrail payload contract

Each stage's `check()` receives exactly one type, so guardrails are written against
concrete objects rather than an untyped blob:

| Stage | Payload |
| --- | --- |
| `INPUT` | `str` — the raw incoming user message |
| `CONTEXT` | `ScopedContext` — assembled context, pre-compile |
| `TOOL` | `ToolProposal` — proposed tool + args, pre-execution |
| `OUTPUT` | `tuple[str, list[ToolResult]]` — final text plus every `ToolResult` from the turn |

A violation never reaches your process: `LoopController` catches `GuardrailViolation`,
records it, and returns a clean `LoopResult` with `final_text=None`. Output guardrails
must fail safe when evidence is missing, not crash the caller.

The reason survives the catch. `stopped_reason` names only the stage, so `LoopResult`
also carries `violation_reason` — otherwise the one sentence explaining the refusal
would exist solely inside an exception you never see:

```python
if result.violation_reason:
    print(result.violation_reason)   # "refund of $800.00 exceeds the $500.00 limit"
```

### The gateway can explain itself

`Tool.authorize` may return `True`, `False`, or **a string** — a string refuses *and*
supplies the reason. It matters more than it looks: a bare `False` produces only
"authorization declined", which a model reads as a transient error and retries, and a
user reads as nothing at all.

```python
async def authorize_refund(session_id: str, arguments: dict) -> bool | str:
    order = fetch(arguments["order_id"])
    if order is None or order["customer"] != customer_for(session_id):
        return "that order is not on this account"
    return True
```

Allowlisting, rate limiting and redaction have no callable to wrap, so they are
invisible from outside — a rate-limited call and a call that was never proposed look
identical in a `LoopResult`. Pass `on_event` to see them:

```python
gateway = ToolGateway(tools, on_event=lambda e: print(e.stage, e.tool_name, e.ok))
# authorize     issue_refund True
# rate_limit    issue_refund False
```

Each event is a `GatewayEvent` with `stage` (`allowlist`, `authorize`, `rate_limit`,
`execute`), `tool_name`, `session_id`, `ok` and an optional `detail`.

## A complete agent, wired to a real model

An order-support agent over Claude. Everything below is the whole program.

```bash
pip install agent-harnessed anthropic
export ANTHROPIC_API_KEY=sk-ant-...      # setx on Windows
```

```python
import asyncio

from anthropic import AsyncAnthropic

from agent_harness import (
    ContextEngine, GuardrailPipeline, InMemoryBackend, LLMTurnStep, LoopController,
    MemoryStore, PromptCompiler, Tool, ToolGateway, TurnStepType,
)

ORDERS = {"A-1001": {"status": "shipped", "carrier": "UPS", "eta": "2026-08-03"}}


async def lookup_order(session_id: str, arguments: dict) -> dict:
    return ORDERS[str(arguments["order_id"]).upper()]


lookup = Tool(
    name="lookup_order",
    description="Look up an order's status by its id, e.g. A-1001.",
    parameters=["order_id"],
    execute=lookup_order,
    read_only=True,
)


def make_call_llm(client: AsyncAnthropic, tools: list[Tool]):
    """The only piece this library leaves to you: prompt in, LLMTurnStep out."""
    schemas = [
        {
            "name": tool.name,
            "description": tool.description,
            "input_schema": {
                "type": "object",
                "properties": {p: {"type": "string"} for p in tool.parameters},
                "required": list(tool.parameters),
            },
        }
        for tool in tools
    ]

    async def call_llm(prompt) -> LLMTurnStep:
        response = await client.messages.create(
            model="claude-sonnet-5",
            max_tokens=16000,
            messages=[{"role": "user", "content": prompt.render()}],
            tools=schemas,
        )
        for block in response.content:
            if block.type == "tool_use":
                return LLMTurnStep(
                    step_type=TurnStepType.TOOL_CALL,
                    tool_name=block.name,
                    tool_arguments=dict(block.input),
                )
        text = "".join(b.text for b in response.content if b.type == "text")
        return LLMTurnStep(step_type=TurnStepType.FINAL, text=text.strip())

    return call_llm


async def main() -> None:
    memory = MemoryStore(InMemoryBackend())
    controller = LoopController(
        context_engine=ContextEngine(memory),
        prompt_compiler=PromptCompiler(role="You are a concise order-support agent."),
        tool_gateway=ToolGateway([lookup]),
        guardrails=GuardrailPipeline(),
        memory=memory,
        call_llm=make_call_llm(AsyncAnthropic(), [lookup]),
    )

    result = await controller.handle_turn("session-1", "Where is order A-1001?")

    print(result.stopped_reason)        # final_answer
    print(result.final_text)            # Order A-1001 has shipped with UPS.
    print(result.tool_results[0].data)  # {'status': 'shipped', 'carrier': 'UPS', ...}


asyncio.run(main())
```

Two round trips to the model, and you wrote neither of them. The first returns a
`tool_use` block, so the gateway runs `lookup_order` and folds the result into the
next prompt as history; the second sees that evidence and answers. `handle_turn`
returns once, when `LoopController` decides the turn is over.

### The adapter is the whole integration

`make_call_llm` above is the entire model-specific surface. Three mappings do all
the work:

| The model does this | The adapter returns |
| --- | --- |
| emits a `tool_use` block | `TurnStepType.TOOL_CALL` |
| calls the `ask_user` tool | `TurnStepType.CLARIFICATION` |
| replies with text only | `TurnStepType.FINAL` |

Tool schemas are derived from the same `list[Tool]` handed to the gateway, so what
the model is told about and what the gateway will actually run cannot drift apart.

The middle row is worth a note. `ask_user` is a **sentinel**: declared to the model
as an ordinary tool but never registered with the `ToolGateway`, so the adapter
intercepts the call rather than executing it. That lets the model ask a question
through the same native tool-calling channel it uses to act, instead of you parsing
its prose to guess whether an answer was really a question.

Nothing here is Claude-specific beyond the SDK call. Any model with native tool
calling maps the same three ways; one without it needs the adapter to parse a
structured response instead.

## The example

```bash
python examples/refund_agent.py --session naveed --trace
```

[`examples/refund_agent.py`](https://github.com/NaveedMunsif/agent-harness/blob/main/examples/refund_agent.py)
is a refund agent over SQLite — the first thing in a support agent that can
actually cost money. Three independent mechanisms stand between the model's
proposal and a debited account: ownership via `Tool.authorize`, a $500 ceiling as
a TOOL guardrail, and an atomic conditional `UPDATE` capped by
`max_calls_per_session`. Each stops a different request at a different stage.

`--trace` prints which stages a turn passed through before it stopped, so the
pipeline is something you read rather than infer:

```
PASS  input guardrail    input-size: passed
  ->  context            intent=refund_order entities={'order_id': 'A-1007'}
  ->  state              fresh frame (pivot -- earlier task dropped)
  ->  memory             recalled 1 semantic, 1 procedural, 0 episodic
  ->  prompt             v1.0.0, 6 sections: task_frame, known_facts, procedures...
  ->  model              turn 1: proposes issue_refund(order_id=A-1007)
STOP  tool guardrail     refund-ceiling: refund of $800.00 exceeds the $500.00 limit
  ->  memory             wrote episodic: guardrail_violation
  ==  stopped            guardrail_violation:tool
```

The model never writes SQL. It names a tool and supplies a value; every statement
is parameterized and lives in your code. And note the second-to-last line — a
blocked turn still writes its audit record.

## Clarification: a return, not a suspension

When the model needs to ask something, `handle_turn` **returns immediately**. It does
not suspend, block, or hold an open coroutine. The caller owns the wait — print the
question, collect an answer over whatever transport it has (HTTP request, websocket,
SMS, tomorrow), then call `handle_turn` again with that answer and the frame it got
back. The controller keeps no state between calls beyond what you pass in.

```python
asked = await controller.handle_turn("session-1", "Can you check on my order?")
# asked.stopped_reason == "clarification_needed"
# asked.final_text     == "Which order number should I look up?"

# ... your transport waits here, for however long it takes ...

resolved = await controller.handle_turn(
    "session-1",
    "It's A-1001 - where is it?",
    current_frame=asked.task_frame,      # hand the same frame back
)
# resolved.stopped_reason == "final_answer"
```

Because the intent matches and no entity conflicts, that second turn is a
**continuation**: `ContextEngine.build` extends the frame via `merged_with()` and
`is_pivot` is `False`. Ask about a *different* order and the same intent now carries a
conflicting `order_id`, so the frame is rebuilt via `TaskFrame.fresh()` with
`is_pivot=True` — dropping the abandoned task's plan and in-flight tool ids.

Running the example with `--trace` shows which of the two happened on every turn.

## What the episodic record looks like

Memory write-back is a correctness requirement, not an optimization. Tools that caused
real side effects are recorded **the moment they complete**, so nothing about a forced
stop can erase them.

| Exit | Episodic records written |
| --- | --- |
| tool call → final | one `tool_call` per executed tool, then one `turn_complete` holding the user message and the final answer |
| clarification | one `turn_complete` holding the user message and the question — no tool records, since no tool ran |
| `max_*_exceeded` / `repeated_call_detected` | every `tool_call` that already ran, then one `forced_stop` naming the specific `stopped_reason` |
| `guardrail_violation:*` | every `tool_call` that already ran, then one `guardrail_violation` naming the stage and reason |

So a forced stop after two lookups leaves `[tool_call, tool_call, forced_stop]` — no
answer was produced, but the audit trail shows both executions *and* why the turn was
cut short. A resolved turn never loses that granularity either: the consolidated
`turn_complete` record sits alongside the per-tool records, not instead of them.

## Out of scope

- **The model client.** Injected as an async callable. This library ships no model client.
- **Pause-in-place clarification.** Return-based only, as described above.
- **Long-term persistence.** Beyond in-process state, that is the caller's job — implement
  `MemoryBackend` against your own store.
- **Concurrency safety.** A single instance is not guaranteed coroutine-safe. If you might
  call `handle_turn` concurrently for the same session, serialize it yourself (one
  `asyncio.Lock` per session).

## Tests

```bash
pytest
```

116 tests covering tool authorization/redaction/rate-limiting, pivot-vs-continuation
semantics and the `fresh()`/`merged_with()` branch in `ContextEngine.build`, all three
loop exits, every forced stop (including that already-executed `ToolResult`s survive
one), guardrail violations converting to clean results at each stage — carrying their
reason — `LLMTurnStep` validation rejecting malformed steps, repeated-call detection
over non-JSON-native arguments, a repeat being corrected once before it ends a turn,
`authorize` refusing with a string (and a *truthy* string still refusing), a
`GatewayEvent` for every gateway decision including the two that have no callable to
wrap, the per-stage guardrail payload contract, and that the topic-change note reaches
the prompt on a real pivot but never on a session's opening message.
