Hooks (Lifecycle Callbacks)

Hooks let you intercept and modify agent behavior at every step of the lifecycle — before/after tool use, on prompt submission, on stop, and more.

Registering hooks

At construction time (options-level)

Pass a dict of HookMatcher lists keyed by event name:

from agentix import AgentixAgentOptions, HookMatcher

options = AgentixAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(
                matcher="Bash",            # regex matching the tool name; None = match all
                hooks=[my_pre_bash_hook],  # list of async callbacks
            ),
        ],
        "Stop": [
            HookMatcher(matcher=None, hooks=[my_stop_hook]),
        ],
    }
)

At runtime (client-level)

client.on("PreToolUse", my_callback)
client.on("Stop", my_stop_callback)

client.on() registers a single callback that matches all invocations of the event (equivalent to matcher=None).

---

HookMatcher

HookMatcher pairs an optional filter with one or more callbacks:

from agentix import HookMatcher

HookMatcher(
    matcher=None,          # str | None — regex against tool name; None = match all
    hooks=[my_callback],   # list[AsyncCallable] — called in order
    timeout=60.0,          # float — per-callback timeout in seconds; 0 = no timeout
)

The matcher string is a regex pattern matched against the relevant key for each event. For PreToolUse, PostToolUse, and PostToolUseFailure this is the tool name:

# Match only "Bash"
HookMatcher(matcher="Bash", hooks=[my_hook])

# Match any write-capable tool
HookMatcher(matcher="Bash|Write|Edit", hooks=[my_hook])

# Match all tools (no filter)
HookMatcher(matcher=None, hooks=[my_hook])

---

Hook events

Event	Key hook_input fields	Typed Context
PreToolUse	tool_name, tool_input, tool_use_id	ToolHookContext
PostToolUse	tool_name, tool_input, tool_result	ToolHookContext
PostToolUseFailure	tool_name, error	ToolHookContext
UserPromptSubmit	prompt, session_id	PromptHookContext
Stop	reason, final_text, session_id	StopHookContext
SubagentStart	child_name, session_id	SubagentHookContext
SubagentStop	child_name, session_id	SubagentHookContext
PreCompact	current_count, session_id	CompactHookContext
Notification	message, level	NotificationHookContext
PermissionRequest	tool_name, decision	PermissionHookContext

---

Hook callback signature

Every hook callback is an async function. Return None to take no action, or one of the output types to modify behavior:

from agentix import SyncHookJSONOutput, ToolHookContext

async def my_hook(
    hook_input: dict,
    context: ToolHookContext,
) -> SyncHookJSONOutput | None:
    # Return None — allow and continue unchanged
    # Return SyncHookJSONOutput — modify behavior synchronously
    return None

HookJSONOutput is exported as a type alias for SyncHookJSONOutput | AsyncHookJSONOutput. In practice, always instantiate one of the two concrete types directly.

---

Return types: when to use which

Return value	When to use
None	Hook is purely observational — no effect on execution
SyncHookJSONOutput	Hook needs to affect execution: block a tool, modify input/output, stop the chain
AsyncHookJSONOutput	Hook fires background work (telemetry, audit log) with no synchronous effect

---

SyncHookJSONOutput

Use SyncHookJSONOutput whenever the hook needs to influence what the agent does next. It is a dataclass — all fields have defaults, so SyncHookJSONOutput() with no arguments is a valid no-op.

from agentix import SyncHookJSONOutput

SyncHookJSONOutput(
    continue_=True,          # False — stop the hook chain early for this event
    decision=None,           # "block" | "allow" — PreToolUse only
    updated_output=None,     # str — replace the tool result (PostToolUse only)
    additional_context=None, # str — append to the tool result (PostToolUse only)
    reason=None,             # str — explanation recorded in logs
)

Which fields apply to which events

Field	PreToolUse	PostToolUse	All other events
continue_	✅ stops chain	✅ stops chain	✅ stops chain
decision="block"	✅ denies tool call	—	—
updated_output	—	✅ replaces result seen by LLM	—
additional_context	—	✅ appended to result	—
reason	✅ logged	✅ logged	✅ logged

PostToolUse accumulation rules: when multiple hooks run on the same event, updated_output — the last non-None value wins; additional_context — all non-None values are concatenated with \n in registration order.

---

AsyncHookJSONOutput

Use AsyncHookJSONOutput when the hook fires background work and intentionally has no effect on the tool call or its result. The harness treats it as a no-op — exactly like returning None — but the pattern signals clearly that background work is in flight.

from agentix import AsyncHookJSONOutput
import asyncio

async def send_to_telemetry(tool_name: str, duration_ms: float) -> None:
    # ... async HTTP call to your telemetry backend ...
    pass

async def telemetry_hook(hook_input: dict, context) -> AsyncHookJSONOutput:
    asyncio.create_task(
        send_to_telemetry(
            tool_name=hook_input["tool_name"],
            duration_ms=hook_input.get("duration_ms", 0),
        )
    )
    return AsyncHookJSONOutput(async_=True)

Important: never use AsyncHookJSONOutput in a PreToolUse hook to try to block a tool — it has no decision field and no synchronous effect. Use SyncHookJSONOutput(decision="block") for that.

---

Examples

Logging every tool call

import logging

async def log_tool_use(hook_input: dict, context) -> None:
    logging.info("Tool: %s  input=%s", hook_input["tool_name"], hook_input.get("tool_input"))

client.on("PreToolUse", log_tool_use)

Blocking a specific tool

from agentix import SyncHookJSONOutput

async def block_bash(hook_input: dict, context) -> SyncHookJSONOutput | None:
    if hook_input["tool_name"] == "Bash":
        return SyncHookJSONOutput(decision="block", reason="Shell access disabled.")
    return None

client.on("PreToolUse", block_bash)

Mutating tool input before execution

PreToolUse hooks can rewrite the arguments before the tool runs. Pass the new arguments via hookSpecificOutput:

from agentix import SyncHookJSONOutput

async def force_readonly(hook_input: dict, context) -> SyncHookJSONOutput | None:
    if hook_input["tool_name"] == "Bash":
        cmd = hook_input["tool_input"].get("command", "")
        # Prepend a safety prefix — agent can only run read-like commands
        return SyncHookJSONOutput(
            hookSpecificOutput={"updatedInput": {"command": f"set -e; {cmd}"}}
        )
    return None

client.on("PreToolUse", force_readonly)

Sanitizing tool output

from agentix import SyncHookJSONOutput

async def redact_secrets(hook_input: dict, context) -> SyncHookJSONOutput | None:
    result = hook_input.get("tool_result", "")
    if "SECRET_TOKEN" in result:
        return SyncHookJSONOutput(
            updated_output=result.replace("SECRET_TOKEN", "[REDACTED]")
        )
    return None

client.on("PostToolUse", redact_secrets)

Appending audit context to a result

Multiple PostToolUse hooks can each append their own note — the LLM sees all of them concatenated:

from agentix import SyncHookJSONOutput
from datetime import datetime, timezone

async def add_timestamp(hook_input: dict, context) -> SyncHookJSONOutput:
    ts = datetime.now(timezone.utc).isoformat()
    return SyncHookJSONOutput(additional_context=f"[executed at {ts}]")

async def add_user_audit(hook_input: dict, context) -> SyncHookJSONOutput:
    return SyncHookJSONOutput(additional_context="[audit: reviewed by policy engine]")

# Both additional_context values are concatenated: "[executed at ...]\n[audit: ...]"
client.on("PostToolUse", add_timestamp)
client.on("PostToolUse", add_user_audit)

Fire-and-forget telemetry (AsyncHookJSONOutput)

from agentix import AsyncHookJSONOutput
import asyncio
import httpx

async def push_metric(tool_name: str) -> None:
    async with httpx.AsyncClient() as http:
        await http.post("https://metrics.example.com/event", json={"tool": tool_name})

async def telemetry_hook(hook_input: dict, context) -> AsyncHookJSONOutput:
    # Fire background task — does not block or affect the tool call
    asyncio.create_task(push_metric(hook_input["tool_name"]))
    return AsyncHookJSONOutput(async_=True)

client.on("PostToolUse", telemetry_hook)

Stop hook

from agentix import StopHookContext

async def on_stop(hook_input: dict, context: StopHookContext) -> None:
    print(f"Agent stopped: reason={hook_input['reason']}")
    print(f"Final text: {hook_input.get('final_text', '')[:200]}")

client.on("Stop", on_stop)

Metrics counter

from agentix import AgentixAgentOptions, HookMatcher

tool_counts: dict[str, int] = {}

async def count_tool(hook_input: dict, context) -> None:
    name = hook_input["tool_name"]
    tool_counts[name] = tool_counts.get(name, 0) + 1

options = AgentixAgentOptions(
    hooks={"PreToolUse": [HookMatcher(matcher=None, hooks=[count_tool])]}
)

---

Hook chain semantics

Multiple matchers on the same event are processed in order:

• PreToolUse: Each callback can mutate tool_input. A decision="block" from any callback stops the chain and denies the tool call.
• PostToolUse: All callbacks see the original result. updated_output — last non-None value wins. additional_context — all non-None values are concatenated with \n.
• Other events: callbacks run in order; continue_=False stops the chain early.

---

Hook timeout

Callbacks that exceed their timeout are cancelled and logged as a warning — they do not abort the agent run. Default: 60 seconds.

HookMatcher(matcher=None, hooks=[my_slow_hook], timeout=10.0)   # 10s limit
HookMatcher(matcher=None, hooks=[my_hook], timeout=0)           # no timeout

---

Built-in hook helpers

from agentix.hooks import logging_hook_matchers, metrics_hook_matchers

# Structured console logging for all events
options = AgentixAgentOptions(hooks=logging_hook_matchers())

# Simple call counter (increments counter[tool_name] on PreToolUse)
counts: dict[str, int] = {}
options = AgentixAgentOptions(hooks=metrics_hook_matchers(counts))