From a85e100bec8fd8d4f6aac10f010421d89d006635 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Sat, 11 Jul 2026 14:14:39 -0300 Subject: [PATCH] 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 de2315e3a..730cf65a1 100644 --- a/lib/crewai/src/crewai/agents/crew_agent_executor.py +++ b/lib/crewai/src/crewai/agents/crew_agent_executor.py @@ -46,8 +46,8 @@ from crewai.hooks.llm_hooks import ( ) from crewai.hooks.tool_hooks import ( ToolCallHookContext, - get_after_tool_call_hooks, - get_before_tool_call_hooks, + run_after_tool_call_hooks, + run_before_tool_call_hooks, ) from crewai.types.callback import SerializableCallable from crewai.utilities.agent_utils import ( @@ -951,7 +951,6 @@ class CrewAgentExecutor(BaseAgentExecutor): track_delegation_if_needed(func_name, args_dict or {}, self.task) - hook_blocked = False before_hook_context = ToolCallHookContext( tool_name=func_name, tool_input=args_dict or {}, @@ -960,19 +959,7 @@ class CrewAgentExecutor(BaseAgentExecutor): task=self.task, crew=self.crew, ) - before_hooks = get_before_tool_call_hooks() - try: - for hook in before_hooks: - hook_result = hook(before_hook_context) - if hook_result is False: - hook_blocked = True - break - except Exception as hook_error: - if self.agent.verbose: - PRINTER.print( - content=f"Error in before_tool_call hook: {hook_error}", - color="red", - ) + hook_blocked = run_before_tool_call_hooks(before_hook_context) if hook_blocked: result = f"Tool execution blocked by hook. Tool: {func_name}" @@ -1033,19 +1020,7 @@ class CrewAgentExecutor(BaseAgentExecutor): tool_result=result, raw_tool_result=raw_tool_result, ) - after_hooks = get_after_tool_call_hooks() - try: - for after_hook in after_hooks: - after_hook_result = after_hook(after_hook_context) - if after_hook_result is not None: - result = after_hook_result - after_hook_context.tool_result = result - except Exception as hook_error: - if self.agent.verbose: - PRINTER.print( - content=f"Error in after_tool_call hook: {hook_error}", - color="red", - ) + 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 000000000..b620bcfe1 --- /dev/null +++ b/lib/crewai/src/crewai/events/types/hook_events.py @@ -0,0 +1,19 @@ +from typing import Literal + +from crewai.events.base_events import BaseEvent + + +class HookDispatchedEvent(BaseEvent): + """Event emitted whenever an interception point dispatches to hooks. + + Only emitted when at least one hook is registered for the point, so the + no-op fast path stays free of event overhead. + """ + + type: Literal["hook_dispatched"] = "hook_dispatched" + interception_point: str + outcome: Literal["proceeded", "modified", "aborted"] + hook_count: int + duration_ms: float + abort_reason: str | None = None + abort_source: str | None = None diff --git a/lib/crewai/src/crewai/experimental/agent_executor.py b/lib/crewai/src/crewai/experimental/agent_executor.py index 3c76bd8d8..3b4515c8a 100644 --- a/lib/crewai/src/crewai/experimental/agent_executor.py +++ b/lib/crewai/src/crewai/experimental/agent_executor.py @@ -62,8 +62,8 @@ from crewai.hooks.llm_hooks import ( ) from crewai.hooks.tool_hooks import ( ToolCallHookContext, - get_after_tool_call_hooks, - get_before_tool_call_hooks, + run_after_tool_call_hooks, + run_before_tool_call_hooks, ) from crewai.hooks.types import ( AfterLLMCallHookCallable, @@ -1975,7 +1975,6 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): track_delegation_if_needed(func_name, args_dict, self.task) - hook_blocked = False before_hook_context = ToolCallHookContext( tool_name=func_name, tool_input=args_dict, @@ -1984,19 +1983,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): task=self.task, crew=self.crew, ) - before_hooks = get_before_tool_call_hooks() - try: - for hook in before_hooks: - hook_result = hook(before_hook_context) - if hook_result is False: - hook_blocked = True - break - except Exception as hook_error: - if self.agent.verbose: - PRINTER.print( - content=f"Error in before_tool_call hook: {hook_error}", - color="red", - ) + hook_blocked = run_before_tool_call_hooks(before_hook_context) if hook_blocked: result = f"Tool execution blocked by hook. Tool: {func_name}" @@ -2060,19 +2047,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): tool_result=result, raw_tool_result=raw_tool_result, ) - after_hooks = get_after_tool_call_hooks() - try: - for after_hook in after_hooks: - after_hook_result = after_hook(after_hook_context) - if after_hook_result is not None: - result = after_hook_result - after_hook_context.tool_result = result - except Exception as hook_error: - if self.agent.verbose: - PRINTER.print( - content=f"Error in after_tool_call hook: {hook_error}", - color="red", - ) + 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 5b9fbf7bd..16c6fe0fc 100644 --- a/lib/crewai/src/crewai/hooks/__init__.py +++ b/lib/crewai/src/crewai/hooks/__init__.py @@ -6,6 +6,17 @@ from crewai.hooks.decorators import ( before_llm_call, before_tool_call, ) +from crewai.hooks.dispatch import ( + HookAborted, + InterceptionPoint, + clear as clear_hooks, + clear_all as clear_all_hooks, + dispatch, + get_hooks, + on, + register as register_hook, + unregister as unregister_hook, +) from crewai.hooks.llm_hooks import ( LLMCallHookContext, clear_after_llm_call_hooks, @@ -74,6 +85,8 @@ def clear_all_global_hooks() -> dict[str, tuple[int, int]]: __all__ = [ + "HookAborted", + "InterceptionPoint", "LLMCallHookContext", "ToolCallHookContext", "after_llm_call", @@ -83,20 +96,27 @@ __all__ = [ "clear_after_llm_call_hooks", "clear_after_tool_call_hooks", "clear_all_global_hooks", + "clear_all_hooks", "clear_all_llm_call_hooks", "clear_all_tool_call_hooks", "clear_before_llm_call_hooks", "clear_before_tool_call_hooks", + "clear_hooks", + "dispatch", "get_after_llm_call_hooks", "get_after_tool_call_hooks", "get_before_llm_call_hooks", "get_before_tool_call_hooks", + "get_hooks", + "on", "register_after_llm_call_hook", "register_after_tool_call_hook", "register_before_llm_call_hook", "register_before_tool_call_hook", + "register_hook", "unregister_after_llm_call_hook", "unregister_after_tool_call_hook", "unregister_before_llm_call_hook", "unregister_before_tool_call_hook", + "unregister_hook", ] diff --git a/lib/crewai/src/crewai/hooks/dispatch.py b/lib/crewai/src/crewai/hooks/dispatch.py new file mode 100644 index 000000000..f93f8d15e --- /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 67108c01c..55563e84e 100644 --- a/lib/crewai/src/crewai/hooks/llm_hooks.py +++ b/lib/crewai/src/crewai/hooks/llm_hooks.py @@ -5,6 +5,11 @@ from typing import TYPE_CHECKING, Any, cast from crewai_core.printer import PRINTER from crewai.events.event_listener import event_listener +from crewai.hooks.dispatch import ( + HookAborted, + InterceptionPoint, + get_global_hook_list, +) from crewai.hooks.types import ( AfterLLMCallHookCallable, AfterLLMCallHookType, @@ -150,8 +155,37 @@ class LLMCallHookContext: event_listener.formatter.resume_live_updates() -_before_llm_call_hooks: list[BeforeLLMCallHookType | BeforeLLMCallHookCallable] = [] -_after_llm_call_hooks: list[AfterLLMCallHookType | AfterLLMCallHookCallable] = [] +# The legacy registries are aliased to the generic dispatcher's global hook +# lists for the model-call points, so legacy registrations and new-dialect +# ``@on(InterceptionPoint.PRE_MODEL_CALL)`` hooks share one ordered queue. +_before_llm_call_hooks: list[BeforeLLMCallHookType | BeforeLLMCallHookCallable] = ( + get_global_hook_list(InterceptionPoint.PRE_MODEL_CALL) +) +_after_llm_call_hooks: list[AfterLLMCallHookType | AfterLLMCallHookCallable] = ( + get_global_hook_list(InterceptionPoint.POST_MODEL_CALL) +) + + +def before_llm_call_reducer(context: LLMCallHookContext, result: object) -> bool: + """Legacy calling convention for ``pre_model_call`` hooks. + + A ``False`` return aborts the call (mapped to :class:`HookAborted`); messages + are modified in place, so no payload replacement occurs here. + """ + if result is False: + raise HookAborted(reason="before_llm_call hook returned False") + return False + + +def after_llm_call_reducer(context: LLMCallHookContext, result: object) -> bool: + """Legacy calling convention for ``post_model_call`` hooks. + + A non-empty string return replaces the response on the context. + """ + if result is not None and isinstance(result, str): + context.response = result + return True + return False def register_before_llm_call_hook( diff --git a/lib/crewai/src/crewai/hooks/tool_hooks.py b/lib/crewai/src/crewai/hooks/tool_hooks.py index a860c01d4..12a7503df 100644 --- a/lib/crewai/src/crewai/hooks/tool_hooks.py +++ b/lib/crewai/src/crewai/hooks/tool_hooks.py @@ -5,6 +5,12 @@ from typing import TYPE_CHECKING, Any from crewai_core.printer import PRINTER from crewai.events.event_listener import event_listener +from crewai.hooks.dispatch import ( + HookAborted, + InterceptionPoint, + dispatch, + get_global_hook_list, +) from crewai.hooks.types import ( AfterToolCallHookCallable, AfterToolCallHookType, @@ -121,8 +127,72 @@ class ToolCallHookContext: event_listener.formatter.resume_live_updates() -_before_tool_call_hooks: list[BeforeToolCallHookType | BeforeToolCallHookCallable] = [] -_after_tool_call_hooks: list[AfterToolCallHookType | AfterToolCallHookCallable] = [] +# The legacy registries are aliased to the generic dispatcher's global hook +# lists for the tool-call points, so legacy registrations and new-dialect +# ``@on(InterceptionPoint.PRE_TOOL_CALL)`` hooks share one ordered queue. +_before_tool_call_hooks: list[BeforeToolCallHookType | BeforeToolCallHookCallable] = ( + get_global_hook_list(InterceptionPoint.PRE_TOOL_CALL) +) +_after_tool_call_hooks: list[AfterToolCallHookType | AfterToolCallHookCallable] = ( + get_global_hook_list(InterceptionPoint.POST_TOOL_CALL) +) + + +def before_tool_call_reducer(context: ToolCallHookContext, result: object) -> bool: + """Legacy calling convention for ``pre_tool_call`` hooks. + + A ``False`` return blocks the call (mapped to :class:`HookAborted`); tool + input is modified in place, so no payload replacement occurs here. + """ + if result is False: + raise HookAborted(reason="before_tool_call hook returned False") + return False + + +def after_tool_call_reducer(context: ToolCallHookContext, result: object) -> bool: + """Legacy calling convention for ``post_tool_call`` hooks. + + A non-None return replaces the tool result on the context. + """ + if 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 ece84fde0..03e80915b 100644 --- a/lib/crewai/src/crewai/llms/base_llm.py +++ b/lib/crewai/src/crewai/llms/base_llm.py @@ -1007,13 +1007,14 @@ class BaseLLM(BaseModel, ABC): from crewai_core.printer import PRINTER + from crewai.hooks.dispatch import HookAborted, InterceptionPoint, dispatch from crewai.hooks.llm_hooks import ( LLMCallHookContext, + 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 @@ class BaseLLM(BaseModel, ABC): task=None, crew=None, ) - verbose = getattr(from_agent, "verbose", True) if from_agent else True try: - for hook in before_hooks: - result = hook(hook_context) - if result is False: - if verbose: - PRINTER.print( - content="LLM call blocked by before_llm_call hook", - color="yellow", - ) - return False - except Exception as e: - if verbose: - PRINTER.print( - content=f"Error in before_llm_call hook: {e}", - color="yellow", - ) + dispatch( + InterceptionPoint.PRE_MODEL_CALL, + hook_context, + reducer=before_llm_call_reducer, + ) + except HookAborted: + PRINTER.print( + content="LLM call blocked by before_llm_call hook", + color="yellow", + ) + return False return True @@ -1074,15 +1070,14 @@ class BaseLLM(BaseModel, ABC): if from_agent is not None or not isinstance(response, str): return response - from crewai_core.printer import PRINTER - + from crewai.hooks.dispatch import InterceptionPoint, dispatch from crewai.hooks.llm_hooks import ( LLMCallHookContext, + 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 @@ class BaseLLM(BaseModel, ABC): crew=None, response=response, ) - verbose = getattr(from_agent, "verbose", True) if from_agent else True - modified_response = response - try: - for hook in after_hooks: - result = hook(hook_context) - if result is not None and isinstance(result, str): - modified_response = result - hook_context.response = modified_response - except Exception as e: - if verbose: - PRINTER.print( - content=f"Error in after_llm_call hook: {e}", - color="yellow", - ) + dispatch( + InterceptionPoint.POST_MODEL_CALL, + hook_context, + reducer=after_llm_call_reducer, + ) - return modified_response + return hook_context.response if hook_context.response is not None else response diff --git a/lib/crewai/src/crewai/utilities/agent_utils.py b/lib/crewai/src/crewai/utilities/agent_utils.py index d2090b0a0..cf483d414 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 original_messages = executor_context.messages @@ -1762,18 +1753,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 dcb25594c..893922ab6 100644 --- a/lib/crewai/src/crewai/utilities/tool_utils.py +++ b/lib/crewai/src/crewai/utilities/tool_utils.py @@ -6,15 +6,14 @@ from crewai.agents.parser import AgentAction from crewai.agents.tools_handler import ToolsHandler from crewai.hooks.tool_hooks import ( ToolCallHookContext, - get_after_tool_call_hooks, - get_before_tool_call_hooks, + run_after_tool_call_hooks, + run_before_tool_call_hooks, ) from crewai.security.fingerprint import Fingerprint from crewai.tools.structured_tool import CrewStructuredTool from crewai.tools.tool_types import ToolResult from crewai.tools.tool_usage import ToolUsage, ToolUsageError from crewai.utilities.i18n import I18N_DEFAULT -from crewai.utilities.logger import Logger from crewai.utilities.string_utils import sanitize_tool_name @@ -57,11 +56,10 @@ async def aexecute_tool_and_check_finality( fingerprint_context: Optional context for fingerprinting. crew: Optional crew instance for hook context. - Returns: + Returns: ToolResult containing the execution result and whether it should be treated as a final answer. """ - logger = Logger(verbose=crew.verbose if crew else False) tool_name_to_tool_map = {sanitize_tool_name(tool.name): tool for tool in tools} if agent_key and agent_role and agent: @@ -102,18 +100,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 000000000..da832a031 --- /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