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. The fixture search handler makes the example independent of a search service; replace the handler with your integration when needed.

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

This shows pipeline injection; register the application tools separately. 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.

A permission-denied or invalid call does not execute a handler and does not take the normal successful-handler postprocessing path. Middleware transformations are not guaranteed to receive every synthetic error result.

Middleware added directly before extension installation stays earlier in the pipeline. Extension middleware is added in extension installation order; later manual additions run afterward. Extension priority orders installation, not an independent priority for individual middleware stages.

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

This exact-string example illustrates the return-value contract; it is not a complete secret-detection system. Preserve is_error and result identity. The result-stage API 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/thinking deltas until response processing completes. It then emits processed text; it does not expose the original incremental text while awaiting a final transformation. This trades immediate token delivery for consistent transformed output. An override that only observes a response still triggers buffering; use hooks when observation is sufficient.

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.

Direct Python function calls and ToolRegistry.execute() do not run the agent's middleware pipeline. Use the agent engine for these execution guarantees.

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, ErrorData, SkillInvokedData, KnowledgeAccessedData, SandboxExecData, CheckpointEventData). 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, and ERROR. 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 contains the request as sent: data["system"], the rendered system prompt, data["model"], data["max_tokens"], data["temperature"], data["stream"], data["message_count"], data["tool_count"], and data["prefix_key"], the prompt-cache key of that request, or None when caching is disabled. Consecutive iterations of one run share a 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 by Sandbox(config, hooks=...) for every execute() and execute_command() with kind, tier, exit_code, timed_out, and execution_time_ms; Agent(sandbox=...) supplies the agent's own hooks when the sandbox has none, so the bash built-in's commands reach agent.hooks. CHECKPOINT is emitted by AgentRuntime at every persisted phase boundary with session_id, run_id, status, and phase.

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 complete feature that installs middleware, a tool, a hook, and saved state.