diff --git a/lib/crewai/src/crewai/agents/crew_agent_executor.py b/lib/crewai/src/crewai/agents/crew_agent_executor.py index de2315e3a9..040520e1fa 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,9 @@ 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", - ) + 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 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..96308f9c9e 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,9 @@ 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", - ) + 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 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..2822c73dd8 --- /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 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..a4509bce7c 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,81 @@ 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 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 ece84fde08..a71126f585 100644 --- a/lib/crewai/src/crewai/llms/base_llm.py +++ b/lib/crewai/src/crewai/llms/base_llm.py @@ -1007,15 +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, - 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 @@ 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,17 +1068,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, - 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 @@ 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/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 8a5fd9a071..e3fa001dc2 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 dcb25594cd..2d6e3c1421 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,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}") - - return ToolResult(modified_result, tool.result_as_answer) + modified_result = run_after_tool_call_hooks(after_hook_context) + + 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_dispatch.py b/lib/crewai/tests/hooks/test_dispatch.py new file mode 100644 index 0000000000..9aa5d59127 --- /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 7b8add6974..0c917712d4 100644 --- a/lib/crewai/tests/hooks/test_llm_hooks.py +++ b/lib/crewai/tests/hooks/test_llm_hooks.py @@ -337,6 +337,105 @@ 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_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 @@ 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]" 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)."""