Middleware & hooks

Transform requests and results, observe execution, and choose between middleware, hooks, and extensions.

Middleware transforms data at defined points in the agent's execution pipeline. Hooks observe lifecycle events. An extension packages middleware, hooks, tools, and other capabilities with installation, saved state, and cleanup.

Which one should I use?

Need	Choose	Why
Normalize tool arguments	Middleware	The returned call becomes the call the engine validates and authorizes
Transform a model response or tool result	Middleware	The returned value continues through execution
Log a tool completion or collect diagnostic metrics	Hook	Observe an event without changing the result
Reuse a feature across agents with one constructor option	Extension	Bundle installation and configuration
Combine a tool, middleware, hooks, and saved state	Extension containing middleware and hooks	Give the whole feature a lifecycle
Approve or deny tool execution	Permission manager and execution checks	Hooks are best-effort observers, not an enforcement boundary

Start with standalone middleware for a focused transformation. Package it in an extension when callers should configure one reusable feature, or when the feature needs resources or persistence. An extension can install middleware with ctx.add_middleware(...); these are complementary interfaces.

The four middleware stages

Subclass Middleware and override only the stages you need. The base methods return their inputs unchanged. Every override is async and must return the expected value, including when it chooses not to change anything.

Method	Receives	Must return	Runs before
`before_llm_call(messages, tools)`	Conversation messages and canonical tool schemas	`(messages, tools)`	The provider request
`after_llm_call(response)`	Completed provider response	A valid response, preferably `ProviderResponse`	Response enters memory and is used for further decisions
`before_tool_execution(tool_call)`	`ToolCall`	`ToolCall` with the same `id`	Schema validation, permission resolution, and dispatch
`after_tool_execution(result)`	`ToolResult`	`ToolResult` with the same `tool_call_id`	Processed tool result enters the conversation

before_llm_call does not receive the system prompt, model, or provider settings. Use AgentConfig for static configuration or an extension prompt provider for dynamic system context. Changing the tool schemas sent to the model does not register handlers or replace the permission policy in the executor.

A complete standalone middleware example

This middleware strips whitespace from a search query while preserving call identity.

from dataclasses import replace
from harnessx import Agent, 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

def search(query: str) -> str:
    """Return a search result for the requested query."""
    return f"Results for {query}"

agent = Agent(tools=[search])
registration = agent.middleware.add(NormalizeSearch())
# Run the agent inside `async with agent:` when your provider is configured.
# To remove this middleware during setup: registration.close()

The same transformation applies to run, run_stream, and durable engine runs. The engine validates and checks permissions against the transformed call. Preserve call.id even if you intentionally change its name or arguments. Rewriting a tool name still requires that the target tool be registered and permitted.

Install a pipeline through the constructor

from harnessx import Agent, Middleware, MiddlewarePipeline, ToolCall

class RequireQuery(Middleware):
    async def before_tool_execution(self, call: ToolCall) -> ToolCall:
        if call.name == "search" and not str(call.input.get("query", "")).strip():
            raise ValueError("Search query must not be blank")
        return call

pipeline = MiddlewarePipeline()
pipeline.add(RequireQuery())
agent = Agent(middleware=pipeline)

Throwing an exception is an execution failure, not a request for human approval. Use ASK and a permission manager for approvals.

pipeline.add(...) returns an idempotent registration handle. Call close() to remove it. Configure or remove middleware between runs, not during an active run. Standalone middleware has no automatic async teardown contract; own resources in application code or put the middleware inside an extension.

Ordering and execution flow

If you add middleware A, then B, each stage runs A, then B. Response stages do not unwind in reverse order. B receives A's returned value.

The principal sequence is:

Compose the system prompt and trim conversation context.
Run before_llm_call, then call the provider.
Run after_llm_call, then store the processed response.
For requested tools, run before_tool_execution, validate the result, and resolve permissions.
Immediately before dispatch, check current execution policy again.
Execute the handler, retaining the raw outcome for durable recovery when applicable.
Run after_tool_execution, emit the tool-end hook, and collect processed results into memory.
Continue the model loop or finish the run.

Transforming results safely

from dataclasses import replace
from harnessx import Agent, Middleware, ToolResult

class HideFixtureToken(Middleware):
    async def after_tool_execution(self, result: ToolResult) -> ToolResult:
        return replace(
            result,
            content=result.content.replace("fixture-token", "[redacted]"),
        )

agent = Agent()
agent.middleware.add(HideFixtureToken())

Preserve is_error and result identity. The result stage receives the result, not the corresponding call arguments.

Result middleware changes what continues into the conversation. It does not prove the original content was never stored. Durable execution commits raw tool results before postprocessing, and a flight recorder can capture raw model responses before response middleware. Redact at the data source when original content must never enter storage or recording.

Streaming, errors, and retries

Overriding after_llm_call makes the current engine buffer that model response's text and thinking deltas until response processing completes. It then emits the processed text. This trades immediate token delivery for consistent transformed output. An override that only observes a response still triggers buffering, so use hooks when observation is enough.

Middleware exceptions are not swallowed like hook exceptions. They can fail the run. Handler exceptions usually become error tool results, which can then pass through result middleware. Inspect RunResult.status, error, and pending.

Durable recovery can rerun processing. In particular, a committed raw tool result can be postprocessed again without rerunning the handler. Middleware must tolerate repeated execution; a database write or notification inside it does not acquire exactly-once semantics. Prefer pure transformations and use registered tools with appropriate recovery policies for external actions.

Provider-owned content

Some response content belongs to the provider. Native Anthropic blocks reach after_llm_call as ordered canonical blocks, or as opaque {"type": "provider", "provider": "anthropic", "data": {...}} blocks. Middleware must leave signed thinking, opaque blocks, and a pending provider continuation as they are. Changing them raises ValueError and fails the run. Other providers' adapters leave out opaque blocks that belong to another vendor.

Hooks for observation

from typing import cast
from harnessx import Agent, HookContext, HookEvent
from harnessx.hooks import HOOK_PAYLOADS, ToolCallEndData

agent = Agent()

async def log_result(ctx: HookContext) -> None:
    data = cast(ToolCallEndData, ctx.data)  # typed view of the same dict
    result = data["result"]
    print("Tool result:", result.tool_call_id, "error:", result.is_error)

registration = agent.hooks.on(HookEvent.TOOL_CALL_END, log_result)
assert HOOK_PAYLOADS[HookEvent.TOOL_CALL_END] is ToolCallEndData
# registration.close() removes the observer.

Hooks may be synchronous or async. hooks.on(...), before_tool(...), after_tool(...), and on_error(...) all return a Registration handle whose close() removes the hook. The three shortcuts still work as decorators, but the decorated name is then the registration rather than the callback:

@agent.hooks.after_tool
async def audit(ctx: HookContext) -> None:
    print(ctx.data["tool_call"].name)

audit.close()  # audit is the Registration

HookContext supplies the event, agent, event-specific data, and a timestamp. data stays a dict, but the keys every built-in event carries are declared in harnessx.hooks as TypedDicts (AgentStartData, AgentEndData, LoopIterationData, LLMRequestData, LLMResponseData, ToolCallStartData, ToolCallEndData, RetryData, ContextCondensedData, RepetitionData, ErrorData, SkillInvokedData, KnowledgeAccessedData, SandboxExecData, CheckpointEventData, RecoveryData, TodosUpdatedData). HOOK_PAYLOADS maps each HookEvent to its shape; cast ctx.data to it for typed access instead of assuming every event contains a tool result.

Common engine events include AGENT_START, AGENT_END, LOOP_ITERATION_START, LOOP_ITERATION_END, LLM_REQUEST, LLM_RESPONSE, TOOL_CALL_START, TOOL_CALL_END, RETRY, CONTEXT_CONDENSED, REPETITION, and ERROR. CONTEXT_CONDENSED carries messages_dropped, summary_chars and summarized; REPETITION carries period (how many calls the repeating pattern has), laps (how many times it repeated, which is the guard's threshold) and tool (the first tool the model called in that turn). See Long-run reliability. AGENT_END contains data["result"], a final RunResult for completion, failure, or cancellation. An approval wait or pause is not a terminal completion.

LLM_REQUEST carries the request as sent:

Key in `ctx.data`	Value
`system`	The rendered system prompt
`model`	The model identifier
`max_tokens`	The reply budget for this call
`temperature`	The sampling temperature
`stream`	Whether the request streams
`message_count`	Number of conversation messages sent
`tool_count`	Number of tool schemas sent
`prefix_key`	The prompt-cache key, or `None` when caching is disabled

Consecutive iterations of one run share a prefix_key unless the system prompt or tool list changed.

Four events come from features rather than the core loop. SKILL_INVOKED carries skill, found, source_path, and body_chars. KNOWLEDGE_ACCESSED is emitted by the knowledge tools with tool, bundle, found, and either query and hits or path and body_chars. SANDBOX_EXEC is emitted for every sandbox execution; see sandbox for its payload and setup. CHECKPOINT is emitted by AgentRuntime at every persisted phase boundary with session_id, run_id, status, and phase.

Recovery events

HookEvent.RECOVERY fires when the loop recovers from a reply or request that hit a limit, instead of stopping. Its payload is RecoveryData:

Key in `ctx.data`	Value
`reason`	What happened; see below
`attempt`	How many recoveries of this kind the run has made, this one included
`reply_budget`	The reply budget the next call asks for, or `None`

`reason`	Meaning
`truncated_tool_call`	A tool call was cut off at the reply budget. It was refused and the model asked again with a larger budget
`truncated_text`	The reply was cut off. The model was asked to continue
`empty_reply`	The model said nothing. It was asked to go on
`reply_limit`	The provider refused the reply budget as too large. The call was repeated within the limit it named
`context_overflow`	The request did not fit the context window. The history was condensed and the call repeated
`context_overflow_spill`	The request did not fit and the history was too short to condense. Large tool results went to disk and the call was repeated
`spurious_overflow`	A request far below the window was refused as too long. It was tried once more
`malformed_tool_input`	A streamed tool call could not be read or identified. The reply was asked for again, once
`paused_turn`	The server paused a long turn. It was sent back to resume
`iteration_limit`	The run reached `max_iterations`. The model was asked for a final answer without tools
`unfinished_plan`	The model stopped with no answer and its task list still open. It was asked once to finish or say what is left

from harnessx import Agent, HookEvent

agent = Agent()
agent.hooks.on(HookEvent.RECOVERY, lambda ctx: print(
    "recovered:", ctx.data["reason"], "budget now", ctx.data["reply_budget"]))

See Long-run reliability for the limits behind each reason. The payload type is RecoveryData in harnessx.hooks.

Task list events

HookEvent.TODOS_UPDATED fires when the agent changes its task list with write_todos. Its payload is TodosUpdatedData, with the whole list every time:

Key in `ctx.data`	Value
`todos`	Every item, as `{"content": ..., "status": ...}`; status is `pending`, `in_progress` or `completed`
`completed`	How many items are completed
`total`	How many items there are
`in_progress`	The item being worked on, or `None`

from harnessx import Agent, HookEvent

agent = Agent()
agent.hooks.on(HookEvent.TODOS_UPDATED, lambda ctx: print(
    f"{ctx.data['completed']}/{ctx.data['total']} done"))

The same payload reaches the run stream as RunEventType.TODOS_UPDATED; see Streaming.

Observe retries and serving providers

HookEvent.RETRY fires when the engine schedules another model or tool attempt, and LLM_RESPONSE carries provider, the provider that answered. A long wait before a model retry, like a server's Retry-After, also shows up on the run stream as RunEventType.WAITING with {"on": "retry", "seconds": n}, where seconds is how long it has waited so far. See Retries & fallback models for both payloads, an observer example, and configuration.

Observer exceptions are suppressed. Hooks must not be the sole authorization, required audit, or verification step. Return values do not transform the pipeline; use middleware for that. Avoid mutating engine data through ctx.agent.

Continue with the extensions guide for a feature that installs middleware, a tool, a hook, and prompt context together.