Skip to content

ant_ai.hooks.layer

HookLayer pydantic-model

Bases: BaseModel

Orchestrates all registered hooks at each agent lifecycle point.

Before hooks run in registration order (inward); after hooks run in reverse order (outward). For hooks [A, B, C] the call order is: A.before → B.before → C.before → [core] → C.after → B.after → A.after

wrap_model_call and wrap_tool_call follow the same onion pattern, building an ASGI-style chain where each hook wraps the next.

Config:

  • arbitrary_types_allowed: True

Fields:

Source code in src/ant_ai/hooks/layer.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class HookLayer(BaseModel):
    """Orchestrates all registered hooks at each agent lifecycle point.

    Before hooks run in registration order (inward); after hooks run in
    reverse order (outward). For hooks [A, B, C] the call order is:
    A.before → B.before → C.before → [core] → C.after → B.after → A.after

    `wrap_model_call` and `wrap_tool_call` follow the same onion pattern,
    building an ASGI-style chain where each hook wraps the next.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    hooks: list[AgentHook] = Field(
        default_factory=list,
        description="Hooks to run at each lifecycle point, in registration order.",
    )

    def is_empty(self) -> bool:
        """Returns True when no hooks are registered."""
        return not self.hooks

    async def run_before_agent(
        self, state: State, ctx: InvocationContext | None
    ) -> None:
        """Runs `before_agent` on all hooks in registration order."""
        for h in self.hooks:
            await h.before_agent(state, ctx)

    async def run_after_agent(
        self, state: State, ctx: InvocationContext | None
    ) -> None:
        """Runs `after_agent` on all hooks in reverse registration order."""
        for h in reversed(self.hooks):
            await h.after_agent(state, ctx)

    async def run_before_model(
        self, state: State, ctx: InvocationContext | None
    ) -> None:
        """Runs `before_model` on all hooks in registration order."""
        for h in self.hooks:
            await h.before_model(state, ctx)

    async def run_after_model(
        self,
        result: StepResult,
        ctx: InvocationContext | None,
    ) -> PostModelDecision:
        """Runs `after_model` on all hooks and merges their decisions.

        Uses worst-verdict-wins: Fallback > Block > Retry > Pass. Returns
        `PostModelPass` directly when no hooks are registered.

        Args:
            result: The step result produced by the LLM step.
            ctx: Invocation context, or None if not available.

        Returns:
            The merged decision from all hooks.
        """
        if self.is_empty():
            return PostModelPass(result=result)

        decisions: list[PostModelDecision] = []
        for h in reversed(self.hooks):
            decisions.append(await h.after_model(result, ctx))

        return _merge_decisions(decisions, result)

    def wrap_model_call(self, core: WrapCall) -> WrapCall:
        """Chains all `wrap_model_call` hooks around `core`, outermost-first.

        Args:
            core: The innermost callable to wrap.

        Returns:
            A callable with all hooks layered around `core`.
        """
        wrapped: WrapCall = core
        for hook in reversed(self.hooks):
            wrapped: WrapCall = _make_layer("wrap_model_call", hook, wrapped)
        return wrapped

    def wrap_tool_call(self, core: WrapCall) -> WrapCall:
        """Chains all `wrap_tool_call` hooks around `core`, outermost-first.

        Args:
            core: The innermost callable to wrap.

        Returns:
            A callable with all hooks layered around `core`.
        """
        wrapped: WrapCall = core
        for hook in reversed(self.hooks):
            wrapped: WrapCall = _make_layer("wrap_tool_call", hook, wrapped)
        return wrapped

hooks pydantic-field 

hooks: list[AgentHook]

Hooks to run at each lifecycle point, in registration order.

is_empty 

is_empty() -> bool

Returns True when no hooks are registered.

Source code in src/ant_ai/hooks/layer.py 
38
39
40
def is_empty(self) -> bool:
    """Returns True when no hooks are registered."""
    return not self.hooks

run_before_agent async 

run_before_agent(
    state: State, ctx: InvocationContext | None
) -> None

Runs before_agent on all hooks in registration order.

Source code in src/ant_ai/hooks/layer.py 
42
43
44
45
46
47
async def run_before_agent(
    self, state: State, ctx: InvocationContext | None
) -> None:
    """Runs `before_agent` on all hooks in registration order."""
    for h in self.hooks:
        await h.before_agent(state, ctx)

run_after_agent async 

run_after_agent(
    state: State, ctx: InvocationContext | None
) -> None

Runs after_agent on all hooks in reverse registration order.

Source code in src/ant_ai/hooks/layer.py 
49
50
51
52
53
54
async def run_after_agent(
    self, state: State, ctx: InvocationContext | None
) -> None:
    """Runs `after_agent` on all hooks in reverse registration order."""
    for h in reversed(self.hooks):
        await h.after_agent(state, ctx)

run_before_model async 

run_before_model(
    state: State, ctx: InvocationContext | None
) -> None

Runs before_model on all hooks in registration order.

Source code in src/ant_ai/hooks/layer.py 
56
57
58
59
60
61
async def run_before_model(
    self, state: State, ctx: InvocationContext | None
) -> None:
    """Runs `before_model` on all hooks in registration order."""
    for h in self.hooks:
        await h.before_model(state, ctx)

run_after_model async 

run_after_model(
    result: StepResult, ctx: InvocationContext | None
) -> PostModelDecision

Runs after_model on all hooks and merges their decisions.

Uses worst-verdict-wins: Fallback > Block > Retry > Pass. Returns PostModelPass directly when no hooks are registered.

Parameters:

Name Type Description Default
result StepResult

The step result produced by the LLM step.

 required
ctx InvocationContext | None

Invocation context, or None if not available.

 required

Returns:

Type Description
PostModelDecision

The merged decision from all hooks.
Source code in src/ant_ai/hooks/layer.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
async def run_after_model(
    self,
    result: StepResult,
    ctx: InvocationContext | None,
) -> PostModelDecision:
    """Runs `after_model` on all hooks and merges their decisions.

    Uses worst-verdict-wins: Fallback > Block > Retry > Pass. Returns
    `PostModelPass` directly when no hooks are registered.

    Args:
        result: The step result produced by the LLM step.
        ctx: Invocation context, or None if not available.

    Returns:
        The merged decision from all hooks.
    """
    if self.is_empty():
        return PostModelPass(result=result)

    decisions: list[PostModelDecision] = []
    for h in reversed(self.hooks):
        decisions.append(await h.after_model(result, ctx))

    return _merge_decisions(decisions, result)

wrap_model_call 

wrap_model_call(core: WrapCall) -> WrapCall

Chains all wrap_model_call hooks around core, outermost-first.

Parameters:

Name Type Description Default
core WrapCall

The innermost callable to wrap.

 required

Returns:

Type Description
WrapCall

A callable with all hooks layered around core.
Source code in src/ant_ai/hooks/layer.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def wrap_model_call(self, core: WrapCall) -> WrapCall:
    """Chains all `wrap_model_call` hooks around `core`, outermost-first.

    Args:
        core: The innermost callable to wrap.

    Returns:
        A callable with all hooks layered around `core`.
    """
    wrapped: WrapCall = core
    for hook in reversed(self.hooks):
        wrapped: WrapCall = _make_layer("wrap_model_call", hook, wrapped)
    return wrapped

wrap_tool_call 

wrap_tool_call(core: WrapCall) -> WrapCall

Chains all wrap_tool_call hooks around core, outermost-first.

Parameters:

Name Type Description Default
core WrapCall

The innermost callable to wrap.

 required

Returns:

Type Description
WrapCall

A callable with all hooks layered around core.
Source code in src/ant_ai/hooks/layer.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
def wrap_tool_call(self, core: WrapCall) -> WrapCall:
    """Chains all `wrap_tool_call` hooks around `core`, outermost-first.

    Args:
        core: The innermost callable to wrap.

    Returns:
        A callable with all hooks layered around `core`.
    """
    wrapped: WrapCall = core
    for hook in reversed(self.hooks):
        wrapped: WrapCall = _make_layer("wrap_tool_call", hook, wrapped)
    return wrapped