From c20d85a116992ca09aacbd9aab854d0bd1f0926f Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Sat, 11 Jul 2026 14:14:39 -0300 Subject: [PATCH 1/6] 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. --- .../src/crewai/agents/crew_agent_executor.py | 33 +- .../src/crewai/events/types/hook_events.py | 19 + .../src/crewai/experimental/agent_executor.py | 33 +- lib/crewai/src/crewai/hooks/__init__.py | 20 + lib/crewai/src/crewai/hooks/dispatch.py | 418 ++++++++++++++++++ lib/crewai/src/crewai/hooks/llm_hooks.py | 38 +- lib/crewai/src/crewai/hooks/tool_hooks.py | 74 +++- lib/crewai/src/crewai/llms/base_llm.py | 60 +-- .../src/crewai/utilities/agent_utils.py | 74 ++-- lib/crewai/src/crewai/utilities/tool_utils.py | 65 +-- lib/crewai/tests/hooks/test_dispatch.py | 229 ++++++++++ 11 files changed, 871 insertions(+), 192 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 de2315e3a9..730cf65a12 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.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 @@ def _execute_single_native_tool_call( 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 @@ def _execute_single_native_tool_call( 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,7 @@ def _execute_single_native_tool_call( 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", - ) + result = run_after_tool_call_hooks(after_hook_context) 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 0000000000..b620bcfe1c --- /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 3c76bd8d86..3b4515c8a6 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.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 @@ def _execute_single_native_tool_call(self, tool_call: Any) -> dict[str, Any]: 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 @@ def _execute_single_native_tool_call(self, tool_call: Any) -> dict[str, Any]: 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,7 @@ def _execute_single_native_tool_call(self, tool_call: Any) -> dict[str, Any]: 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", - ) + result = run_after_tool_call_hooks(after_hook_context) 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 5b9fbf7bd6..16c6fe0fc0 100644 --- a/lib/crewai/src/crewai/hooks/__init__.py +++ b/lib/crewai/src/crewai/hooks/__init__.py @@ -6,6 +6,17 @@ 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 @@ def clear_all_global_hooks() -> dict[str, tuple[int, int]]: "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 0000000000..f93f8d15e6 --- /dev/null +++ b/lib/crewai/src/crewai/hooks/dispatch.py @@ -0,0 +1,418 @@ +"""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): + """Catalog of every interception point in the framework. + + The full catalog is frozen from day zero. Points without a live consumer + seam yet (``FILE_ACCESS``, ``ARTIFACT_OUTPUT``) can still be registered + against; dispatch for them is simply never triggered, which is the same + semantics as any point with no hooks. + """ + + # Execution-level boundaries + EXECUTION_START = "execution_start" + INPUT = "input" + OUTPUT = "output" + EXECUTION_END = "execution_end" + + # 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" + + # Step & agent points + PRE_STEP = "pre_step" + POST_STEP = "post_step" + TOOL_SELECTION = "tool_selection" + PRE_DELEGATION = "pre_delegation" + RETRY_ATTEMPT = "retry_attempt" + + # Subsystem points + MEMORY_WRITE = "memory_write" + MEMORY_READ = "memory_read" + KNOWLEDGE_RETRIEVAL = "knowledge_retrieval" + PRE_CODE_EXECUTION = "pre_code_execution" + MCP_CONNECT = "mcp_connect" + FILE_ACCESS = "file_access" + ARTIFACT_OUTPUT = "artifact_output" + + # Flow-specific points + FLOW_TRANSITION = "flow_transition" + ROUTER_DECISION = "router_decision" + + +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.""" + try: + _global_hooks[point].remove(hook) + 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 _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 name: + 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, # type: ignore[arg-type] + 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``.""" + if result is not None: + if 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) + if agent is not None and getattr(agent, "role", None) 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) + + 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 67108c01c8..55563e84e5 100644 --- a/lib/crewai/src/crewai/hooks/llm_hooks.py +++ b/lib/crewai/src/crewai/hooks/llm_hooks.py @@ -5,6 +5,11 @@ 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 @@ def request_human_input( 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 a860c01d45..12a7503df2 100644 --- a/lib/crewai/src/crewai/hooks/tool_hooks.py +++ b/lib/crewai/src/crewai/hooks/tool_hooks.py @@ -5,6 +5,12 @@ 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,72 @@ def request_human_input( 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 result is not None: + context.tool_result = result + return True + return 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=False, + ) + 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=False, + ) + 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 ece84fde08..03e80915b1 100644 --- a/lib/crewai/src/crewai/llms/base_llm.py +++ b/lib/crewai/src/crewai/llms/base_llm.py @@ -1007,13 +1007,14 @@ def _invoke_before_llm_call_hooks( from crewai_core.printer import PRINTER + from crewai.hooks.dispatch import HookAborted, InterceptionPoint, dispatch from crewai.hooks.llm_hooks import ( LLMCallHookContext, + before_llm_call_reducer, get_before_llm_call_hooks, ) - before_hooks = get_before_llm_call_hooks() - if not before_hooks: + if not get_before_llm_call_hooks(): return True hook_context = LLMCallHookContext( @@ -1024,24 +1025,19 @@ def _invoke_before_llm_call_hooks( 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,15 +1070,14 @@ def _invoke_after_llm_call_hooks( 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, + after_llm_call_reducer, get_after_llm_call_hooks, ) - after_hooks = get_after_llm_call_hooks() - if not after_hooks: + if not get_after_llm_call_hooks(): return response hook_context = LLMCallHookContext( @@ -1094,20 +1089,11 @@ def _invoke_after_llm_call_hooks( 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/utilities/agent_utils.py b/lib/crewai/src/crewai/utilities/agent_utils.py index 8a5fd9a071..1a286d08bb 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,7 @@ 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 + result = run_after_tool_call_hooks(after_hook_context) if not error_event_emitted: crewai_event_bus.emit( @@ -1691,27 +1677,31 @@ def _setup_before_llm_call_hooks( 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 + from crewai.hooks.dispatch import ( + HookAborted, + InterceptionPoint, + run_hooks, + ) + from crewai.hooks.llm_hooks import LLMCallHookContext, before_llm_call_reducer 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, + executor_context.before_llm_call_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: @@ -1749,7 +1739,8 @@ def _setup_after_llm_call_hooks( 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 + from crewai.hooks.dispatch import InterceptionPoint, 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). @@ -1768,18 +1759,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, + executor_context.after_llm_call_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 dcb25594cd..893922ab63 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.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,11 @@ 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}" + ) + return ToolResult(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,16 +120,7 @@ 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) @@ -181,7 +163,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 +203,11 @@ 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}" + ) + return ToolResult(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,16 +223,7 @@ 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) diff --git a/lib/crewai/tests/hooks/test_dispatch.py b/lib/crewai/tests/hooks/test_dispatch.py new file mode 100644 index 0000000000..da832a0315 --- /dev/null +++ b/lib/crewai/tests/hooks/test_dispatch.py @@ -0,0 +1,229 @@ +"""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, +) +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 + + +@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.INPUT, 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.INPUT, double) + ctx = _Ctx(payload="ab") + dispatch(InterceptionPoint.INPUT, ctx) + assert ctx.payload == "abab" + + def test_in_place_mutation_is_honored(self): + def mutate(ctx): + ctx.payload.append(1) + return None + + register(InterceptionPoint.INPUT, mutate) + ctx = _Ctx(payload=[]) + dispatch(InterceptionPoint.INPUT, ctx) + assert ctx.payload == [1] + + def test_hooks_run_in_registration_order(self): + order: list[int] = [] + register(InterceptionPoint.INPUT, lambda ctx: order.append(1)) + register(InterceptionPoint.INPUT, lambda ctx: order.append(2)) + dispatch(InterceptionPoint.INPUT, _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.INPUT, blocker) + with pytest.raises(HookAborted) as exc: + dispatch(InterceptionPoint.INPUT, _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.INPUT, boom) + register(InterceptionPoint.INPUT, after) + dispatch(InterceptionPoint.INPUT, _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.MEMORY_WRITE) + def hook(ctx): + return None + + assert hook in get_hooks(InterceptionPoint.MEMORY_WRITE) + + 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"] + + +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.OUTPUT, lambda ctx: order.append("global")) + + with scoped_hooks(): + register_scoped(InterceptionPoint.OUTPUT, lambda ctx: order.append("scoped")) + dispatch(InterceptionPoint.OUTPUT, _Ctx()) + + # outside the scope the scoped hook is gone + dispatch(InterceptionPoint.OUTPUT, _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.INPUT, _Ctx()) + + assert events == [] + + def test_event_reports_outcome(self): + events: list[HookDispatchedEvent] = [] + + register(InterceptionPoint.INPUT, lambda ctx: "changed") + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(HookDispatchedEvent) + def _capture(_source, event): + events.append(event) + + dispatch(InterceptionPoint.INPUT, _Ctx()) + + assert len(events) == 1 + assert events[0].interception_point == "input" + assert events[0].outcome == "modified" + assert events[0].hook_count == 1 + + +class TestNoOpOverhead: + """The no-op fast path must stay cheap (a single dict lookup).""" + + def test_noop_dispatch_overhead_budget(self): + ctx = _Ctx() + iterations = 100_000 + start = time.perf_counter() + for _ in range(iterations): + dispatch(InterceptionPoint.INPUT, ctx) + elapsed = time.perf_counter() - start + # Generous CI-safe budget: < 5µs per no-op dispatch on average. + assert elapsed / iterations < 5e-6 From e7213d8beff964928b8025449f287ef15fd0d544 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Sat, 11 Jul 2026 19:12:15 -0300 Subject: [PATCH 2/6] 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. --- lib/crewai/src/crewai/hooks/dispatch.py | 32 ++++++-- lib/crewai/src/crewai/hooks/tool_hooks.py | 13 +++- lib/crewai/src/crewai/utilities/tool_utils.py | 30 +++++++- lib/crewai/tests/hooks/test_dispatch.py | 75 ++++++++++++++++++- 4 files changed, 135 insertions(+), 15 deletions(-) diff --git a/lib/crewai/src/crewai/hooks/dispatch.py b/lib/crewai/src/crewai/hooks/dispatch.py index f93f8d15e6..adf6cda21d 100644 --- a/lib/crewai/src/crewai/hooks/dispatch.py +++ b/lib/crewai/src/crewai/hooks/dispatch.py @@ -129,9 +129,17 @@ def register(point: InterceptionPoint, hook: HookFn) -> None: def unregister(point: InterceptionPoint, hook: HookFn) -> bool: - """Unregister a specific global hook. Returns True if it was removed.""" + """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: - _global_hooks[point].remove(hook) + hooks.remove(target) return True except ValueError: return False @@ -235,10 +243,14 @@ def _emit_telemetry( def _default_reducer(ctx: Any, result: Any) -> bool: - """Default payload semantics: a non-None return replaces ``ctx.payload``.""" - if result is not None: - if hasattr(ctx, "payload"): - ctx.payload = result + """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 @@ -368,7 +380,10 @@ def filtered(ctx: Any) -> Any: return None if agents: agent = getattr(ctx, "agent", None) - if agent is not None and getattr(agent, "role", None) not in agents: + 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) @@ -412,6 +427,9 @@ def decorator(func: HookFn) -> HookFn: 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 diff --git a/lib/crewai/src/crewai/hooks/tool_hooks.py b/lib/crewai/src/crewai/hooks/tool_hooks.py index 12a7503df2..4f3977d54d 100644 --- a/lib/crewai/src/crewai/hooks/tool_hooks.py +++ b/lib/crewai/src/crewai/hooks/tool_hooks.py @@ -160,6 +160,15 @@ def after_tool_call_reducer(context: ToolCallHookContext, result: object) -> boo 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. @@ -173,7 +182,7 @@ def run_before_tool_call_hooks(context: ToolCallHookContext) -> bool: InterceptionPoint.PRE_TOOL_CALL, context, reducer=before_tool_call_reducer, - verbose=False, + verbose=_hook_verbose(context), ) return False except HookAborted: @@ -190,7 +199,7 @@ def run_after_tool_call_hooks(context: ToolCallHookContext) -> str | None: InterceptionPoint.POST_TOOL_CALL, context, reducer=after_tool_call_reducer, - verbose=False, + verbose=_hook_verbose(context), ) return context.tool_result diff --git a/lib/crewai/src/crewai/utilities/tool_utils.py b/lib/crewai/src/crewai/utilities/tool_utils.py index 893922ab63..debfdcb2a2 100644 --- a/lib/crewai/src/crewai/utilities/tool_utils.py +++ b/lib/crewai/src/crewai/utilities/tool_utils.py @@ -104,7 +104,20 @@ async def aexecute_tool_and_check_finality( blocked_message = ( f"Tool execution blocked by hook. Tool: {tool_calling.tool_name}" ) - return ToolResult(blocked_message, False) + # 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, False) tool_result = await tool_usage.ause(tool_calling, agent_action.text) raw_tool_result = tool_usage.get_last_raw_result(tool_result) @@ -207,7 +220,20 @@ def execute_tool_and_check_finality( blocked_message = ( f"Tool execution blocked by hook. Tool: {tool_calling.tool_name}" ) - return ToolResult(blocked_message, False) + # 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, False) tool_result = tool_usage.use(tool_calling, agent_action.text) raw_tool_result = tool_usage.get_last_raw_result(tool_result) diff --git a/lib/crewai/tests/hooks/test_dispatch.py b/lib/crewai/tests/hooks/test_dispatch.py index da832a0315..f4151f73ea 100644 --- a/lib/crewai/tests/hooks/test_dispatch.py +++ b/lib/crewai/tests/hooks/test_dispatch.py @@ -23,6 +23,7 @@ register, register_scoped, scoped_hooks, + unregister as unregister_hook, ) from crewai.hooks.llm_hooks import ( get_before_llm_call_hooks, @@ -36,6 +37,7 @@ class _Ctx: payload: object = None tool_name: str | None = None agent: object = None + agent_role: str | None = None @pytest.fixture(autouse=True) @@ -143,6 +145,27 @@ def hook(ctx): 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_STEP, 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_STEP, _Ctx(agent_role="Writer")) + dispatch(InterceptionPoint.PRE_STEP, _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.""" @@ -208,22 +231,66 @@ def _capture(_source, event): events.append(event) dispatch(InterceptionPoint.INPUT, _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 == "input" 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.INPUT, 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.INPUT, _Ctx()) + crewai_event_bus.flush() + + assert len(events) == 1 + assert events[0].interception_point == "input" + 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_budget(self): + 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.INPUT, 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.INPUT, ctx) - elapsed = time.perf_counter() - start - # Generous CI-safe budget: < 5µs per no-op dispatch on average. - assert elapsed / iterations < 5e-6 + noop = time.perf_counter() - start + + assert noop < baseline * 50 + 5e-3 From 3e7d718749748f17e418bfe2f8ac2ece21fd50d5 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Mon, 13 Jul 2026 11:21:18 -0300 Subject: [PATCH 3/6] fix: honor scoped hooks on direct llm calls and register @on crew methods MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../src/crewai/agents/crew_agent_executor.py | 4 +- .../src/crewai/experimental/agent_executor.py | 4 +- lib/crewai/src/crewai/hooks/dispatch.py | 4 +- lib/crewai/src/crewai/hooks/tool_hooks.py | 2 +- lib/crewai/src/crewai/llms/base_llm.py | 12 +-- lib/crewai/src/crewai/project/crew_base.py | 29 +++++++- .../src/crewai/utilities/agent_utils.py | 4 +- lib/crewai/src/crewai/utilities/tool_utils.py | 20 ++++- .../tests/hooks/test_crew_scoped_hooks.py | 56 ++++++++++++++ lib/crewai/tests/hooks/test_llm_hooks.py | 74 +++++++++++++++++++ 10 files changed, 190 insertions(+), 19 deletions(-) diff --git a/lib/crewai/src/crewai/agents/crew_agent_executor.py b/lib/crewai/src/crewai/agents/crew_agent_executor.py index 730cf65a12..040520e1fa 100644 --- a/lib/crewai/src/crewai/agents/crew_agent_executor.py +++ b/lib/crewai/src/crewai/agents/crew_agent_executor.py @@ -1020,7 +1020,9 @@ def _execute_single_native_tool_call( tool_result=result, raw_tool_result=raw_tool_result, ) - result = run_after_tool_call_hooks(after_hook_context) + 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/experimental/agent_executor.py b/lib/crewai/src/crewai/experimental/agent_executor.py index 3b4515c8a6..96308f9c9e 100644 --- a/lib/crewai/src/crewai/experimental/agent_executor.py +++ b/lib/crewai/src/crewai/experimental/agent_executor.py @@ -2047,7 +2047,9 @@ def _execute_single_native_tool_call(self, tool_call: Any) -> dict[str, Any]: tool_result=result, raw_tool_result=raw_tool_result, ) - result = run_after_tool_call_hooks(after_hook_context) + 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/dispatch.py b/lib/crewai/src/crewai/hooks/dispatch.py index adf6cda21d..686d8bdc24 100644 --- a/lib/crewai/src/crewai/hooks/dispatch.py +++ b/lib/crewai/src/crewai/hooks/dispatch.py @@ -209,7 +209,7 @@ def _source_name(source: Any) -> str | None: if isinstance(source, str): return source name = getattr(source, "__name__", None) - if name: + if isinstance(name, str): return name return type(source).__name__ @@ -231,7 +231,7 @@ def _emit_telemetry( _TELEMETRY_SOURCE, event=HookDispatchedEvent( interception_point=point.value, - outcome=outcome, # type: ignore[arg-type] + outcome=outcome, hook_count=hook_count, duration_ms=duration_ms, abort_reason=abort_reason, diff --git a/lib/crewai/src/crewai/hooks/tool_hooks.py b/lib/crewai/src/crewai/hooks/tool_hooks.py index 4f3977d54d..a4509bce7c 100644 --- a/lib/crewai/src/crewai/hooks/tool_hooks.py +++ b/lib/crewai/src/crewai/hooks/tool_hooks.py @@ -154,7 +154,7 @@ def after_tool_call_reducer(context: ToolCallHookContext, result: object) -> boo A non-None return replaces the tool result on the context. """ - if result is not None: + if isinstance(result, str): context.tool_result = result return True return False diff --git a/lib/crewai/src/crewai/llms/base_llm.py b/lib/crewai/src/crewai/llms/base_llm.py index 03e80915b1..a71126f585 100644 --- a/lib/crewai/src/crewai/llms/base_llm.py +++ b/lib/crewai/src/crewai/llms/base_llm.py @@ -1011,12 +1011,10 @@ def _invoke_before_llm_call_hooks( from crewai.hooks.llm_hooks import ( LLMCallHookContext, before_llm_call_reducer, - get_before_llm_call_hooks, ) - if not get_before_llm_call_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, @@ -1074,12 +1072,10 @@ def _invoke_after_llm_call_hooks( from crewai.hooks.llm_hooks import ( LLMCallHookContext, after_llm_call_reducer, - get_after_llm_call_hooks, ) - if not get_after_llm_call_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, diff --git a/lib/crewai/src/crewai/project/crew_base.py b/lib/crewai/src/crewai/project/crew_base.py index 2d35a35b30..fa945f0c56 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 filtered(context: ToolCallHookContext) -> str | 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 1a286d08bb..3beef80929 100644 --- a/lib/crewai/src/crewai/utilities/agent_utils.py +++ b/lib/crewai/src/crewai/utilities/agent_utils.py @@ -1580,7 +1580,9 @@ def execute_single_native_tool_call( tool_result=result, raw_tool_result=raw_tool_result, ) - result = run_after_tool_call_hooks(after_hook_context) + 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/utilities/tool_utils.py b/lib/crewai/src/crewai/utilities/tool_utils.py index debfdcb2a2..2d6e3c1421 100644 --- a/lib/crewai/src/crewai/utilities/tool_utils.py +++ b/lib/crewai/src/crewai/utilities/tool_utils.py @@ -117,7 +117,10 @@ async def aexecute_tool_and_check_finality( raw_tool_result=blocked_message, ) modified_result = run_after_tool_call_hooks(blocked_hook_context) - return ToolResult(modified_result, False) + 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) @@ -135,7 +138,10 @@ async def aexecute_tool_and_check_finality( 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, @@ -233,7 +239,10 @@ def execute_tool_and_check_finality( raw_tool_result=blocked_message, ) modified_result = run_after_tool_call_hooks(blocked_hook_context) - return ToolResult(modified_result, False) + 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) @@ -251,7 +260,10 @@ def execute_tool_and_check_finality( 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 cf567a3331..b0f8e1eb22 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 @@ def crew(self): 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_llm_hooks.py b/lib/crewai/tests/hooks/test_llm_hooks.py index 7b8add6974..e7edbddbdf 100644 --- a/lib/crewai/tests/hooks/test_llm_hooks.py +++ b/lib/crewai/tests/hooks/test_llm_hooks.py @@ -497,3 +497,77 @@ def after_hook(context: LLMCallHookContext) -> str | None: 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]" From 984ea4441149a5612297b9903bf5e4edb89d9093 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Mon, 13 Jul 2026 15:12:40 -0300 Subject: [PATCH 4/6] 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. --- lib/crewai/src/crewai/hooks/dispatch.py | 36 ++------------ lib/crewai/tests/hooks/test_dispatch.py | 62 ++++++++++++------------- 2 files changed, 36 insertions(+), 62 deletions(-) diff --git a/lib/crewai/src/crewai/hooks/dispatch.py b/lib/crewai/src/crewai/hooks/dispatch.py index 686d8bdc24..343dcc4284 100644 --- a/lib/crewai/src/crewai/hooks/dispatch.py +++ b/lib/crewai/src/crewai/hooks/dispatch.py @@ -38,46 +38,20 @@ class InterceptionPoint(str, Enum): - """Catalog of every interception point in the framework. + """Interception points wired by this layer. - The full catalog is frozen from day zero. Points without a live consumer - seam yet (``FILE_ACCESS``, ``ARTIFACT_OUTPUT``) can still be registered - against; dispatch for them is simply never triggered, which is the same - semantics as any point with no hooks. + 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. """ - # Execution-level boundaries - EXECUTION_START = "execution_start" - INPUT = "input" - OUTPUT = "output" - EXECUTION_END = "execution_end" - # 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" - # Step & agent points - PRE_STEP = "pre_step" - POST_STEP = "post_step" - TOOL_SELECTION = "tool_selection" - PRE_DELEGATION = "pre_delegation" - RETRY_ATTEMPT = "retry_attempt" - - # Subsystem points - MEMORY_WRITE = "memory_write" - MEMORY_READ = "memory_read" - KNOWLEDGE_RETRIEVAL = "knowledge_retrieval" - PRE_CODE_EXECUTION = "pre_code_execution" - MCP_CONNECT = "mcp_connect" - FILE_ACCESS = "file_access" - ARTIFACT_OUTPUT = "artifact_output" - - # Flow-specific points - FLOW_TRANSITION = "flow_transition" - ROUTER_DECISION = "router_decision" - class HookAborted(Exception): # noqa: N818 - public contract name from OSS-86 """Raised by a hook (or a legacy adapter) to abort the intercepted operation. diff --git a/lib/crewai/tests/hooks/test_dispatch.py b/lib/crewai/tests/hooks/test_dispatch.py index f4151f73ea..9aa5d59127 100644 --- a/lib/crewai/tests/hooks/test_dispatch.py +++ b/lib/crewai/tests/hooks/test_dispatch.py @@ -53,7 +53,7 @@ class TestDispatchContract: def test_noop_fast_path_returns_context_unchanged(self): ctx = _Ctx(payload="original") - result = dispatch(InterceptionPoint.INPUT, ctx) + result = dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx) assert result is ctx assert ctx.payload == "original" @@ -61,9 +61,9 @@ def test_return_value_replaces_payload(self): def double(ctx): return ctx.payload * 2 - register(InterceptionPoint.INPUT, double) + register(InterceptionPoint.PRE_MODEL_CALL, double) ctx = _Ctx(payload="ab") - dispatch(InterceptionPoint.INPUT, ctx) + dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx) assert ctx.payload == "abab" def test_in_place_mutation_is_honored(self): @@ -71,25 +71,25 @@ def mutate(ctx): ctx.payload.append(1) return None - register(InterceptionPoint.INPUT, mutate) + register(InterceptionPoint.PRE_MODEL_CALL, mutate) ctx = _Ctx(payload=[]) - dispatch(InterceptionPoint.INPUT, ctx) + dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx) assert ctx.payload == [1] def test_hooks_run_in_registration_order(self): order: list[int] = [] - register(InterceptionPoint.INPUT, lambda ctx: order.append(1)) - register(InterceptionPoint.INPUT, lambda ctx: order.append(2)) - dispatch(InterceptionPoint.INPUT, _Ctx()) + 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.INPUT, blocker) + register(InterceptionPoint.PRE_MODEL_CALL, blocker) with pytest.raises(HookAborted) as exc: - dispatch(InterceptionPoint.INPUT, _Ctx()) + dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx()) assert exc.value.reason == "nope" assert exc.value.source == "policy" @@ -103,9 +103,9 @@ def boom(ctx): def after(ctx): ran.append("after") - register(InterceptionPoint.INPUT, boom) - register(InterceptionPoint.INPUT, after) - dispatch(InterceptionPoint.INPUT, _Ctx(), verbose=False) + register(InterceptionPoint.PRE_MODEL_CALL, boom) + register(InterceptionPoint.PRE_MODEL_CALL, after) + dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx(), verbose=False) assert ran == ["boom", "after"] @@ -113,11 +113,11 @@ class TestOnDecorator: """The @on decorator registers and filters like the legacy decorators.""" def test_on_registers_global_hook(self): - @on(InterceptionPoint.MEMORY_WRITE) + @on(InterceptionPoint.POST_TOOL_CALL) def hook(ctx): return None - assert hook in get_hooks(InterceptionPoint.MEMORY_WRITE) + assert hook in get_hooks(InterceptionPoint.POST_TOOL_CALL) def test_tool_filter_skips_non_matching_tools(self): seen: list[str] = [] @@ -148,13 +148,13 @@ def hook(ctx): def test_agent_filter_falls_back_to_agent_role(self): seen: list[str] = [] - @on(InterceptionPoint.PRE_STEP, agents=["Researcher"]) + @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_STEP, _Ctx(agent_role="Writer")) - dispatch(InterceptionPoint.PRE_STEP, _Ctx(agent_role="Researcher")) + 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): @@ -192,14 +192,14 @@ class TestScopedHooks: def test_scoped_runs_after_global_then_cleared(self): order: list[str] = [] - register(InterceptionPoint.OUTPUT, lambda ctx: order.append("global")) + register(InterceptionPoint.POST_MODEL_CALL, lambda ctx: order.append("global")) with scoped_hooks(): - register_scoped(InterceptionPoint.OUTPUT, lambda ctx: order.append("scoped")) - dispatch(InterceptionPoint.OUTPUT, _Ctx()) + 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.OUTPUT, _Ctx()) + dispatch(InterceptionPoint.POST_MODEL_CALL, _Ctx()) assert order == ["global", "scoped", "global"] @@ -215,14 +215,14 @@ def test_no_event_on_empty_fast_path(self): def _capture(_source, event): events.append(event) - dispatch(InterceptionPoint.INPUT, _Ctx()) + dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx()) assert events == [] def test_event_reports_outcome(self): events: list[HookDispatchedEvent] = [] - register(InterceptionPoint.INPUT, lambda ctx: "changed") + register(InterceptionPoint.PRE_MODEL_CALL, lambda ctx: "changed") with crewai_event_bus.scoped_handlers(): @@ -230,13 +230,13 @@ def test_event_reports_outcome(self): def _capture(_source, event): events.append(event) - dispatch(InterceptionPoint.INPUT, _Ctx()) + 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 == "input" + assert events[0].interception_point == "pre_model_call" assert events[0].outcome == "modified" assert events[0].hook_count == 1 @@ -246,7 +246,7 @@ def test_event_reports_abort_outcome(self): def blocker(ctx): raise HookAborted(reason="blocked", source="policy") - register(InterceptionPoint.INPUT, blocker) + register(InterceptionPoint.PRE_MODEL_CALL, blocker) with crewai_event_bus.scoped_handlers(): @@ -255,11 +255,11 @@ def _capture(_source, event): events.append(event) with pytest.raises(HookAborted): - dispatch(InterceptionPoint.INPUT, _Ctx()) + dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx()) crewai_event_bus.flush() assert len(events) == 1 - assert events[0].interception_point == "input" + 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" @@ -280,7 +280,7 @@ def _baseline(_c): return _c for _ in range(1000): # warm up both paths - dispatch(InterceptionPoint.INPUT, ctx) + dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx) _baseline(ctx) start = time.perf_counter() @@ -290,7 +290,7 @@ def _baseline(_c): start = time.perf_counter() for _ in range(iterations): - dispatch(InterceptionPoint.INPUT, ctx) + dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx) noop = time.perf_counter() - start assert noop < baseline * 50 + 5e-3 From 1a4729dfcb92372f2baa1205a08fcc19377f69d7 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Mon, 13 Jul 2026 15:38:16 -0300 Subject: [PATCH 5/6] 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. --- lib/crewai/tests/hooks/test_llm_hooks.py | 54 ++++++++++++++++++ lib/crewai/tests/hooks/test_tool_hooks.py | 69 +++++++++++++++++++++++ 2 files changed, 123 insertions(+) diff --git a/lib/crewai/tests/hooks/test_llm_hooks.py b/lib/crewai/tests/hooks/test_llm_hooks.py index e7edbddbdf..cabe552140 100644 --- a/lib/crewai/tests/hooks/test_llm_hooks.py +++ b/lib/crewai/tests/hooks/test_llm_hooks.py @@ -337,6 +337,60 @@ def test_hook(context): 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_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.""" diff --git a/lib/crewai/tests/hooks/test_tool_hooks.py b/lib/crewai/tests/hooks/test_tool_hooks.py index 14c6848a8f..4978f14b8a 100644 --- a/lib/crewai/tests/hooks/test_tool_hooks.py +++ b/lib/crewai/tests/hooks/test_tool_hooks.py @@ -576,6 +576,75 @@ def after_tool_call_hook(context: ToolCallHookContext) -> str | None: 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).""" From 8a243309b4d9f70fdb9f02ed13850d14adebdac1 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Mon, 13 Jul 2026 16:18:33 -0300 Subject: [PATCH 6/6] 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. --- lib/crewai/src/crewai/hooks/dispatch.py | 14 ++++++ .../src/crewai/utilities/agent_utils.py | 29 +++++++++--- lib/crewai/tests/hooks/test_llm_hooks.py | 45 +++++++++++++++++++ 3 files changed, 83 insertions(+), 5 deletions(-) diff --git a/lib/crewai/src/crewai/hooks/dispatch.py b/lib/crewai/src/crewai/hooks/dispatch.py index 343dcc4284..2822c73dd8 100644 --- a/lib/crewai/src/crewai/hooks/dispatch.py +++ b/lib/crewai/src/crewai/hooks/dispatch.py @@ -165,6 +165,20 @@ def register_scoped(point: InterceptionPoint, hook: HookFn) -> None: 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] diff --git a/lib/crewai/src/crewai/utilities/agent_utils.py b/lib/crewai/src/crewai/utilities/agent_utils.py index 3beef80929..e3fa001dc2 100644 --- a/lib/crewai/src/crewai/utilities/agent_utils.py +++ b/lib/crewai/src/crewai/utilities/agent_utils.py @@ -1678,14 +1678,24 @@ 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: + 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) @@ -1693,7 +1703,7 @@ def _setup_before_llm_call_hooks( run_hooks( InterceptionPoint.PRE_MODEL_CALL, hook_context, - executor_context.before_llm_call_hooks, + hooks, reducer=before_llm_call_reducer, verbose=verbose, ) @@ -1740,8 +1750,8 @@ 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.dispatch import InterceptionPoint, run_hooks + 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 @@ -1750,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): @@ -1764,7 +1783,7 @@ def _setup_after_llm_call_hooks( run_hooks( InterceptionPoint.POST_MODEL_CALL, hook_context, - executor_context.after_llm_call_hooks, + hooks, reducer=after_llm_call_reducer, verbose=verbose, ) diff --git a/lib/crewai/tests/hooks/test_llm_hooks.py b/lib/crewai/tests/hooks/test_llm_hooks.py index cabe552140..0c917712d4 100644 --- a/lib/crewai/tests/hooks/test_llm_hooks.py +++ b/lib/crewai/tests/hooks/test_llm_hooks.py @@ -367,6 +367,51 @@ def later_hook(context): 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