From 7d21283630f20f1a71fcb621fa77bb5b957a5260 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Tue, 14 Jul 2026 11:27:34 -0300 Subject: [PATCH] feat: add generic interception-hook dispatcher (#6516) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: add generic interception-hook dispatcher Introduces `crewai/hooks/dispatch.py` as a single engine behind every interception point: a hook receives a typed context, may mutate or replace its `payload`, or raise `HookAborted(reason, source)` to stop the operation. The full `InterceptionPoint` catalog is frozen from day zero, with global and contextvar-scoped registries, an `@on` decorator, a no-op fast path, and a `HookDispatchedEvent` for telemetry. The four existing `before/after_llm_call` and `before/after_tool_call` hooks become adapters over the dispatcher, so the legacy dialect and `return False` semantics keep working unchanged while gaining the new contract. * fix: harden interception dispatcher against review findings Corrects several dispatcher edge cases surfaced in review. `_default_reducer` now reports a modification only when a `payload` is actually applied, the `agents=` filter falls back to `agent_role` for contexts without an `agent` object, and `unregister` resolves the filter wrapper stashed by `on` so a filtered hook can be removed. The tool-hook runners honor the executing agent's `verbose` flag instead of silently swallowing hook errors, and the ReAct tool path now runs `POST_TOOL_CALL` on blocked calls to match the native paths. Also adds abort-telemetry coverage and replaces the flaky absolute no-op timing budget with a relative one. * fix: honor scoped hooks on direct llm calls and register @on crew methods Direct agent-less LLM calls short-circuited on the empty global hook list, so hooks registered only for the current `scoped_hooks()` context never ran; the direct-call helpers now defer to `dispatch`, which resolves scoped hooks behind its own no-op fast path. `CrewBase` likewise only scanned the legacy `is_*_hook` markers, so `@on(InterceptionPoint.X)` methods were silently dropped — it now registers them on the dispatcher with filters applied and `self` bound. Also tightens result typing across the tool-call seams so `mypy` stays green. * refactor: scope InterceptionPoint to the points this layer wires The dispatcher only fires the model- and tool-call boundaries, so `InterceptionPoint` now lists just those four rather than the full future catalog. New points are introduced alongside the seams that dispatch them, keeping every layer free of enum members with no live consumer. The dispatcher unit tests that borrowed unused points as generic examples are remapped onto the four kept points. * test: pin per-hook fail-open at the LLM and tool seams The dispatcher swallows a hook's exception per hook rather than around the whole loop, so one buggy hook no longer silently skips every hook registered after it. These seam-level tests pin that behavior through `_setup_before_llm_call_hooks` and `run_before/after_tool_call_hooks`, and confirm an intentional `return False` block still short-circuits later hooks. * fix: run execution-scoped hooks on the agent executor model seams `_setup_before/after_llm_call_hooks` only ran the executor's snapshot hook lists, so hooks registered via `scoped_hooks()` never fired on `PRE/POST_MODEL_CALL` during normal agent execution, while the tool seams (which go through `dispatch`) merged them. The seams now append the current scope's hooks after the snapshot via `get_scoped_hooks`, matching dispatch's global-then-scoped ordering, and a scoped-only registration no longer short-circuits the seam. --- .../src/crewai/agents/crew_agent_executor.py | 35 +- .../src/crewai/events/types/hook_events.py | 19 + .../src/crewai/experimental/agent_executor.py | 35 +- lib/crewai/src/crewai/hooks/__init__.py | 20 + lib/crewai/src/crewai/hooks/dispatch.py | 424 ++++++++++++++++++ lib/crewai/src/crewai/hooks/llm_hooks.py | 38 +- lib/crewai/src/crewai/hooks/tool_hooks.py | 83 +++- lib/crewai/src/crewai/llms/base_llm.py | 68 ++- lib/crewai/src/crewai/project/crew_base.py | 29 +- .../src/crewai/utilities/agent_utils.py | 99 ++-- lib/crewai/src/crewai/utilities/tool_utils.py | 107 ++--- .../tests/hooks/test_crew_scoped_hooks.py | 56 +++ lib/crewai/tests/hooks/test_dispatch.py | 296 ++++++++++++ lib/crewai/tests/hooks/test_llm_hooks.py | 173 +++++++ lib/crewai/tests/hooks/test_tool_hooks.py | 69 +++ 15 files changed, 1348 insertions(+), 203 deletions(-) create mode 100644 lib/crewai/src/crewai/events/types/hook_events.py create mode 100644 lib/crewai/src/crewai/hooks/dispatch.py create mode 100644 lib/crewai/tests/hooks/test_dispatch.py diff --git a/lib/crewai/src/crewai/agents/crew_agent_executor.py b/lib/crewai/src/crewai/agents/crew_agent_executor.py index de2315e3a..040520e1f 100644 --- a/lib/crewai/src/crewai/agents/crew_agent_executor.py +++ b/lib/crewai/src/crewai/agents/crew_agent_executor.py @@ -46,8 +46,8 @@ from crewai.hooks.llm_hooks import ( ) from crewai.hooks.tool_hooks import ( ToolCallHookContext, - get_after_tool_call_hooks, - get_before_tool_call_hooks, + run_after_tool_call_hooks, + run_before_tool_call_hooks, ) from crewai.types.callback import SerializableCallable from crewai.utilities.agent_utils import ( @@ -951,7 +951,6 @@ class CrewAgentExecutor(BaseAgentExecutor): track_delegation_if_needed(func_name, args_dict or {}, self.task) - hook_blocked = False before_hook_context = ToolCallHookContext( tool_name=func_name, tool_input=args_dict or {}, @@ -960,19 +959,7 @@ class CrewAgentExecutor(BaseAgentExecutor): task=self.task, crew=self.crew, ) - before_hooks = get_before_tool_call_hooks() - try: - for hook in before_hooks: - hook_result = hook(before_hook_context) - if hook_result is False: - hook_blocked = True - break - except Exception as hook_error: - if self.agent.verbose: - PRINTER.print( - content=f"Error in before_tool_call hook: {hook_error}", - color="red", - ) + hook_blocked = run_before_tool_call_hooks(before_hook_context) if hook_blocked: result = f"Tool execution blocked by hook. Tool: {func_name}" @@ -1033,19 +1020,9 @@ class CrewAgentExecutor(BaseAgentExecutor): tool_result=result, raw_tool_result=raw_tool_result, ) - after_hooks = get_after_tool_call_hooks() - try: - for after_hook in after_hooks: - after_hook_result = after_hook(after_hook_context) - if after_hook_result is not None: - result = after_hook_result - after_hook_context.tool_result = result - except Exception as hook_error: - if self.agent.verbose: - PRINTER.print( - content=f"Error in after_tool_call hook: {hook_error}", - color="red", - ) + modified_result = run_after_tool_call_hooks(after_hook_context) + if modified_result is not None: + result = modified_result if not error_event_emitted: crewai_event_bus.emit( diff --git a/lib/crewai/src/crewai/events/types/hook_events.py b/lib/crewai/src/crewai/events/types/hook_events.py new file mode 100644 index 000000000..b620bcfe1 --- /dev/null +++ b/lib/crewai/src/crewai/events/types/hook_events.py @@ -0,0 +1,19 @@ +from typing import Literal + +from crewai.events.base_events import BaseEvent + + +class HookDispatchedEvent(BaseEvent): + """Event emitted whenever an interception point dispatches to hooks. + + Only emitted when at least one hook is registered for the point, so the + no-op fast path stays free of event overhead. + """ + + type: Literal["hook_dispatched"] = "hook_dispatched" + interception_point: str + outcome: Literal["proceeded", "modified", "aborted"] + hook_count: int + duration_ms: float + abort_reason: str | None = None + abort_source: str | None = None diff --git a/lib/crewai/src/crewai/experimental/agent_executor.py b/lib/crewai/src/crewai/experimental/agent_executor.py index 3c76bd8d8..96308f9c9 100644 --- a/lib/crewai/src/crewai/experimental/agent_executor.py +++ b/lib/crewai/src/crewai/experimental/agent_executor.py @@ -62,8 +62,8 @@ from crewai.hooks.llm_hooks import ( ) from crewai.hooks.tool_hooks import ( ToolCallHookContext, - get_after_tool_call_hooks, - get_before_tool_call_hooks, + run_after_tool_call_hooks, + run_before_tool_call_hooks, ) from crewai.hooks.types import ( AfterLLMCallHookCallable, @@ -1975,7 +1975,6 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): track_delegation_if_needed(func_name, args_dict, self.task) - hook_blocked = False before_hook_context = ToolCallHookContext( tool_name=func_name, tool_input=args_dict, @@ -1984,19 +1983,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): task=self.task, crew=self.crew, ) - before_hooks = get_before_tool_call_hooks() - try: - for hook in before_hooks: - hook_result = hook(before_hook_context) - if hook_result is False: - hook_blocked = True - break - except Exception as hook_error: - if self.agent.verbose: - PRINTER.print( - content=f"Error in before_tool_call hook: {hook_error}", - color="red", - ) + hook_blocked = run_before_tool_call_hooks(before_hook_context) if hook_blocked: result = f"Tool execution blocked by hook. Tool: {func_name}" @@ -2060,19 +2047,9 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): tool_result=result, raw_tool_result=raw_tool_result, ) - after_hooks = get_after_tool_call_hooks() - try: - for after_hook in after_hooks: - after_hook_result = after_hook(after_hook_context) - if after_hook_result is not None: - result = after_hook_result - after_hook_context.tool_result = result - except Exception as hook_error: - if self.agent.verbose: - PRINTER.print( - content=f"Error in after_tool_call hook: {hook_error}", - color="red", - ) + modified_result = run_after_tool_call_hooks(after_hook_context) + if modified_result is not None: + result = modified_result if not error_event_emitted: crewai_event_bus.emit( diff --git a/lib/crewai/src/crewai/hooks/__init__.py b/lib/crewai/src/crewai/hooks/__init__.py index 5b9fbf7bd..16c6fe0fc 100644 --- a/lib/crewai/src/crewai/hooks/__init__.py +++ b/lib/crewai/src/crewai/hooks/__init__.py @@ -6,6 +6,17 @@ from crewai.hooks.decorators import ( before_llm_call, before_tool_call, ) +from crewai.hooks.dispatch import ( + HookAborted, + InterceptionPoint, + clear as clear_hooks, + clear_all as clear_all_hooks, + dispatch, + get_hooks, + on, + register as register_hook, + unregister as unregister_hook, +) from crewai.hooks.llm_hooks import ( LLMCallHookContext, clear_after_llm_call_hooks, @@ -74,6 +85,8 @@ def clear_all_global_hooks() -> dict[str, tuple[int, int]]: __all__ = [ + "HookAborted", + "InterceptionPoint", "LLMCallHookContext", "ToolCallHookContext", "after_llm_call", @@ -83,20 +96,27 @@ __all__ = [ "clear_after_llm_call_hooks", "clear_after_tool_call_hooks", "clear_all_global_hooks", + "clear_all_hooks", "clear_all_llm_call_hooks", "clear_all_tool_call_hooks", "clear_before_llm_call_hooks", "clear_before_tool_call_hooks", + "clear_hooks", + "dispatch", "get_after_llm_call_hooks", "get_after_tool_call_hooks", "get_before_llm_call_hooks", "get_before_tool_call_hooks", + "get_hooks", + "on", "register_after_llm_call_hook", "register_after_tool_call_hook", "register_before_llm_call_hook", "register_before_tool_call_hook", + "register_hook", "unregister_after_llm_call_hook", "unregister_after_tool_call_hook", "unregister_before_llm_call_hook", "unregister_before_tool_call_hook", + "unregister_hook", ] diff --git a/lib/crewai/src/crewai/hooks/dispatch.py b/lib/crewai/src/crewai/hooks/dispatch.py new file mode 100644 index 000000000..2822c73dd --- /dev/null +++ b/lib/crewai/src/crewai/hooks/dispatch.py @@ -0,0 +1,424 @@ +"""Generic interception-hook dispatcher. + +This module is the single engine behind every CrewAI interception point. A hook +receives a typed context, may mutate it in place and/or return a replacement +payload, and may raise :class:`HookAborted` to stop the intercepted operation +with a reason and source. + +The four public hook families (``before/after_llm_call`` and +``before/after_tool_call``) are adapters registered on this dispatcher, so the +legacy dialect (``register_*``/decorators/``return False``) and the new dialect +(``@on(point)`` / ``HookAborted``) share one ordered queue per point. + +Design notes: +- Global registration order is preserved; execution-scoped hooks (via + ``contextvars``) run after global ones, mirroring + ``events/event_bus.py``'s ``_runtime_state_var`` scoping pattern. +- ``dispatch`` has a no-op fast path (a single dict lookup) when no hooks are + registered for a point. +- Hooks are synchronous. They may be invoked from async seams, so they must not + block on heavy I/O (same restriction as the legacy hooks). +- ``HookAborted`` propagates by design. Any other exception raised by a hook is + swallowed (fail-open) to preserve the framework's protection against a buggy + user hook. +""" + +from __future__ import annotations + +from collections.abc import Callable, Iterator +from contextlib import contextmanager +import contextvars +from enum import Enum +from functools import wraps +import inspect +import time +from typing import Any + +from crewai.utilities.string_utils import sanitize_tool_name + + +class InterceptionPoint(str, Enum): + """Interception points wired by this layer. + + New points are introduced alongside the seams that dispatch them, so the + enum only ever lists points with a live consumer. This layer ships the + model- and tool-call boundaries, which back the legacy + ``before/after_llm_call`` and ``before/after_tool_call`` hooks. + """ + + # Model / tool boundaries (legacy-compatible) + PRE_MODEL_CALL = "pre_model_call" + POST_MODEL_CALL = "post_model_call" + PRE_TOOL_CALL = "pre_tool_call" + POST_TOOL_CALL = "post_tool_call" + + +class HookAborted(Exception): # noqa: N818 - public contract name from OSS-86 + """Raised by a hook (or a legacy adapter) to abort the intercepted operation. + + Args: + reason: Human-readable explanation of why the operation was aborted. + source: Optional identifier of the aborting hook (callable, string, or + any object). Used for telemetry and failure messages. + """ + + def __init__(self, reason: str, source: Any = None) -> None: + super().__init__(reason) + self.reason = reason + self.source = source + + +HookFn = Callable[[Any], Any] + +# (ctx, result) -> modified? A reducer maps a hook's return value onto the +# context using point-specific semantics. It may raise HookAborted. +Reducer = Callable[[Any, Any], bool] + + +_global_hooks: dict[InterceptionPoint, list[HookFn]] = { + point: [] for point in InterceptionPoint +} + +_scoped_hooks_var: contextvars.ContextVar[ + dict[InterceptionPoint, list[HookFn]] | None +] = contextvars.ContextVar("crewai_scoped_hooks", default=None) + + +_TELEMETRY_SOURCE = object() + + +def get_global_hook_list(point: InterceptionPoint) -> list[HookFn]: + """Return the live global hook list for a point. + + The returned list object is stable for the lifetime of the process, which + lets legacy modules alias their module-level registries to it. Mutate it in + place (append/remove/clear); never rebind it. + """ + return _global_hooks[point] + + +def register(point: InterceptionPoint, hook: HookFn) -> None: + """Register a global hook for an interception point.""" + _global_hooks[point].append(hook) + + +def unregister(point: InterceptionPoint, hook: HookFn) -> bool: + """Unregister a specific global hook. Returns True if it was removed. + + When ``hook`` was registered through :func:`on` with ``agents``/``tools`` + filters, the stored callable is a wrapper rather than ``hook`` itself. The + wrapper is stashed on ``hook._registered_hook`` at registration time, so it + can be resolved and removed here. + """ + hooks = _global_hooks[point] + target = hook if hook in hooks else getattr(hook, "_registered_hook", hook) + try: + hooks.remove(target) + return True + except ValueError: + return False + + +def get_hooks(point: InterceptionPoint) -> list[HookFn]: + """Return a copy of the global hooks registered for a point.""" + return _global_hooks[point].copy() + + +def clear(point: InterceptionPoint) -> int: + """Clear all global hooks for a point. Returns the number cleared.""" + count = len(_global_hooks[point]) + _global_hooks[point].clear() + return count + + +def clear_all() -> None: + """Clear all global hooks across every interception point.""" + for hooks in _global_hooks.values(): + hooks.clear() + + +@contextmanager +def scoped_hooks( + hooks: dict[InterceptionPoint, list[HookFn]] | None = None, +) -> Iterator[dict[InterceptionPoint, list[HookFn]]]: + """Enter an execution-scoped hook registry. + + Hooks registered inside this context (via :func:`register_scoped`) run after + global hooks and are discarded when the context exits. Mirrors the event + bus's scoped-handler pattern. + """ + scope: dict[InterceptionPoint, list[HookFn]] = hooks if hooks is not None else {} + token = _scoped_hooks_var.set(scope) + try: + yield scope + finally: + _scoped_hooks_var.reset(token) + + +def register_scoped(point: InterceptionPoint, hook: HookFn) -> None: + """Register a hook scoped to the current :func:`scoped_hooks` context.""" + scope = _scoped_hooks_var.get() + if scope is None: + raise RuntimeError( + "register_scoped() called outside of a scoped_hooks() context" + ) + scope.setdefault(point, []).append(hook) + + +def get_scoped_hooks(point: InterceptionPoint) -> list[HookFn]: + """Return the hooks registered in the current execution scope for a point. + + Used by seams that carry a pre-snapshotted hook list (e.g. the agent + executors' per-executor LLM hook lists) so they can merge in + execution-scoped hooks with the same snapshot-then-scoped ordering that + :func:`dispatch` applies to global vs scoped hooks. + """ + scope = _scoped_hooks_var.get() + if not scope: + return [] + return list(scope.get(point, [])) + + +def _resolve_hooks(point: InterceptionPoint) -> list[HookFn]: + """Resolve the ordered hooks for a point: global first, then scoped.""" + global_hooks = _global_hooks[point] + scope = _scoped_hooks_var.get() + if scope: + scoped = scope.get(point) + if scoped: + return [*global_hooks, *scoped] + return global_hooks + + +def _source_name(source: Any) -> str | None: + """Best-effort readable name for a hook source.""" + if source is None: + return None + if isinstance(source, str): + return source + name = getattr(source, "__name__", None) + if isinstance(name, str): + return name + return type(source).__name__ + + +def _emit_telemetry( + point: InterceptionPoint, + outcome: str, + hook_count: int, + duration_ms: float, + abort_reason: str | None, + abort_source: str | None, +) -> None: + """Emit a HookDispatchedEvent. Never raises.""" + try: + from crewai.events.event_bus import crewai_event_bus + from crewai.events.types.hook_events import HookDispatchedEvent + + crewai_event_bus.emit( + _TELEMETRY_SOURCE, + event=HookDispatchedEvent( + interception_point=point.value, + outcome=outcome, + hook_count=hook_count, + duration_ms=duration_ms, + abort_reason=abort_reason, + abort_source=abort_source, + ), + ) + except Exception: # noqa: S110 - telemetry must never break dispatch + pass + + +def _default_reducer(ctx: Any, result: Any) -> bool: + """Default payload semantics: a non-None return replaces ``ctx.payload``. + + Only reports a modification when the payload was actually applied, so a + context without a ``payload`` attribute does not produce a misleading + ``"modified"`` telemetry outcome. + """ + if result is not None and hasattr(ctx, "payload"): + ctx.payload = result + return True + return False + + +def _invoke_hook( + point: InterceptionPoint, + hook: HookFn, + ctx: Any, + reducer: Reducer, + verbose: bool, +) -> bool: + """Run a single hook and apply its result via the reducer. + + Returns whether the context was modified. Raises :class:`HookAborted` (with + ``source`` populated) to abort; any other exception is swallowed (fail-open). + """ + try: + result = hook(ctx) + return reducer(ctx, result) + except HookAborted as aborted: + if aborted.source is None: + aborted.source = hook + raise + except Exception as error: + if verbose: + from crewai_core.printer import PRINTER + + PRINTER.print( + content=f"Error in {point.value} hook: {error}", + color="yellow", + ) + return False + + +def run_hooks( + point: InterceptionPoint, + ctx: Any, + hooks: list[HookFn], + *, + reducer: Reducer | None = None, + verbose: bool = True, +) -> Any: + """Execute an explicit list of hooks against a context. + + This is the shared engine used both by :func:`dispatch` (which resolves + global + scoped hooks) and by seams that carry a pre-snapshotted hook list + (e.g. per-executor LLM hook lists). + + Args: + point: The interception point being dispatched. + ctx: The typed context passed to each hook (mutated in place). + hooks: The ordered hooks to run. + reducer: Maps each hook's return value onto ``ctx``. Defaults to + :func:`_default_reducer` (payload replacement). May raise + :class:`HookAborted`. + verbose: Whether to print swallowed-hook-error warnings. + + Returns: + The (possibly mutated) context. + + Raises: + HookAborted: If a hook or the reducer aborts the operation. Telemetry is + still emitted before propagating. + """ + if not hooks: + return ctx + + active_reducer = reducer if reducer is not None else _default_reducer + start = time.perf_counter() + outcome = "proceeded" + abort_reason: str | None = None + abort_source: str | None = None + modified = False + + try: + for hook in list(hooks): + if _invoke_hook(point, hook, ctx, active_reducer, verbose): + modified = True + outcome = "modified" if modified else "proceeded" + return ctx + except HookAborted as aborted: + outcome = "aborted" + abort_reason = aborted.reason + abort_source = _source_name(aborted.source) + raise + finally: + _emit_telemetry( + point, + outcome, + len(hooks), + (time.perf_counter() - start) * 1000.0, + abort_reason, + abort_source, + ) + + +def dispatch( + point: InterceptionPoint, + ctx: Any, + *, + reducer: Reducer | None = None, + verbose: bool = True, +) -> Any: + """Dispatch a context to all hooks registered for a point. + + Resolves global then scoped hooks and runs them through :func:`run_hooks`. + No-op fast path when nothing is registered. + """ + hooks = _resolve_hooks(point) + if not hooks: + return ctx + return run_hooks(point, ctx, hooks, reducer=reducer, verbose=verbose) + + +def _wrap_with_filters( + func: HookFn, + agents: list[str] | None, + tools: list[str] | None, +) -> HookFn: + """Wrap a hook so it only runs for matching agents/tools (context-shape aware).""" + + @wraps(func) + def filtered(ctx: Any) -> Any: + if tools: + tool_name = getattr(ctx, "tool_name", None) + if tool_name is not None and tool_name not in tools: + return None + if agents: + agent = getattr(ctx, "agent", None) + role = getattr(agent, "role", None) if agent is not None else None + if role is None: + role = getattr(ctx, "agent_role", None) + if role is not None and role not in agents: + return None + return func(ctx) + + return filtered + + +def on( + point: InterceptionPoint, + *, + agents: list[str] | None = None, + tools: list[str] | None = None, +) -> Callable[[HookFn], HookFn]: + """Register a function as a hook for an interception point. + + Mirrors the legacy decorators' ergonomics: supports ``agents=`` / ``tools=`` + filters and, when applied to a method inside a ``@CrewBase`` class, defers + registration to crew initialization (crew-scoped) instead of registering + globally. + + Example: + >>> @on(InterceptionPoint.PRE_TOOL_CALL, tools=["delete_file"]) + ... def guard(ctx): + ... raise HookAborted("deletion not allowed") + """ + normalized_tools = [sanitize_tool_name(t) for t in tools] if tools else None + + def decorator(func: HookFn) -> HookFn: + func._interception_point = point # type: ignore[attr-defined] + if normalized_tools: + func._filter_tools = normalized_tools # type: ignore[attr-defined] + if agents: + func._filter_agents = agents # type: ignore[attr-defined] + + params = list(inspect.signature(func).parameters.keys()) + is_method = len(params) >= 2 and params[0] == "self" + + if not is_method: + hook = ( + _wrap_with_filters(func, agents, normalized_tools) + if (agents or normalized_tools) + else func + ) + register(point, hook) + # Remember the actually-registered callable so unregister_hook(func) + # can resolve the filter wrapper. + func._registered_hook = hook # type: ignore[attr-defined] + + return func + + return decorator diff --git a/lib/crewai/src/crewai/hooks/llm_hooks.py b/lib/crewai/src/crewai/hooks/llm_hooks.py index 67108c01c..55563e84e 100644 --- a/lib/crewai/src/crewai/hooks/llm_hooks.py +++ b/lib/crewai/src/crewai/hooks/llm_hooks.py @@ -5,6 +5,11 @@ from typing import TYPE_CHECKING, Any, cast from crewai_core.printer import PRINTER from crewai.events.event_listener import event_listener +from crewai.hooks.dispatch import ( + HookAborted, + InterceptionPoint, + get_global_hook_list, +) from crewai.hooks.types import ( AfterLLMCallHookCallable, AfterLLMCallHookType, @@ -150,8 +155,37 @@ class LLMCallHookContext: event_listener.formatter.resume_live_updates() -_before_llm_call_hooks: list[BeforeLLMCallHookType | BeforeLLMCallHookCallable] = [] -_after_llm_call_hooks: list[AfterLLMCallHookType | AfterLLMCallHookCallable] = [] +# The legacy registries are aliased to the generic dispatcher's global hook +# lists for the model-call points, so legacy registrations and new-dialect +# ``@on(InterceptionPoint.PRE_MODEL_CALL)`` hooks share one ordered queue. +_before_llm_call_hooks: list[BeforeLLMCallHookType | BeforeLLMCallHookCallable] = ( + get_global_hook_list(InterceptionPoint.PRE_MODEL_CALL) +) +_after_llm_call_hooks: list[AfterLLMCallHookType | AfterLLMCallHookCallable] = ( + get_global_hook_list(InterceptionPoint.POST_MODEL_CALL) +) + + +def before_llm_call_reducer(context: LLMCallHookContext, result: object) -> bool: + """Legacy calling convention for ``pre_model_call`` hooks. + + A ``False`` return aborts the call (mapped to :class:`HookAborted`); messages + are modified in place, so no payload replacement occurs here. + """ + if result is False: + raise HookAborted(reason="before_llm_call hook returned False") + return False + + +def after_llm_call_reducer(context: LLMCallHookContext, result: object) -> bool: + """Legacy calling convention for ``post_model_call`` hooks. + + A non-empty string return replaces the response on the context. + """ + if result is not None and isinstance(result, str): + context.response = result + return True + return False def register_before_llm_call_hook( diff --git a/lib/crewai/src/crewai/hooks/tool_hooks.py b/lib/crewai/src/crewai/hooks/tool_hooks.py index a860c01d4..a4509bce7 100644 --- a/lib/crewai/src/crewai/hooks/tool_hooks.py +++ b/lib/crewai/src/crewai/hooks/tool_hooks.py @@ -5,6 +5,12 @@ from typing import TYPE_CHECKING, Any from crewai_core.printer import PRINTER from crewai.events.event_listener import event_listener +from crewai.hooks.dispatch import ( + HookAborted, + InterceptionPoint, + dispatch, + get_global_hook_list, +) from crewai.hooks.types import ( AfterToolCallHookCallable, AfterToolCallHookType, @@ -121,8 +127,81 @@ class ToolCallHookContext: event_listener.formatter.resume_live_updates() -_before_tool_call_hooks: list[BeforeToolCallHookType | BeforeToolCallHookCallable] = [] -_after_tool_call_hooks: list[AfterToolCallHookType | AfterToolCallHookCallable] = [] +# The legacy registries are aliased to the generic dispatcher's global hook +# lists for the tool-call points, so legacy registrations and new-dialect +# ``@on(InterceptionPoint.PRE_TOOL_CALL)`` hooks share one ordered queue. +_before_tool_call_hooks: list[BeforeToolCallHookType | BeforeToolCallHookCallable] = ( + get_global_hook_list(InterceptionPoint.PRE_TOOL_CALL) +) +_after_tool_call_hooks: list[AfterToolCallHookType | AfterToolCallHookCallable] = ( + get_global_hook_list(InterceptionPoint.POST_TOOL_CALL) +) + + +def before_tool_call_reducer(context: ToolCallHookContext, result: object) -> bool: + """Legacy calling convention for ``pre_tool_call`` hooks. + + A ``False`` return blocks the call (mapped to :class:`HookAborted`); tool + input is modified in place, so no payload replacement occurs here. + """ + if result is False: + raise HookAborted(reason="before_tool_call hook returned False") + return False + + +def after_tool_call_reducer(context: ToolCallHookContext, result: object) -> bool: + """Legacy calling convention for ``post_tool_call`` hooks. + + A non-None return replaces the tool result on the context. + """ + if isinstance(result, str): + context.tool_result = result + return True + return False + + +def _hook_verbose(context: ToolCallHookContext) -> bool: + """Whether swallowed-hook-error warnings should be printed. + + Mirrors the pre-dispatcher behavior where a failing tool hook surfaced a + warning when the executing agent was verbose. + """ + return bool(getattr(context.agent, "verbose", False)) + + +def run_before_tool_call_hooks(context: ToolCallHookContext) -> bool: + """Run all ``pre_tool_call`` hooks against a context. + + Returns: + True if a hook blocked execution (returned False or raised + :class:`HookAborted`), False otherwise. Tool input mutations on the + context persist regardless. + """ + try: + dispatch( + InterceptionPoint.PRE_TOOL_CALL, + context, + reducer=before_tool_call_reducer, + verbose=_hook_verbose(context), + ) + return False + except HookAborted: + return True + + +def run_after_tool_call_hooks(context: ToolCallHookContext) -> str | None: + """Run all ``post_tool_call`` hooks against a context. + + Returns: + The (possibly modified) tool result carried on the context. + """ + dispatch( + InterceptionPoint.POST_TOOL_CALL, + context, + reducer=after_tool_call_reducer, + verbose=_hook_verbose(context), + ) + return context.tool_result def register_before_tool_call_hook( diff --git a/lib/crewai/src/crewai/llms/base_llm.py b/lib/crewai/src/crewai/llms/base_llm.py index ece84fde0..a71126f58 100644 --- a/lib/crewai/src/crewai/llms/base_llm.py +++ b/lib/crewai/src/crewai/llms/base_llm.py @@ -1007,15 +1007,14 @@ class BaseLLM(BaseModel, ABC): from crewai_core.printer import PRINTER + from crewai.hooks.dispatch import HookAborted, InterceptionPoint, dispatch from crewai.hooks.llm_hooks import ( LLMCallHookContext, - get_before_llm_call_hooks, + before_llm_call_reducer, ) - before_hooks = get_before_llm_call_hooks() - if not before_hooks: - return True - + # No early global-list guard: dispatch resolves global + execution-scoped + # hooks and has its own no-op fast path, so scoped hooks still run here. hook_context = LLMCallHookContext( executor=None, messages=messages, @@ -1024,24 +1023,19 @@ class BaseLLM(BaseModel, ABC): task=None, crew=None, ) - verbose = getattr(from_agent, "verbose", True) if from_agent else True try: - for hook in before_hooks: - result = hook(hook_context) - if result is False: - if verbose: - PRINTER.print( - content="LLM call blocked by before_llm_call hook", - color="yellow", - ) - return False - except Exception as e: - if verbose: - PRINTER.print( - content=f"Error in before_llm_call hook: {e}", - color="yellow", - ) + dispatch( + InterceptionPoint.PRE_MODEL_CALL, + hook_context, + reducer=before_llm_call_reducer, + ) + except HookAborted: + PRINTER.print( + content="LLM call blocked by before_llm_call hook", + color="yellow", + ) + return False return True @@ -1074,17 +1068,14 @@ class BaseLLM(BaseModel, ABC): if from_agent is not None or not isinstance(response, str): return response - from crewai_core.printer import PRINTER - + from crewai.hooks.dispatch import InterceptionPoint, dispatch from crewai.hooks.llm_hooks import ( LLMCallHookContext, - get_after_llm_call_hooks, + after_llm_call_reducer, ) - after_hooks = get_after_llm_call_hooks() - if not after_hooks: - return response - + # No early global-list guard: dispatch resolves global + execution-scoped + # hooks and has its own no-op fast path, so scoped hooks still run here. hook_context = LLMCallHookContext( executor=None, messages=messages, @@ -1094,20 +1085,11 @@ class BaseLLM(BaseModel, ABC): crew=None, response=response, ) - verbose = getattr(from_agent, "verbose", True) if from_agent else True - modified_response = response - try: - for hook in after_hooks: - result = hook(hook_context) - if result is not None and isinstance(result, str): - modified_response = result - hook_context.response = modified_response - except Exception as e: - if verbose: - PRINTER.print( - content=f"Error in after_llm_call hook: {e}", - color="yellow", - ) + dispatch( + InterceptionPoint.POST_MODEL_CALL, + hook_context, + reducer=after_llm_call_reducer, + ) - return modified_response + return hook_context.response if hook_context.response is not None else response diff --git a/lib/crewai/src/crewai/project/crew_base.py b/lib/crewai/src/crewai/project/crew_base.py index 2d35a35b3..fa945f0c5 100644 --- a/lib/crewai/src/crewai/project/crew_base.py +++ b/lib/crewai/src/crewai/project/crew_base.py @@ -452,7 +452,15 @@ def _register_crew_hooks(instance: CrewInstance, cls: type) -> None: ) } - if not hook_methods: + # Methods decorated with @on(InterceptionPoint.X) carry ``_interception_point`` + # instead of the legacy markers above. + on_methods = { + name: method + for name, method in cls.__dict__.items() + if hasattr(method, "_interception_point") + } + + if not hook_methods and not on_methods: return from crewai.hooks import ( @@ -588,6 +596,25 @@ def _register_crew_hooks(instance: CrewInstance, cls: type) -> None: ("after_tool_call", after_tool_hook) ) + if on_methods: + from crewai.hooks.dispatch import ( + _wrap_with_filters, + register as register_interception_hook, + ) + + for on_method in on_methods.values(): + point = on_method._interception_point + bound_hook = on_method.__get__(instance, cls) + tools_filter = getattr(on_method, "_filter_tools", None) + agents_filter = getattr(on_method, "_filter_agents", None) + hook = ( + _wrap_with_filters(bound_hook, agents_filter, tools_filter) + if (tools_filter or agents_filter) + else bound_hook + ) + register_interception_hook(point, hook) + instance._registered_hook_functions.append((point.value, hook)) + instance._hooks_being_registered = False diff --git a/lib/crewai/src/crewai/utilities/agent_utils.py b/lib/crewai/src/crewai/utilities/agent_utils.py index 8a5fd9a07..e3fa001dc 100644 --- a/lib/crewai/src/crewai/utilities/agent_utils.py +++ b/lib/crewai/src/crewai/utilities/agent_utils.py @@ -1453,8 +1453,8 @@ def execute_single_native_tool_call( ) from crewai.hooks.tool_hooks import ( ToolCallHookContext, - get_after_tool_call_hooks, - get_before_tool_call_hooks, + run_after_tool_call_hooks, + run_before_tool_call_hooks, ) info = extract_tool_call_info(tool_call) @@ -1517,7 +1517,6 @@ def execute_single_native_tool_call( track_delegation_if_needed(func_name, args_dict, task) - hook_blocked = False before_hook_context = ToolCallHookContext( tool_name=func_name, tool_input=args_dict, @@ -1526,13 +1525,7 @@ def execute_single_native_tool_call( task=task, crew=crew, ) - try: - for hook in get_before_tool_call_hooks(): - if hook(before_hook_context) is False: - hook_blocked = True - break - except Exception: # noqa: S110 - pass + hook_blocked = run_before_tool_call_hooks(before_hook_context) error_event_emitted = False if hook_blocked: @@ -1587,14 +1580,9 @@ def execute_single_native_tool_call( tool_result=result, raw_tool_result=raw_tool_result, ) - try: - for after_hook in get_after_tool_call_hooks(): - hook_result = after_hook(after_hook_context) - if hook_result is not None: - result = hook_result - after_hook_context.tool_result = result - except Exception: # noqa: S110 - pass + modified_result = run_after_tool_call_hooks(after_hook_context) + if modified_result is not None: + result = modified_result if not error_event_emitted: crewai_event_bus.emit( @@ -1690,28 +1678,42 @@ def _setup_before_llm_call_hooks( Returns: True if LLM execution should proceed, False if blocked by a hook. """ - if executor_context and executor_context.before_llm_call_hooks: - from crewai.hooks.llm_hooks import LLMCallHookContext + if executor_context: + from crewai.hooks.dispatch import ( + HookAborted, + InterceptionPoint, + get_scoped_hooks, + run_hooks, + ) + from crewai.hooks.llm_hooks import LLMCallHookContext, before_llm_call_reducer + + # Executor snapshot first, then execution-scoped hooks — the same + # ordering dispatch() applies to global vs scoped hooks. + hooks: list[Any] = [ + *executor_context.before_llm_call_hooks, + *get_scoped_hooks(InterceptionPoint.PRE_MODEL_CALL), + ] + if not hooks: + return True original_messages = executor_context.messages hook_context = LLMCallHookContext(executor_context) try: - for hook in executor_context.before_llm_call_hooks: - result = hook(hook_context) - if result is False: - if verbose: - printer.print( - content="LLM call blocked by before_llm_call hook", - color="yellow", - ) - return False - except Exception as e: + run_hooks( + InterceptionPoint.PRE_MODEL_CALL, + hook_context, + hooks, + reducer=before_llm_call_reducer, + verbose=verbose, + ) + except HookAborted: if verbose: printer.print( - content=f"Error in before_llm_call hook: {e}", + content="LLM call blocked by before_llm_call hook", color="yellow", ) + return False if not isinstance(executor_context.messages, list): if verbose: @@ -1748,8 +1750,9 @@ def _setup_after_llm_call_hooks( Returns: The potentially modified response (string or Pydantic model). """ - if executor_context and executor_context.after_llm_call_hooks: - from crewai.hooks.llm_hooks import LLMCallHookContext + if executor_context: + from crewai.hooks.dispatch import InterceptionPoint, get_scoped_hooks, run_hooks + from crewai.hooks.llm_hooks import LLMCallHookContext, after_llm_call_reducer # Don't stringify structured tool-call payloads: the executor would # treat the result as a final answer and skip tool execution (#6529). @@ -1757,6 +1760,15 @@ def _setup_after_llm_call_hooks( if not isinstance(answer, (str, BaseModel)): return answer + # Executor snapshot first, then execution-scoped hooks — the same + # ordering dispatch() applies to global vs scoped hooks. + hooks: list[Any] = [ + *executor_context.after_llm_call_hooks, + *get_scoped_hooks(InterceptionPoint.POST_MODEL_CALL), + ] + if not hooks: + return answer + original_messages = executor_context.messages if isinstance(answer, BaseModel): @@ -1768,18 +1780,15 @@ def _setup_after_llm_call_hooks( hook_response = str(answer) hook_context = LLMCallHookContext(executor_context, response=hook_response) - try: - for hook in executor_context.after_llm_call_hooks: - modified_response = hook(hook_context) - if modified_response is not None and isinstance(modified_response, str): - hook_response = modified_response - - except Exception as e: - if verbose: - printer.print( - content=f"Error in after_llm_call hook: {e}", - color="yellow", - ) + run_hooks( + InterceptionPoint.POST_MODEL_CALL, + hook_context, + hooks, + reducer=after_llm_call_reducer, + verbose=verbose, + ) + if hook_context.response is not None: + hook_response = hook_context.response if not isinstance(executor_context.messages, list): if verbose: diff --git a/lib/crewai/src/crewai/utilities/tool_utils.py b/lib/crewai/src/crewai/utilities/tool_utils.py index dcb25594c..2d6e3c142 100644 --- a/lib/crewai/src/crewai/utilities/tool_utils.py +++ b/lib/crewai/src/crewai/utilities/tool_utils.py @@ -6,15 +6,14 @@ from crewai.agents.parser import AgentAction from crewai.agents.tools_handler import ToolsHandler from crewai.hooks.tool_hooks import ( ToolCallHookContext, - get_after_tool_call_hooks, - get_before_tool_call_hooks, + run_after_tool_call_hooks, + run_before_tool_call_hooks, ) from crewai.security.fingerprint import Fingerprint from crewai.tools.structured_tool import CrewStructuredTool from crewai.tools.tool_types import ToolResult from crewai.tools.tool_usage import ToolUsage, ToolUsageError from crewai.utilities.i18n import I18N_DEFAULT -from crewai.utilities.logger import Logger from crewai.utilities.string_utils import sanitize_tool_name @@ -57,11 +56,10 @@ async def aexecute_tool_and_check_finality( fingerprint_context: Optional context for fingerprinting. crew: Optional crew instance for hook context. - Returns: + Returns: ToolResult containing the execution result and whether it should be treated as a final answer. """ - logger = Logger(verbose=crew.verbose if crew else False) tool_name_to_tool_map = {sanitize_tool_name(tool.name): tool for tool in tools} if agent_key and agent_role and agent: @@ -102,18 +100,27 @@ async def aexecute_tool_and_check_finality( crew=crew, ) - before_hooks = get_before_tool_call_hooks() - try: - for hook in before_hooks: - result = hook(hook_context) - if result is False: - blocked_message = ( - f"Tool execution blocked by hook. " - f"Tool: {tool_calling.tool_name}" - ) - return ToolResult(blocked_message, False) - except Exception as e: - logger.log("error", f"Error in before_tool_call hook: {e}") + if run_before_tool_call_hooks(hook_context): + blocked_message = ( + f"Tool execution blocked by hook. Tool: {tool_calling.tool_name}" + ) + # Run POST_TOOL_CALL even on a blocked call so monitoring hooks + # still fire, matching the native tool-call paths. + blocked_hook_context = ToolCallHookContext( + tool_name=sanitized_tool_name, + tool_input=tool_input, + tool=tool, + agent=agent, + task=task, + crew=crew, + tool_result=blocked_message, + raw_tool_result=blocked_message, + ) + modified_result = run_after_tool_call_hooks(blocked_hook_context) + return ToolResult( + modified_result if modified_result is not None else blocked_message, + False, + ) tool_result = await tool_usage.ause(tool_calling, agent_action.text) raw_tool_result = tool_usage.get_last_raw_result(tool_result) @@ -129,18 +136,12 @@ async def aexecute_tool_and_check_finality( raw_tool_result=raw_tool_result, ) - after_hooks = get_after_tool_call_hooks() - modified_result: str = tool_result - try: - for after_hook in after_hooks: - hook_result = after_hook(after_hook_context) - if hook_result is not None: - modified_result = hook_result - after_hook_context.tool_result = modified_result - except Exception as e: - logger.log("error", f"Error in after_tool_call hook: {e}") + modified_result = run_after_tool_call_hooks(after_hook_context) - return ToolResult(modified_result, tool.result_as_answer) + return ToolResult( + modified_result if modified_result is not None else tool_result, + tool.result_as_answer, + ) tool_result = I18N_DEFAULT.errors("wrong_tool_name").format( tool=sanitized_tool_name, @@ -181,7 +182,6 @@ def execute_tool_and_check_finality( Returns: ToolResult containing the execution result and whether it should be treated as a final answer """ - logger = Logger(verbose=crew.verbose if crew else False) tool_name_to_tool_map = {sanitize_tool_name(tool.name): tool for tool in tools} if agent_key and agent_role and agent: @@ -222,18 +222,27 @@ def execute_tool_and_check_finality( crew=crew, ) - before_hooks = get_before_tool_call_hooks() - try: - for hook in before_hooks: - result = hook(hook_context) - if result is False: - blocked_message = ( - f"Tool execution blocked by hook. " - f"Tool: {tool_calling.tool_name}" - ) - return ToolResult(blocked_message, False) - except Exception as e: - logger.log("error", f"Error in before_tool_call hook: {e}") + if run_before_tool_call_hooks(hook_context): + blocked_message = ( + f"Tool execution blocked by hook. Tool: {tool_calling.tool_name}" + ) + # Run POST_TOOL_CALL even on a blocked call so monitoring hooks + # still fire, matching the native tool-call paths. + blocked_hook_context = ToolCallHookContext( + tool_name=sanitized_tool_name, + tool_input=tool_input, + tool=tool, + agent=agent, + task=task, + crew=crew, + tool_result=blocked_message, + raw_tool_result=blocked_message, + ) + modified_result = run_after_tool_call_hooks(blocked_hook_context) + return ToolResult( + modified_result if modified_result is not None else blocked_message, + False, + ) tool_result = tool_usage.use(tool_calling, agent_action.text) raw_tool_result = tool_usage.get_last_raw_result(tool_result) @@ -249,18 +258,12 @@ def execute_tool_and_check_finality( raw_tool_result=raw_tool_result, ) - after_hooks = get_after_tool_call_hooks() - modified_result: str = tool_result - try: - for after_hook in after_hooks: - hook_result = after_hook(after_hook_context) - if hook_result is not None: - modified_result = hook_result - after_hook_context.tool_result = modified_result - except Exception as e: - logger.log("error", f"Error in after_tool_call hook: {e}") + modified_result = run_after_tool_call_hooks(after_hook_context) - return ToolResult(modified_result, tool.result_as_answer) + return ToolResult( + modified_result if modified_result is not None else tool_result, + tool.result_as_answer, + ) tool_result = I18N_DEFAULT.errors("wrong_tool_name").format( tool=sanitized_tool_name, diff --git a/lib/crewai/tests/hooks/test_crew_scoped_hooks.py b/lib/crewai/tests/hooks/test_crew_scoped_hooks.py index cf567a333..b0f8e1eb2 100644 --- a/lib/crewai/tests/hooks/test_crew_scoped_hooks.py +++ b/lib/crewai/tests/hooks/test_crew_scoped_hooks.py @@ -306,6 +306,62 @@ class TestCrewScopedHooks: assert len(execution_log) == 1 +class TestCrewOnDecoratedMethods: + """@on(InterceptionPoint.X) methods inside @CrewBase must register. + + Regression: CrewBase only scanned the legacy ``is_*_hook`` markers, so + methods decorated with the generic ``@on`` decorator (which sets + ``_interception_point``) were silently dropped and never ran. + """ + + def test_on_decorated_method_registers_and_binds_self(self): + from crewai.hooks import InterceptionPoint, on + from crewai.hooks.dispatch import _resolve_hooks + + execution_log = [] + + @CrewBase + class TestCrew: + def __init__(self): + self.name = "on-crew" + + @on(InterceptionPoint.PRE_MODEL_CALL) + def on_pre_model(self, context): + execution_log.append(self.name) + + @agent + def researcher(self): + return Agent(role="Researcher", goal="Research", backstory="Expert") + + @crew + def crew(self): + return Crew(agents=self.agents, tasks=[], verbose=False) + + before = len(_resolve_hooks(InterceptionPoint.PRE_MODEL_CALL)) + + instance = TestCrew() + + hooks = _resolve_hooks(InterceptionPoint.PRE_MODEL_CALL) + assert len(hooks) == before + 1 + + assert ( + InterceptionPoint.PRE_MODEL_CALL.value, + hooks[-1], + ) in instance._registered_hook_functions + + mock_executor = Mock() + mock_executor.messages = [] + mock_executor.agent = Mock(role="Test") + mock_executor.task = Mock() + mock_executor.crew = Mock() + mock_executor.llm = Mock() + mock_executor.iterations = 0 + + hooks[-1](LLMCallHookContext(executor=mock_executor)) + + assert execution_log == ["on-crew"] + + class TestCrewScopedHookAttributes: """Test that crew-scoped hooks have correct attributes set.""" diff --git a/lib/crewai/tests/hooks/test_dispatch.py b/lib/crewai/tests/hooks/test_dispatch.py new file mode 100644 index 000000000..9aa5d5912 --- /dev/null +++ b/lib/crewai/tests/hooks/test_dispatch.py @@ -0,0 +1,296 @@ +"""Unit tests for the generic interception-hook dispatcher. + +These cover the new contract (payload-in/payload-out + HookAborted), the shared +ordered queue between the legacy and new dialects on the four model/tool points, +execution-scoped hooks, fail-open exception handling, telemetry, and the no-op +fast-path overhead budget. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import time + +from crewai.events.event_bus import crewai_event_bus +from crewai.events.types.hook_events import HookDispatchedEvent +from crewai.hooks.dispatch import ( + HookAborted, + InterceptionPoint, + clear_all, + dispatch, + get_hooks, + on, + register, + register_scoped, + scoped_hooks, + unregister as unregister_hook, +) +from crewai.hooks.llm_hooks import ( + get_before_llm_call_hooks, + register_before_llm_call_hook, +) +import pytest + + +@dataclass +class _Ctx: + payload: object = None + tool_name: str | None = None + agent: object = None + agent_role: str | None = None + + +@pytest.fixture(autouse=True) +def clear_dispatch_registry(): + """Ensure every test starts and ends with an empty global registry.""" + clear_all() + yield + clear_all() + + +class TestDispatchContract: + """The core payload-in/payload-out + HookAborted contract.""" + + def test_noop_fast_path_returns_context_unchanged(self): + ctx = _Ctx(payload="original") + result = dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx) + assert result is ctx + assert ctx.payload == "original" + + def test_return_value_replaces_payload(self): + def double(ctx): + return ctx.payload * 2 + + register(InterceptionPoint.PRE_MODEL_CALL, double) + ctx = _Ctx(payload="ab") + dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx) + assert ctx.payload == "abab" + + def test_in_place_mutation_is_honored(self): + def mutate(ctx): + ctx.payload.append(1) + return None + + register(InterceptionPoint.PRE_MODEL_CALL, mutate) + ctx = _Ctx(payload=[]) + dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx) + assert ctx.payload == [1] + + def test_hooks_run_in_registration_order(self): + order: list[int] = [] + register(InterceptionPoint.PRE_MODEL_CALL, lambda ctx: order.append(1)) + register(InterceptionPoint.PRE_MODEL_CALL, lambda ctx: order.append(2)) + dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx()) + assert order == [1, 2] + + def test_hook_aborted_propagates_with_reason_and_source(self): + def blocker(ctx): + raise HookAborted(reason="nope", source="policy") + + register(InterceptionPoint.PRE_MODEL_CALL, blocker) + with pytest.raises(HookAborted) as exc: + dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx()) + assert exc.value.reason == "nope" + assert exc.value.source == "policy" + + def test_ordinary_exception_is_swallowed_and_later_hooks_run(self): + ran: list[str] = [] + + def boom(ctx): + ran.append("boom") + raise ValueError("bug in user hook") + + def after(ctx): + ran.append("after") + + register(InterceptionPoint.PRE_MODEL_CALL, boom) + register(InterceptionPoint.PRE_MODEL_CALL, after) + dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx(), verbose=False) + assert ran == ["boom", "after"] + + +class TestOnDecorator: + """The @on decorator registers and filters like the legacy decorators.""" + + def test_on_registers_global_hook(self): + @on(InterceptionPoint.POST_TOOL_CALL) + def hook(ctx): + return None + + assert hook in get_hooks(InterceptionPoint.POST_TOOL_CALL) + + def test_tool_filter_skips_non_matching_tools(self): + seen: list[str] = [] + + @on(InterceptionPoint.PRE_TOOL_CALL, tools=["allowed_tool"]) + def hook(ctx): + seen.append(ctx.tool_name) + + dispatch(InterceptionPoint.PRE_TOOL_CALL, _Ctx(tool_name="other_tool")) + dispatch(InterceptionPoint.PRE_TOOL_CALL, _Ctx(tool_name="allowed_tool")) + assert seen == ["allowed_tool"] + + def test_agent_filter_skips_non_matching_agents(self): + seen: list[str] = [] + + class _Agent: + def __init__(self, role): + self.role = role + + @on(InterceptionPoint.PRE_MODEL_CALL, agents=["Researcher"]) + def hook(ctx): + seen.append(ctx.agent.role) + + dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx(agent=_Agent("Writer"))) + dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx(agent=_Agent("Researcher"))) + assert seen == ["Researcher"] + + def test_agent_filter_falls_back_to_agent_role(self): + seen: list[str] = [] + + @on(InterceptionPoint.PRE_TOOL_CALL, agents=["Researcher"]) + def hook(ctx): + seen.append(ctx.agent_role) + + # No agent object, only the agent_role string (e.g. flow seams). + dispatch(InterceptionPoint.PRE_TOOL_CALL, _Ctx(agent_role="Writer")) + dispatch(InterceptionPoint.PRE_TOOL_CALL, _Ctx(agent_role="Researcher")) + assert seen == ["Researcher"] + + def test_unregister_resolves_filtered_wrapper(self): + @on(InterceptionPoint.PRE_TOOL_CALL, tools=["allowed_tool"]) + def hook(ctx): + return None + + assert len(get_hooks(InterceptionPoint.PRE_TOOL_CALL)) == 1 + assert unregister_hook(InterceptionPoint.PRE_TOOL_CALL, hook) is True + assert get_hooks(InterceptionPoint.PRE_TOOL_CALL) == [] + + +class TestSharedQueueWithLegacyDialect: + """Legacy registrations and @on hooks compose in one ordered queue.""" + + def test_on_and_legacy_share_pre_model_call_queue(self): + def legacy(ctx): + return None + + @on(InterceptionPoint.PRE_MODEL_CALL) + def modern(ctx): + return None + + register_before_llm_call_hook(legacy) + + queue = get_before_llm_call_hooks() + assert modern in queue + assert legacy in queue + # registration order preserved: modern registered before legacy + assert queue.index(modern) < queue.index(legacy) + + +class TestScopedHooks: + """Execution-scoped hooks run after globals and are discarded on exit.""" + + def test_scoped_runs_after_global_then_cleared(self): + order: list[str] = [] + register(InterceptionPoint.POST_MODEL_CALL, lambda ctx: order.append("global")) + + with scoped_hooks(): + register_scoped(InterceptionPoint.POST_MODEL_CALL, lambda ctx: order.append("scoped")) + dispatch(InterceptionPoint.POST_MODEL_CALL, _Ctx()) + + # outside the scope the scoped hook is gone + dispatch(InterceptionPoint.POST_MODEL_CALL, _Ctx()) + + assert order == ["global", "scoped", "global"] + + +class TestTelemetry: + """dispatch emits a HookDispatchedEvent only when hooks ran.""" + + def test_no_event_on_empty_fast_path(self): + events: list[HookDispatchedEvent] = [] + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(HookDispatchedEvent) + def _capture(_source, event): + events.append(event) + + dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx()) + + assert events == [] + + def test_event_reports_outcome(self): + events: list[HookDispatchedEvent] = [] + + register(InterceptionPoint.PRE_MODEL_CALL, lambda ctx: "changed") + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(HookDispatchedEvent) + def _capture(_source, event): + events.append(event) + + dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx()) + # Telemetry handlers run on the bus's thread pool; flush so the + # assertion doesn't race the emit. + crewai_event_bus.flush() + + assert len(events) == 1 + assert events[0].interception_point == "pre_model_call" + assert events[0].outcome == "modified" + assert events[0].hook_count == 1 + + def test_event_reports_abort_outcome(self): + events: list[HookDispatchedEvent] = [] + + def blocker(ctx): + raise HookAborted(reason="blocked", source="policy") + + register(InterceptionPoint.PRE_MODEL_CALL, blocker) + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(HookDispatchedEvent) + def _capture(_source, event): + events.append(event) + + with pytest.raises(HookAborted): + dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx()) + crewai_event_bus.flush() + + assert len(events) == 1 + assert events[0].interception_point == "pre_model_call" + assert events[0].outcome == "aborted" + assert events[0].abort_reason == "blocked" + assert events[0].abort_source == "policy" + + +class TestNoOpOverhead: + """The no-op fast path must stay cheap (a single dict lookup).""" + + def test_noop_dispatch_overhead_is_bounded(self): + # Relative (not absolute) budget: the no-op fast path is a dict lookup + # plus a guard, so it should stay within a wide multiple of a bare + # function call. This catches accidental O(n) regressions without + # depending on absolute timing on shared CI runners. + ctx = _Ctx() + iterations = 100_000 + + def _baseline(_c): + return _c + + for _ in range(1000): # warm up both paths + dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx) + _baseline(ctx) + + start = time.perf_counter() + for _ in range(iterations): + _baseline(ctx) + baseline = time.perf_counter() - start + + start = time.perf_counter() + for _ in range(iterations): + dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx) + noop = time.perf_counter() - start + + assert noop < baseline * 50 + 5e-3 diff --git a/lib/crewai/tests/hooks/test_llm_hooks.py b/lib/crewai/tests/hooks/test_llm_hooks.py index 7b8add697..0c917712d 100644 --- a/lib/crewai/tests/hooks/test_llm_hooks.py +++ b/lib/crewai/tests/hooks/test_llm_hooks.py @@ -337,6 +337,105 @@ class TestLLMHooksIntegration: hooks = get_before_llm_call_hooks() assert len(hooks) == 0 + def test_raising_before_hook_does_not_skip_later_hooks(self, mock_executor): + """Fail-open is per-hook: a crashing hook must not disable its neighbors. + + Regression guard for the dispatcher migration: previously the + ``except Exception`` wrapped the whole hook loop, so a raising hook + silently skipped every hook registered after it. Now swallowing is + per-hook — later hooks still run and the LLM call still proceeds. + """ + from crewai.utilities.agent_utils import _setup_before_llm_call_hooks + + ran: list[str] = [] + + def crashing_hook(context): + ran.append("crashing") + raise ValueError("bug in user hook") + + def later_hook(context): + ran.append("later") + + register_before_llm_call_hook(crashing_hook) + register_before_llm_call_hook(later_hook) + mock_executor.before_llm_call_hooks = get_before_llm_call_hooks() + + proceed = _setup_before_llm_call_hooks( + mock_executor, printer=Mock(), verbose=False + ) + + assert ran == ["crashing", "later"] + assert proceed is True + + def test_scoped_hooks_fire_on_agent_executor_llm_seams(self, mock_executor): + """register_scoped hooks must run on the executor model seams. + + Regression: `_setup_before/after_llm_call_hooks` only ran the + executor's snapshot lists, so execution-scoped hooks never fired on + PRE/POST_MODEL_CALL during normal agent execution (while tool seams, + which go through `dispatch`, merged them). Scoped hooks run after the + snapshot, matching dispatch's global-then-scoped ordering. + """ + from crewai.hooks import InterceptionPoint + from crewai.hooks.dispatch import register_scoped, scoped_hooks + from crewai.utilities.agent_utils import ( + _setup_after_llm_call_hooks, + _setup_before_llm_call_hooks, + ) + + order: list[str] = [] + + def snapshot_hook(context): + order.append("snapshot") + + mock_executor.before_llm_call_hooks = [snapshot_hook] + mock_executor.after_llm_call_hooks = [] + + with scoped_hooks(): + register_scoped( + InterceptionPoint.PRE_MODEL_CALL, + lambda ctx: order.append("scoped_pre"), + ) + register_scoped( + InterceptionPoint.POST_MODEL_CALL, + lambda ctx: order.append("scoped_post"), + ) + + proceed = _setup_before_llm_call_hooks( + mock_executor, printer=Mock(), verbose=False + ) + answer = _setup_after_llm_call_hooks( + mock_executor, "answer", printer=Mock(), verbose=False + ) + + assert order == ["snapshot", "scoped_pre", "scoped_post"] + assert proceed is True + assert answer == "answer" + + def test_intentional_block_still_short_circuits_later_hooks(self, mock_executor): + """A hook returning False blocks the call and skips later hooks (unchanged).""" + from crewai.utilities.agent_utils import _setup_before_llm_call_hooks + + ran: list[str] = [] + + def blocking_hook(context): + ran.append("blocking") + return False + + def later_hook(context): + ran.append("later") + + register_before_llm_call_hook(blocking_hook) + register_before_llm_call_hook(later_hook) + mock_executor.before_llm_call_hooks = get_before_llm_call_hooks() + + proceed = _setup_before_llm_call_hooks( + mock_executor, printer=Mock(), verbose=False + ) + + assert ran == ["blocking"] + assert proceed is False + @pytest.mark.vcr() def test_lite_agent_hooks_integration_with_real_llm(self): """Test that LiteAgent executes before/after LLM call hooks and prints messages correctly.""" @@ -497,3 +596,77 @@ class TestLLMHooksIntegration: finally: unregister_before_llm_call_hook(before_hook) unregister_after_llm_call_hook(after_hook) + + +class TestDirectLLMScopedHooks: + """Direct (agent-less) LLM calls must honor execution-scoped hooks. + + Regression: the direct-call helpers used to short-circuit when the global + hook list was empty, so hooks registered only for the current + ``scoped_hooks()`` context never ran on this path. + """ + + @staticmethod + def _stub_llm(): + from crewai.llms.base_llm import BaseLLM + + class _StubLLM(BaseLLM): + def call(self, *args: object, **kwargs: object) -> str: + return "" + + return _StubLLM(model="stub") + + def test_scoped_before_hook_runs_on_direct_call(self): + from crewai.hooks import InterceptionPoint + from crewai.hooks.dispatch import register_scoped, scoped_hooks + + llm = self._stub_llm() + seen: list[int] = [] + + with scoped_hooks(): + register_scoped( + InterceptionPoint.PRE_MODEL_CALL, + lambda ctx: seen.append(len(ctx.messages)), + ) + proceed = llm._invoke_before_llm_call_hooks( + [{"role": "user", "content": "hi"}], from_agent=None + ) + + assert proceed is True + assert seen == [1] + + def test_scoped_before_hook_can_block_direct_call(self): + from crewai.hooks import InterceptionPoint + from crewai.hooks.dispatch import HookAborted, register_scoped, scoped_hooks + + llm = self._stub_llm() + + def block(ctx: LLMCallHookContext) -> None: + raise HookAborted(reason="blocked by scoped hook") + + with scoped_hooks(): + register_scoped(InterceptionPoint.PRE_MODEL_CALL, block) + proceed = llm._invoke_before_llm_call_hooks( + [{"role": "user", "content": "hi"}], from_agent=None + ) + + assert proceed is False + + def test_scoped_after_hook_modifies_direct_response(self): + from crewai.hooks import InterceptionPoint + from crewai.hooks.dispatch import register_scoped, scoped_hooks + + llm = self._stub_llm() + + def redact(ctx: LLMCallHookContext) -> str: + return ctx.response.replace("SECRET", "[REDACTED]") + + with scoped_hooks(): + register_scoped(InterceptionPoint.POST_MODEL_CALL, redact) + result = llm._invoke_after_llm_call_hooks( + [{"role": "user", "content": "hi"}], + "contains SECRET", + from_agent=None, + ) + + assert result == "contains [REDACTED]" diff --git a/lib/crewai/tests/hooks/test_tool_hooks.py b/lib/crewai/tests/hooks/test_tool_hooks.py index 14c6848a8..4978f14b8 100644 --- a/lib/crewai/tests/hooks/test_tool_hooks.py +++ b/lib/crewai/tests/hooks/test_tool_hooks.py @@ -576,6 +576,75 @@ class TestToolHooksIntegration: unregister_after_tool_call_hook(after_tool_call_hook) +class TestPerHookFailOpen: + """Fail-open is per-hook: a crashing hook must not disable its neighbors. + + Regression guards for the dispatcher migration: previously each seam's + ``except Exception`` wrapped the whole hook loop, so a raising hook + silently skipped every hook registered after it. + """ + + def test_raising_before_hook_does_not_skip_later_hooks_or_block( + self, mock_tool, mock_agent + ): + from crewai.hooks.tool_hooks import run_before_tool_call_hooks + + mock_agent.verbose = False + ran: list[str] = [] + + def crashing_hook(context): + ran.append("crashing") + raise ValueError("bug in user hook") + + def later_hook(context): + ran.append("later") + + register_before_tool_call_hook(crashing_hook) + register_before_tool_call_hook(later_hook) + + context = ToolCallHookContext( + tool_name="test_tool", + tool_input={"arg": "value"}, + tool=mock_tool, + agent=mock_agent, + ) + blocked = run_before_tool_call_hooks(context) + + assert ran == ["crashing", "later"] + assert blocked is False + + def test_raising_after_hook_does_not_skip_later_result_rewrites( + self, mock_tool, mock_agent + ): + from crewai.hooks.tool_hooks import run_after_tool_call_hooks + + mock_agent.verbose = False + ran: list[str] = [] + + def crashing_hook(context): + ran.append("crashing") + raise ValueError("bug in user hook") + + def rewriting_hook(context): + ran.append("rewriting") + return f"{context.tool_result} [rewritten]" + + register_after_tool_call_hook(crashing_hook) + register_after_tool_call_hook(rewriting_hook) + + context = ToolCallHookContext( + tool_name="test_tool", + tool_input={"arg": "value"}, + tool=mock_tool, + agent=mock_agent, + tool_result="original", + ) + result = run_after_tool_call_hooks(context) + + assert ran == ["crashing", "rewriting"] + assert result == "original [rewritten]" + + class TestNativeToolCallingHooksIntegration: """Integration tests for hooks with native function calling (Agent and Crew)."""