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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
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

    def is_stream_safe(self) -> bool:
        """Returns True when no registered hook can retry/block/fall back a response.

        A hook that doesn't override `after_model` or `wrap_model_call` is
        safe automatically — it can't affect the outcome. A hook that does
        override either is unsafe unless it explicitly declares
        `stream_safe = True` (see `AgentHook.stream_safe`), since such a hook
        may need the complete response to decide, and tokens already
        streamed to a client cannot be retracted.
        """
        return all(
            h.stream_safe
            or (
                type(h).after_model is AgentHook.after_model
                and type(h).wrap_model_call is AgentHook.wrap_model_call
            )
            for h in 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

is_stream_safe 

is_stream_safe() -> bool

Returns True when no registered hook can retry/block/fall back a response.

A hook that doesn't override after_model or wrap_model_call is safe automatically — it can't affect the outcome. A hook that does override either is unsafe unless it explicitly declares stream_safe = True (see AgentHook.stream_safe), since such a hook may need the complete response to decide, and tokens already streamed to a client cannot be retracted.

Source code in src/ant_ai/hooks/layer.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def is_stream_safe(self) -> bool:
    """Returns True when no registered hook can retry/block/fall back a response.

    A hook that doesn't override `after_model` or `wrap_model_call` is
    safe automatically — it can't affect the outcome. A hook that does
    override either is unsafe unless it explicitly declares
    `stream_safe = True` (see `AgentHook.stream_safe`), since such a hook
    may need the complete response to decide, and tokens already
    streamed to a client cannot be retracted.
    """
    return all(
        h.stream_safe
        or (
            type(h).after_model is AgentHook.after_model
            and type(h).wrap_model_call is AgentHook.wrap_model_call
        )
        for h in 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 
61
62
63
64
65
66
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 
68
69
70
71
72
73
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 
75
76
77
78
79
80
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 
 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
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 
108
109
110
111
112
113
114
115
116
117
118
119
120
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 
122
123
124
125
126
127
128
129
130
131
132
133
134
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