Hooks API

hooks

Lifecycle hooks for tool execution.

A hook can either observe (legacy: return None) or decide (return a :class:HookDecision). Decisions let hooks redirect, skip, or abort tool execution — the foundation for security gates, self-healing watchdogs, and policy enforcement.

Hook errors are logged but never propagated, so a broken hook cannot disrupt tool execution. A hook that crashes while computing a decision defaults to :class:Continue.

The decision protocol applies at compound-flow boundaries: :class:~orqest.agents.compound_tool.CompoundTool, :func:~orqest.agents.retry.run_with_retry, and :meth:~orqest.autonomy.meta.MetaOrchestrator._execute_subtask. It does NOT intercept pydantic-AI's internal tool dispatch — for that future expansion, each :class:pydantic_ai.Tool.function would need to be wrapped with a hook-aware shim at construction time.

Continue

Proceed with the tool call as-is. The default no-op decision.

Skip

Skip the tool call. The compound flow returns stub_result in place of the executor result. after_tool still fires so observers see the skip. reason surfaces in resulting events.

Redirect

Replace the call's args and/or target tool before execution.

At least one of new_args/new_tool must be set.

Abort

Halt the compound flow. Raises :class:HookAbortError upstream; the surrounding :class:CompoundTool / :class:MetaOrchestrator / :func:run_with_retry handle it via their normal error paths.

HookAbortError

HookAbortError(reason: str, source_hook: str | None = None)

Raised when a hook returns :class:Abort.

Source code in orqest/hooks.py

def __init__(self, reason: str, source_hook: str | None = None) -> None:
    super().__init__(reason)
    self.reason = reason
    self.source_hook = source_hook

ToolHook

Protocol for tool lifecycle hooks.

Methods may return None (legacy fire-and-forget) or a :class:HookDecision (new — decision-issuing). :class:HookRunner auto-wraps None into :class:Continue so legacy hooks remain unchanged.

Implement any subset of methods. The HookRunner checks for method existence before calling, so a hook that only implements before_tool works without raising on the others.

before_tool async

before_tool(
    tool_name: str, args: dict[str, Any], state: Any
) -> HookDecision | None

Run before a tool executes.

Source code in orqest/hooks.py

async def before_tool(
    self, tool_name: str, args: dict[str, Any], state: Any
) -> HookDecision | None:
    """Run before a tool executes."""
    ...

after_tool async

after_tool(
    tool_name: str,
    args: dict[str, Any],
    result: Any,
    state: Any,
    duration_ms: float,
) -> HookDecision | None

Run after a tool completes successfully.

Source code in orqest/hooks.py

async def after_tool(
    self,
    tool_name: str,
    args: dict[str, Any],
    result: Any,
    state: Any,
    duration_ms: float,
) -> HookDecision | None:
    """Run after a tool completes successfully."""
    ...

on_error async

on_error(
    tool_name: str,
    args: dict[str, Any],
    error: Exception,
    state: Any,
) -> HookDecision | None

Handle errors when a tool raises an exception.

Source code in orqest/hooks.py

async def on_error(
    self,
    tool_name: str,
    args: dict[str, Any],
    error: Exception,
    state: Any,
) -> HookDecision | None:
    """Handle errors when a tool raises an exception."""
    ...

HookRunner

HookRunner(hooks: list[ToolHook] | None = None)

Dispatches hook events to registered hooks and aggregates decisions.

Aggregation rule: first-non-Continue wins, with Abort short-circuiting.

1. Iterate hooks in registration order.
2. If any hook returns :class:Abort, raise :class:HookAbortError immediately.
3. Otherwise, the first hook returning a non-:class:Continue decision (:class:Skip or :class:Redirect) is the active decision.
4. Subsequent hooks still run (observers may want to log), but their decisions are recorded as "shadowed" and logged at INFO.

Errors in hooks are logged at WARNING level and never re-raised. A hook that crashes while computing a decision defaults to :class:Continue.

Source code in orqest/hooks.py

def __init__(self, hooks: list[ToolHook] | None = None) -> None:
    self._hooks: list[ToolHook] = list(hooks) if hooks else []

run_before async

run_before(
    tool_name: str, args: dict[str, Any], state: Any
) -> HookDecision

Aggregate before_tool decisions across hooks.

Source code in orqest/hooks.py

async def run_before(
    self, tool_name: str, args: dict[str, Any], state: Any
) -> HookDecision:
    """Aggregate ``before_tool`` decisions across hooks."""
    return await self._aggregate("before_tool", tool_name, args, state)

run_after async

run_after(
    tool_name: str,
    args: dict[str, Any],
    result: Any,
    state: Any,
    duration_ms: float,
) -> HookDecision

Aggregate after_tool decisions across hooks.

Note: Skip from after_tool is meaningless (the executor already ran) — it's logged at WARNING and treated as :class:Continue. Redirect from after_tool requests a bounded re-execution (the caller decides how to honor it).

Source code in orqest/hooks.py

async def run_after(
    self,
    tool_name: str,
    args: dict[str, Any],
    result: Any,
    state: Any,
    duration_ms: float,
) -> HookDecision:
    """Aggregate ``after_tool`` decisions across hooks.

    Note: ``Skip`` from ``after_tool`` is meaningless (the executor
    already ran) — it's logged at WARNING and treated as
    :class:`Continue`. ``Redirect`` from ``after_tool`` requests a
    bounded re-execution (the caller decides how to honor it).
    """
    return await self._aggregate(
        "after_tool", tool_name, args, result, state, duration_ms
    )

run_error async

run_error(
    tool_name: str,
    args: dict[str, Any],
    error: Exception,
    state: Any,
) -> HookDecision

Aggregate on_error decisions across hooks.

Source code in orqest/hooks.py

async def run_error(
    self,
    tool_name: str,
    args: dict[str, Any],
    error: Exception,
    state: Any,
) -> HookDecision:
    """Aggregate ``on_error`` decisions across hooks."""
    return await self._aggregate("on_error", tool_name, args, error, state)