fix: dedupe tool_selection and honor payload on catalog seams

Addresses review findings on the remaining interception points. `TOOL_SELECTION`
is no longer dispatched in `Crew._prepare_tools`; the single dispatch in
`Agent.create_agent_executor` covers both crew and standalone runs, so hooks no
longer fire twice (and additive edits no longer duplicate tools). `PRE_DELEGATION`
now dispatches outside the delegation `try/except`, so a `HookAborted` propagates
instead of being swallowed into a tool-error string. The `PRE_STEP` (flow),
`PRE_CODE_EXECUTION`, and `MCP_CONNECT` seams now apply the returned payload:
flow step params are rebound onto the call, script code is recompiled from the
edited source, and the MCP transport is replaced before connecting.

docs/docs.json

@@ -375,7 +375,8 @@
                       "edge/en/learn/using-annotations",
                       "edge/en/learn/execution-hooks",
                       "edge/en/learn/llm-hooks",
-                      "edge/en/learn/tool-hooks"
+                      "edge/en/learn/tool-hooks",
+                      "edge/en/learn/interception-hooks"
                     ]
                   },
                   {

docs/edge/en/learn/execution-hooks.mdx

@@ -42,6 +42,14 @@ Control and monitor tool execution:
 
 [View Tool Hooks Documentation →](/learn/tool-hooks)
 
+<Note>
+LLM and tool hooks are two points in a larger catalog. See
+[Interception Hooks](/learn/interception-hooks) for every framework-native
+interception point (execution boundaries, steps, memory, knowledge, flow
+transitions, and more) and the shared payload-in/payload-out contract they all
+follow.
+</Note>
+
 ## Hook Registration Methods
 
 ### 1. Decorator-Based Hooks (Recommended)

docs/edge/en/learn/interception-hooks.mdx

@@ -0,0 +1,168 @@
+---
+title: Interception Hooks
+description: The full catalog of framework-native interception points and the payload-in/payload-out contract every hook follows
+mode: "wide"
+---
+
+Interception hooks give you a single, uniform way to observe and modify CrewAI's
+runtime at well-defined points — from the moment an execution starts, through
+every model call, tool call, memory read, and flow transition, down to the final
+output. All points share one contract and one registration API.
+
+The four LLM/tool hooks documented in [LLM Hooks](/learn/llm-hooks) and
+[Tool Hooks](/learn/tool-hooks) are the same mechanism. Their existing
+decorators (`@before_llm_call`, `@before_tool_call`, ...) and `return False`
+semantics keep working unchanged; interception hooks generalize the same engine
+to the rest of the framework.
+
+## The contract
+
+Every hook is a **synchronous** callable that receives a single typed context:
+
+```python
+from crewai.hooks import on, HookAborted, InterceptionPoint
+
+@on(InterceptionPoint.INPUT)
+def add_defaults(ctx):
+    # 1. Observe: read anything off the context.
+    # 2. Mutate in place: change ctx.payload or nested fields directly.
+    ctx.payload.setdefault("locale", "en-US")
+    # 3. Or replace: return a new value to swap ctx.payload.
+    # 4. Or abort: raise HookAborted(reason, source) to stop the operation.
+    return None
+```
+
+A hook may do any of four things:
+
+| Action | How | Effect |
+|--------|-----|--------|
+| **Proceed** | `return None` (or nothing) | Operation continues unchanged |
+| **Mutate** | Change `ctx.payload` / fields in place | Change is visible downstream |
+| **Replace** | `return new_payload` | A non-`None` return replaces `ctx.payload` |
+| **Abort** | `raise HookAborted(reason, source)` | Operation is stopped; the reason propagates |
+
+### Composition, ordering, and fail-open
+
+- Multiple hooks on the same point run in **registration order**, global hooks
+  first, then execution-scoped hooks.
+- The (possibly mutated) payload flows from one hook to the next.
+- `HookAborted` **propagates by design** and stops the chain.
+- Any *other* exception raised by a hook is **swallowed** (fail-open) so a single
+  buggy hook can't crash a run — the same protection the legacy hooks provide.
+- When no hook is registered for a point, dispatch is a single dict lookup
+  (no-op fast path), so unused points cost effectively nothing.
+
+## Registering hooks
+
+Use the `@on` decorator for global hooks. It mirrors the legacy decorators'
+ergonomics, including `agents=` / `tools=` filters:
+
+```python
+from crewai.hooks import on, InterceptionPoint, HookAborted
+
+@on(InterceptionPoint.PRE_TOOL_CALL, tools=["delete_file"])
+def guard_deletes(ctx):
+    raise HookAborted(reason="file deletion is not allowed", source="policy")
+```
+
+Applied to a method inside a `@CrewBase` class, `@on` registers a crew-scoped
+hook (active only while that crew runs), matching the existing crew-scoped hook
+behavior.
+
+## Interception point catalog
+
+`payload` is the value a hook may mutate or replace at each point.
+
+### Execution boundaries
+
+| Point | When | `payload` |
+|-------|------|-----------|
+| `EXECUTION_START` | A crew or flow is about to begin | inputs `dict` |
+| `INPUT` | Resolved inputs for the execution | inputs `dict` |
+| `OUTPUT` | Final result is ready | the output object |
+| `EXECUTION_END` | A crew or flow has finished | the output object |
+
+### Model & tool boundaries (legacy-compatible)
+
+| Point | When | `payload` |
+|-------|------|-----------|
+| `PRE_MODEL_CALL` | Before an LLM call | `LLMCallHookContext` |
+| `POST_MODEL_CALL` | After an LLM call | response |
+| `PRE_TOOL_CALL` | Before a tool runs | `ToolCallHookContext` |
+| `POST_TOOL_CALL` | After a tool runs | tool result |
+
+### Step & agent points
+
+| Point | When | `payload` |
+|-------|------|-----------|
+| `PRE_STEP` | Before a task or flow-method step | step input |
+| `POST_STEP` | After a task or flow-method step | step output |
+| `TOOL_SELECTION` | Tools are offered to an agent | list of tools |
+| `PRE_DELEGATION` | An agent is about to delegate | delegation input |
+| `RETRY_ATTEMPT` | An operation is about to be retried | retry input |
+
+`PRE_STEP` / `POST_STEP` carry `ctx.kind` (`"task"` or `"flow_method"`) and
+`ctx.step_name`.
+
+### Subsystem points
+
+| Point | When | `payload` |
+|-------|------|-----------|
+| `MEMORY_WRITE` | A value is about to be stored in memory | value |
+| `MEMORY_READ` | A memory query is issued | query |
+| `KNOWLEDGE_RETRIEVAL` | A knowledge query is issued | query |
+| `PRE_CODE_EXECUTION` | Code is about to run (flow `ScriptAction`) | code string |
+| `MCP_CONNECT` | An MCP client is about to connect | connection params |
+| `FILE_ACCESS` | Reserved — no live seam yet | path |
+| `ARTIFACT_OUTPUT` | Reserved — no live seam yet | artifact |
+
+`FILE_ACCESS` and `ARTIFACT_OUTPUT` are part of the frozen catalog but have no
+consumer seam yet: registering against them is accepted and simply never fires,
+the same as any point with no hooks.
+
+### Flow-specific points
+
+| Point | When | `payload` |
+|-------|------|-----------|
+| `FLOW_TRANSITION` | A flow moves to its triggered methods | list of target methods |
+| `ROUTER_DECISION` | A flow router picks a route | route label |
+
+## Aborting an operation
+
+`HookAborted` carries a `reason` and an optional `source`. The `source` defaults
+to the aborting hook when omitted, which is useful for telemetry and failure
+messages:
+
+```python
+@on(InterceptionPoint.EXECUTION_START)
+def enforce_policy(ctx):
+    if not ctx.payload.get("authorized"):
+        raise HookAborted(reason="unauthorized execution", source="access-control")
+```
+
+## Telemetry
+
+Whenever a point actually dispatches to at least one hook, CrewAI emits a
+`HookDispatchedEvent` on the event bus with the point, the outcome
+(`proceeded` / `modified` / `aborted`), the hook count, the duration, and — for
+aborts — the reason and source. The no-op fast path emits nothing.
+
+## Managing hooks in tests
+
+```python
+import pytest
+from crewai.hooks import clear_all_hooks
+
+@pytest.fixture(autouse=True)
+def reset_hooks():
+    clear_all_hooks()
+    yield
+    clear_all_hooks()
+```
+
+## Related documentation
+
+- [Execution Hooks Overview →](/learn/execution-hooks)
+- [LLM Call Hooks →](/learn/llm-hooks)
+- [Tool Call Hooks →](/learn/tool-hooks)
+- [Before and After Kickoff Hooks →](/learn/before-and-after-kickoff-hooks)

lib/crewai/src/crewai/agent/core.py

@@ -656,6 +656,22 @@ class Agent(BaseAgent):
 
         return result
 
+    def _dispatch_retry_attempt(self, e: Exception, task: Task) -> None:
+        """Fire the ``retry_attempt`` interception point before re-executing a task."""
+        from crewai.hooks.contexts import RetryAttemptContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        retry_ctx = RetryAttemptContext(
+            agent=self,
+            agent_role=getattr(self, "role", None),
+            task=task,
+            attempt=self._times_executed,
+            max_attempts=self.max_retry_limit,
+            error=e,
+            payload=e,
+        )
+        dispatch(InterceptionPoint.RETRY_ATTEMPT, retry_ctx)
+
     def _check_execution_error(self, e: Exception, task: Task) -> None:
         """Check if an execution error should be re-raised immediately.
 
@@ -709,6 +725,7 @@ class Agent(BaseAgent):
             Result from retried execution.
         """
         self._check_execution_error(e, task)
+        self._dispatch_retry_attempt(e, task)
         return self.execute_task(task, context, tools)
 
     async def _handle_execution_error_async(
@@ -730,6 +747,7 @@ class Agent(BaseAgent):
             Result from retried execution.
         """
         self._check_execution_error(e, task)
+        self._dispatch_retry_attempt(e, task)
         return await self.aexecute_task(task, context, tools)
 
     def message(self, content: str, **kwargs: Any) -> str:
@@ -1054,6 +1072,21 @@ class Agent(BaseAgent):
             An instance of the CrewAgentExecutor class.
         """
         raw_tools: list[BaseTool] = tools or self.tools or []
+
+        from crewai.hooks.contexts import ToolSelectionContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        selection_ctx = ToolSelectionContext(
+            agent=self,
+            agent_role=getattr(self, "role", None),
+            task=task,
+            crew=self.crew,
+            tools=raw_tools,
+            payload=raw_tools,
+        )
+        dispatch(InterceptionPoint.TOOL_SELECTION, selection_ctx)
+        raw_tools = selection_ctx.payload
+
         parsed_tools = parse_tools(raw_tools)
 
         prompt, stop_words, rpm_limit_fn = self._build_execution_prompt(raw_tools)

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(

lib/crewai/src/crewai/crew.py

@@ -1687,6 +1687,9 @@ class Crew(FlowTrackable, BaseModel):
             if files_needing_tool:
                 tools = self._add_file_tools(tools, files_needing_tool)
 
+        # TOOL_SELECTION is dispatched once, in Agent.create_agent_executor,
+        # which every crew task funnels through. Dispatching here as well would
+        # fire the point twice on a crew run (and duplicate additive edits).
         return tools
 
     def _get_agent_to_use(self, task: Task) -> BaseAgent | None:
@@ -1906,6 +1909,30 @@ class Crew(FlowTrackable, BaseModel):
         final_string_output = final_task_output.raw
         self._finish_execution(final_string_output)
         self.token_usage = self.calculate_usage_metrics()
+
+        from crewai.hooks.contexts import ExecutionEndContext, OutputContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        crew_output = CrewOutput(
+            raw=final_task_output.raw,
+            pydantic=final_task_output.pydantic,
+            json_dict=final_task_output.json_dict,
+            tasks_output=task_outputs,
+            token_usage=self.token_usage,
+        )
+
+        # OUTPUT/EXECUTION_END run before the kickoff-completed event (mirroring
+        # the flow OUTPUT-before-FlowFinishedEvent ordering) so a HookAborted
+        # prevents a spurious completed signal and any payload replacement is
+        # honored on the returned output.
+        output_ctx = OutputContext(crew=self, output=crew_output, payload=crew_output)
+        dispatch(InterceptionPoint.OUTPUT, output_ctx)
+        crew_output = output_ctx.payload
+
+        end_ctx = ExecutionEndContext(crew=self, output=crew_output, payload=crew_output)
+        dispatch(InterceptionPoint.EXECUTION_END, end_ctx)
+        crew_output = end_ctx.payload
+
         # Ensure background memory saves finish (and emit their
         # completed/failed events) before the kickoff-completed event below
         # triggers listener teardown/finalization.
@@ -1924,13 +1951,7 @@ class Crew(FlowTrackable, BaseModel):
         # Finalization is handled by trace listener (always initialized)
         # The batch manager checks contextvar to determine if tracing is enabled
 
-        return CrewOutput(
-            raw=final_task_output.raw,
-            pydantic=final_task_output.pydantic,
-            json_dict=final_task_output.json_dict,
-            tasks_output=task_outputs,
-            token_usage=self.token_usage,
-        )
+        return crew_output
 
     def _process_async_tasks(
         self,

lib/crewai/src/crewai/crews/utils.py

@@ -278,6 +278,9 @@ def prepare_kickoff(
         reset_emission_counter()
         reset_last_event_id()
 
+    from crewai.hooks.contexts import ExecutionStartContext, InputContext
+    from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
     normalized: dict[str, Any] | None = None
     if inputs is not None:
         if not isinstance(inputs, Mapping):
@@ -286,11 +289,30 @@ def prepare_kickoff(
             )
         normalized = dict(inputs)
 
+    # ``inputs`` aliases the same object as ``payload`` (not a fresh ``{}`` from
+    # ``or``) so in-place edits to either survive read-back, per the context
+    # contract. ``None`` inputs are preserved rather than coerced to ``{}``.
+    start_ctx = ExecutionStartContext(
+        crew=crew,
+        inputs=normalized if normalized is not None else {},
+        payload=normalized,
+    )
+    dispatch(InterceptionPoint.EXECUTION_START, start_ctx)
+    normalized = start_ctx.payload
+
     for before_callback in crew.before_kickoff_callbacks:
         if normalized is None:
             normalized = {}
         normalized = before_callback(normalized)
 
+    input_ctx = InputContext(
+        crew=crew,
+        inputs=normalized if normalized is not None else {},
+        payload=normalized,
+    )
+    dispatch(InterceptionPoint.INPUT, input_ctx)
+    normalized = input_ctx.payload
+
     if resuming and crew._kickoff_event_id:
         if crew.verbose:
             from crewai.events.utils.console_formatter import ConsoleFormatter

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

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(

lib/crewai/src/crewai/flow/runtime/__init__.py

@@ -1476,6 +1476,22 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             else (resumed_method_output if emit else result)
         )
 
+        # A resumed flow completes here rather than in kickoff_async, so the
+        # OUTPUT/EXECUTION_END seams must fire on this path too (before
+        # FlowFinishedEvent) to expose the final result to policy hooks.
+        from crewai.hooks.contexts import ExecutionEndContext, OutputContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        output_ctx = OutputContext(flow=self, output=final_result, payload=final_result)
+        dispatch(InterceptionPoint.OUTPUT, output_ctx)
+        final_result = output_ctx.payload
+
+        end_ctx = ExecutionEndContext(
+            flow=self, output=final_result, payload=final_result
+        )
+        dispatch(InterceptionPoint.EXECUTION_END, end_ctx)
+        final_result = end_ctx.payload
+
         if self._event_futures:
             await asyncio.gather(
                 *[
@@ -2037,6 +2053,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         flow_name_token = None
         flow_defer_trace_finalization_token = None
         request_id_token = None
+        # Re-published after the INPUT hook so trigger-payload injection reads
+        # the hook-rewritten inputs rather than the pre-hook baggage above.
+        flow_inputs_token = None
         if current_flow_id.get() is None:
             flow_id_token = current_flow_id.set(self.flow_id)
             flow_name_token = current_flow_name.set(
@@ -2062,6 +2081,37 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             self._attach_usage_aggregation_listener()
 
         try:
+            from crewai.hooks.contexts import (
+                ExecutionEndContext,
+                ExecutionStartContext,
+                InputContext,
+                OutputContext,
+            )
+            from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+            # ``inputs`` aliases the same object as ``payload`` (not a fresh
+            # ``{}`` from ``or``) so in-place edits survive read-back.
+            start_ctx = ExecutionStartContext(
+                flow=self,
+                inputs=inputs if inputs is not None else {},
+                payload=inputs,
+            )
+            dispatch(InterceptionPoint.EXECUTION_START, start_ctx)
+            inputs = start_ctx.payload
+
+            input_ctx = InputContext(
+                flow=self,
+                inputs=inputs if inputs is not None else {},
+                payload=inputs,
+            )
+            dispatch(InterceptionPoint.INPUT, input_ctx)
+            inputs = input_ctx.payload
+
+            # Publish the resolved inputs so trigger-payload injection and other
+            # baggage readers observe hook rewrites (the baggage set before the
+            # hooks carried the pre-hook inputs).
+            flow_inputs_token = attach(baggage.set_baggage("flow_inputs", inputs or {}))
+
             # Reset flow state for fresh execution unless restoring from persistence
             is_restoring = (
                 inputs and "id" in inputs and self.persistence is not None
@@ -2297,6 +2347,21 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             method_outputs = self.method_outputs
             final_output = method_outputs[-1] if method_outputs else None
 
+            output_ctx = OutputContext(
+                flow=self, output=final_output, payload=final_output
+            )
+            dispatch(InterceptionPoint.OUTPUT, output_ctx)
+            final_output = output_ctx.payload
+
+            # EXECUTION_END runs before FlowFinishedEvent so a HookAborted
+            # prevents a spurious finished signal and payload replacement is
+            # honored on the emitted result and the returned value.
+            end_ctx = ExecutionEndContext(
+                flow=self, output=final_output, payload=final_output
+            )
+            dispatch(InterceptionPoint.EXECUTION_END, end_ctx)
+            final_output = end_ctx.payload
+
             if self._event_futures:
                 await asyncio.gather(
                     *[asyncio.wrap_future(f) for f in self._event_futures]
@@ -2370,6 +2435,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 current_flow_name.reset(flow_name_token)
             if flow_id_token is not None:
                 current_flow_id.reset(flow_id_token)
+            if flow_inputs_token is not None:
+                detach(flow_inputs_token)
             detach(flow_token)
             crewai_event_bus._exit_runtime_scope(runtime_scope)
 
@@ -2562,6 +2629,33 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 if future:
                     self._event_futures.append(future)
 
+            from crewai.hooks.contexts import StepContext
+            from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+            pre_step_ctx = StepContext(
+                kind="flow_method",
+                step_name=str(method_name),
+                flow=self,
+                payload=dumped_params,
+            )
+            dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
+
+            # Apply hook edits/replacement of the step params back onto the
+            # call. ``dumped_params`` maps positional args to ``_0, _1, ...``
+            # keys and keeps kwargs by name, so reverse that mapping here.
+            updated_params = pre_step_ctx.payload
+            if isinstance(updated_params, dict):
+                positional = sorted(
+                    (k for k in updated_params if k.startswith("_") and k[1:].isdigit()),
+                    key=lambda k: int(k[1:]),
+                )
+                args = tuple(updated_params[k] for k in positional)
+                kwargs = {
+                    k: v
+                    for k, v in updated_params.items()
+                    if not (k.startswith("_") and k[1:].isdigit())
+                }
+
             # Set method name in context so ask() can read it without
             # stack inspection.  Must happen before copy_context() so the
             # value propagates into the thread pool for sync methods.
@@ -2589,6 +2683,16 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     method_name, method_definition.human_feedback, result
                 )
 
+            post_step_ctx = StepContext(
+                kind="flow_method",
+                step_name=str(method_name),
+                flow=self,
+                output=result,
+                payload=result,
+            )
+            dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
+            result = post_step_ctx.payload
+
             self._method_outputs.append({"method": str(method_name), "output": result})
 
             # For @human_feedback methods with emit, the result is the collapsed outcome
@@ -2785,6 +2889,19 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     if isinstance(router_result, enum.Enum)
                     else router_result
                 )
+
+                from crewai.hooks.contexts import RouterDecisionContext
+                from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+                router_ctx = RouterDecisionContext(
+                    flow=self,
+                    router_name=str(router_name),
+                    route=router_result,
+                    payload=router_result,
+                )
+                dispatch(InterceptionPoint.ROUTER_DECISION, router_ctx)
+                router_result = router_ctx.payload
+
                 router_result_str = str(router_result)
                 router_result_event = FlowMethodName(router_result_str)
                 router_results.append(router_result_event)
@@ -2813,6 +2930,19 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     current_trigger, router_only=False
                 )
                 if listeners_triggered:
+                    from crewai.hooks.contexts import FlowTransitionContext
+                    from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+                    transition_ctx = FlowTransitionContext(
+                        flow=self,
+                        from_method=str(trigger_method),
+                        to_methods=[str(name) for name in listeners_triggered],
+                        trigger=str(current_trigger),
+                        payload=listeners_triggered,
+                    )
+                    dispatch(InterceptionPoint.FLOW_TRANSITION, transition_ctx)
+                    listeners_triggered = transition_ctx.payload
+
                     listener_result = router_result_payloads.get(
                         str(current_trigger), result
                     )

lib/crewai/src/crewai/flow/runtime/_actions.py

@@ -224,7 +224,34 @@ class ScriptAction:
 
     def run(self, *args: Any, **kwargs: Any) -> Any:
         local_context = _pop_local_context(kwargs)
-        return self.handler(
+
+        from crewai.hooks.contexts import PreCodeExecutionContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        code_ctx = PreCodeExecutionContext(
+            flow=self.flow,
+            code=self.definition.code,
+            language="python",
+            payload=self.definition.code,
+        )
+        dispatch(InterceptionPoint.PRE_CODE_EXECUTION, code_ctx)
+
+        # Honor a hook that rewrites the code, via either a returned payload
+        # replacement or an in-place ``ctx.code`` edit. Recompile only when the
+        # source actually changed so the common no-hook path stays free.
+        effective_code = (
+            code_ctx.payload
+            if isinstance(code_ctx.payload, str)
+            and code_ctx.payload != self.definition.code
+            else code_ctx.code
+        )
+        handler = (
+            self.handler
+            if effective_code == self.definition.code
+            else self._compile_handler(effective_code)
+        )
+
+        return handler(
             state=self.flow.state,
             outputs=outputs_by_name(
                 self.flow._method_outputs,
@@ -234,7 +261,7 @@ class ScriptAction:
             item=local_context.get("item") if local_context else None,
         )
 
-    def _compile_handler(self) -> Callable[..., Any]:
+    def _compile_handler(self, code: str | None = None) -> Callable[..., Any]:
         raw = os.environ.get(_ALLOW_SCRIPT_EXECUTION_ENV_VAR, "")
         if raw.strip().lower() not in _TRUSTED_SCRIPT_EXECUTION_VALUES:
             raise FlowScriptExecutionDisabledError(
@@ -243,8 +270,9 @@ class ScriptAction:
                 "trusted flow definitions."
             )
 
+        source = code if code is not None else self.definition.code
         filename = f"crewai.flow.script.{self.flow._definition.name}"
-        module = ast.parse(self.definition.code, filename=filename)
+        module = ast.parse(source, filename=filename)
         function = ast.FunctionDef(
             name="_flow_script",
             args=ast.arguments(

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",
 ]

lib/crewai/src/crewai/hooks/contexts.py

@@ -0,0 +1,166 @@
+"""Typed contexts for the interception points wired in phases 2-5.
+
+Each context is a dataclass whose fields are nullable and defaulted, so a field
+that is not meaningful for a given runtime (e.g. ``agent_role`` inside a flow)
+is simply ``None`` rather than an error. Every context exposes a ``payload``
+field: the interceptable value a hook may mutate in place or replace by
+returning a new value.
+
+The legacy ``pre/post_model_call`` and ``pre/post_tool_call`` points keep using
+:class:`~crewai.hooks.llm_hooks.LLMCallHookContext` and
+:class:`~crewai.hooks.tool_hooks.ToolCallHookContext` for backwards
+compatibility; they are intentionally not redefined here.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class InterceptionContext:
+    """Base context shared by the framework-native interception points."""
+
+    payload: Any = None
+    agent: Any = None
+    agent_role: str | None = None
+    task: Any = None
+    crew: Any = None
+    flow: Any = None
+
+
+@dataclass
+class ExecutionStartContext(InterceptionContext):
+    """``execution_start``: a crew or flow is about to begin. ``payload`` = inputs."""
+
+    inputs: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class InputContext(InterceptionContext):
+    """``input``: resolved inputs for an execution. ``payload`` = inputs."""
+
+    inputs: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class OutputContext(InterceptionContext):
+    """``output``: final result of a crew or flow. ``payload`` = the output object."""
+
+    output: Any = None
+
+
+@dataclass
+class ExecutionEndContext(InterceptionContext):
+    """``execution_end``: a crew or flow has finished. ``payload`` = the output object."""
+
+    output: Any = None
+
+
+@dataclass
+class StepContext(InterceptionContext):
+    """``pre_step`` / ``post_step``: a task or flow-method step boundary.
+
+    ``kind`` is ``"task"`` for crew tasks and ``"flow_method"`` for flow methods.
+    ``payload`` is the step input (pre) or step output (post).
+    """
+
+    kind: str | None = None
+    step_name: str | None = None
+    output: Any = None
+
+
+@dataclass
+class ToolSelectionContext(InterceptionContext):
+    """``tool_selection``: the set of tools offered to an agent. ``payload`` = tools list."""
+
+    tools: list[Any] = field(default_factory=list)
+
+
+@dataclass
+class PreDelegationContext(InterceptionContext):
+    """``pre_delegation``: an agent is about to delegate work. ``payload`` = delegation input."""
+
+    coworker: str | None = None
+    delegate_to: Any = None
+
+
+@dataclass
+class RetryAttemptContext(InterceptionContext):
+    """``retry_attempt``: an operation is about to be retried."""
+
+    attempt: int = 0
+    max_attempts: int | None = None
+    error: Any = None
+
+
+@dataclass
+class MemoryWriteContext(InterceptionContext):
+    """``memory_write``: a value is about to be written to memory. ``payload`` = value."""
+
+    memory_type: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class MemoryReadContext(InterceptionContext):
+    """``memory_read``: a memory query is being issued. ``payload`` = query (pre) / results (post)."""
+
+    memory_type: str | None = None
+    query: str | None = None
+
+
+@dataclass
+class KnowledgeRetrievalContext(InterceptionContext):
+    """``knowledge_retrieval``: a knowledge query. ``payload`` = query / retrieved results."""
+
+    query: Any = None
+
+
+@dataclass
+class PreCodeExecutionContext(InterceptionContext):
+    """``pre_code_execution``: code is about to run. ``payload`` = the code string."""
+
+    code: str | None = None
+    language: str | None = None
+
+
+@dataclass
+class MCPConnectContext(InterceptionContext):
+    """``mcp_connect``: an MCP client is about to connect. ``payload`` = connection params."""
+
+    server_name: str | None = None
+    server_params: Any = None
+
+
+@dataclass
+class FileAccessContext(InterceptionContext):
+    """``file_access``: reserved. No live consumer seam yet."""
+
+    path: str | None = None
+    mode: str | None = None
+
+
+@dataclass
+class ArtifactOutputContext(InterceptionContext):
+    """``artifact_output``: reserved. No live consumer seam yet."""
+
+    artifact: Any = None
+
+
+@dataclass
+class FlowTransitionContext(InterceptionContext):
+    """``flow_transition``: a flow is moving to triggered methods."""
+
+    from_method: str | None = None
+    to_methods: list[str] = field(default_factory=list)
+    trigger: str | None = None
+
+
+@dataclass
+class RouterDecisionContext(InterceptionContext):
+    """``router_decision``: a flow router is choosing a route. ``payload`` = route label."""
+
+    router_name: str | None = None
+    route: Any = None

lib/crewai/src/crewai/hooks/dispatch.py

@@ -0,0 +1,436 @@
+"""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.
+
+    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 _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``.
+
+    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

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(

lib/crewai/src/crewai/hooks/tool_hooks.py

@@ -5,6 +5,12 @@ from typing import TYPE_CHECKING, Any
 from crewai_core.printer import PRINTER
 
 from crewai.events.event_listener import event_listener
+from crewai.hooks.dispatch import (
+    HookAborted,
+    InterceptionPoint,
+    dispatch,
+    get_global_hook_list,
+)
 from crewai.hooks.types import (
     AfterToolCallHookCallable,
     AfterToolCallHookType,
@@ -121,8 +127,81 @@ class ToolCallHookContext:
             event_listener.formatter.resume_live_updates()
 
 
-_before_tool_call_hooks: list[BeforeToolCallHookType | BeforeToolCallHookCallable] = []
-_after_tool_call_hooks: list[AfterToolCallHookType | AfterToolCallHookCallable] = []
+# The legacy registries are aliased to the generic dispatcher's global hook
+# lists for the tool-call points, so legacy registrations and new-dialect
+# ``@on(InterceptionPoint.PRE_TOOL_CALL)`` hooks share one ordered queue.
+_before_tool_call_hooks: list[BeforeToolCallHookType | BeforeToolCallHookCallable] = (
+    get_global_hook_list(InterceptionPoint.PRE_TOOL_CALL)
+)
+_after_tool_call_hooks: list[AfterToolCallHookType | AfterToolCallHookCallable] = (
+    get_global_hook_list(InterceptionPoint.POST_TOOL_CALL)
+)
+
+
+def before_tool_call_reducer(context: ToolCallHookContext, result: object) -> bool:
+    """Legacy calling convention for ``pre_tool_call`` hooks.
+
+    A ``False`` return blocks the call (mapped to :class:`HookAborted`); tool
+    input is modified in place, so no payload replacement occurs here.
+    """
+    if result is False:
+        raise HookAborted(reason="before_tool_call hook returned False")
+    return False
+
+
+def after_tool_call_reducer(context: ToolCallHookContext, result: object) -> bool:
+    """Legacy calling convention for ``post_tool_call`` hooks.
+
+    A non-None return replaces the tool result on the context.
+    """
+    if result is not None:
+        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(

lib/crewai/src/crewai/knowledge/knowledge.py

@@ -145,6 +145,13 @@ class Knowledge(BaseModel):
         if self.storage is None:
             raise ValueError("Storage is not initialized.")
 
+        from crewai.hooks.contexts import KnowledgeRetrievalContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        retrieval_ctx = KnowledgeRetrievalContext(query=query, payload=query)
+        dispatch(InterceptionPoint.KNOWLEDGE_RETRIEVAL, retrieval_ctx)
+        query = retrieval_ctx.payload
+
         return self.storage.search(
             query,
             limit=results_limit,
@@ -183,6 +190,13 @@ class Knowledge(BaseModel):
         if self.storage is None:
             raise ValueError("Storage is not initialized.")
 
+        from crewai.hooks.contexts import KnowledgeRetrievalContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        retrieval_ctx = KnowledgeRetrievalContext(query=query, payload=query)
+        dispatch(InterceptionPoint.KNOWLEDGE_RETRIEVAL, retrieval_ctx)
+        query = retrieval_ctx.payload
+
         return await self.storage.asearch(
             query,
             limit=results_limit,

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

lib/crewai/src/crewai/mcp/client.py

@@ -152,6 +152,20 @@ class MCPClient:
         server_name, server_url, transport_type = self._get_server_info()
         is_reconnect = self._was_connected
 
+        from crewai.hooks.contexts import MCPConnectContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        connect_ctx = MCPConnectContext(
+            server_name=server_name,
+            server_params=self.transport,
+            payload=self.transport,
+        )
+        dispatch(InterceptionPoint.MCP_CONNECT, connect_ctx)
+        # Honor a hook that replaces the connection transport/params so the
+        # connection below actually uses the returned value.
+        if connect_ctx.payload is not None:
+            self.transport = connect_ctx.payload
+
         started_at = datetime.now()
         crewai_event_bus.emit(
             self,

lib/crewai/src/crewai/memory/unified_memory.py

@@ -466,6 +466,18 @@ class Memory(BaseModel):
         if self.read_only:
             return None
 
+        from crewai.hooks.contexts import MemoryWriteContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        write_ctx = MemoryWriteContext(
+            agent_role=agent_role,
+            memory_type="unified_memory",
+            metadata=metadata or {},
+            payload=content,
+        )
+        dispatch(InterceptionPoint.MEMORY_WRITE, write_ctx)
+        content = write_ctx.payload
+
         # Determine effective root_scope: per-call override takes precedence
         effective_root = root_scope if root_scope is not None else self.root_scope
 
@@ -561,6 +573,18 @@ class Memory(BaseModel):
         if not contents or self.read_only:
             return []
 
+        from crewai.hooks.contexts import MemoryWriteContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        write_ctx = MemoryWriteContext(
+            agent_role=agent_role,
+            memory_type="unified_memory",
+            metadata=metadata or {},
+            payload=contents,
+        )
+        dispatch(InterceptionPoint.MEMORY_WRITE, write_ctx)
+        contents = write_ctx.payload
+
         # Determine effective root_scope: per-call override takes precedence
         effective_root = root_scope if root_scope is not None else self.root_scope
 
@@ -712,6 +736,17 @@ class Memory(BaseModel):
         # so that the search sees all persisted records.
         self.drain_writes()
 
+        from crewai.hooks.contexts import MemoryReadContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        read_ctx = MemoryReadContext(
+            memory_type="unified_memory",
+            query=query,
+            payload=query,
+        )
+        dispatch(InterceptionPoint.MEMORY_READ, read_ctx)
+        query = read_ctx.payload
+
         effective_scope = scope
         if effective_scope is None and self.root_scope:
             effective_scope = self.root_scope

lib/crewai/src/crewai/task.py

@@ -662,6 +662,21 @@ class Task(BaseModel):
                 crewai_event_bus.emit(
                     self, TaskStartedEvent(context=context, task=self)
                 )
+
+            from crewai.hooks.contexts import StepContext
+            from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+            pre_step_ctx = StepContext(
+                kind="task",
+                step_name=self.name or self.description,
+                agent=agent,
+                agent_role=getattr(agent, "role", None),
+                task=self,
+                payload=context,
+            )
+            dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
+            context = pre_step_ctx.payload
+
             result = await agent.aexecute_task(
                 task=self,
                 context=context,
@@ -718,6 +733,18 @@ class Task(BaseModel):
                     guardrail=self._guardrail,
                 )
 
+            post_step_ctx = StepContext(
+                kind="task",
+                step_name=self.name or self.description,
+                agent=agent,
+                agent_role=getattr(agent, "role", None),
+                task=self,
+                output=task_output,
+                payload=task_output,
+            )
+            dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
+            task_output = post_step_ctx.payload
+
             self.output = task_output
             self.end_time = datetime.datetime.now()
 
@@ -787,6 +814,21 @@ class Task(BaseModel):
                 crewai_event_bus.emit(
                     self, TaskStartedEvent(context=context, task=self)
                 )
+
+            from crewai.hooks.contexts import StepContext
+            from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+            pre_step_ctx = StepContext(
+                kind="task",
+                step_name=self.name or self.description,
+                agent=agent,
+                agent_role=getattr(agent, "role", None),
+                task=self,
+                payload=context,
+            )
+            dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
+            context = pre_step_ctx.payload
+
             result = agent.execute_task(
                 task=self,
                 context=context,
@@ -843,6 +885,18 @@ class Task(BaseModel):
                     guardrail=self._guardrail,
                 )
 
+            post_step_ctx = StepContext(
+                kind="task",
+                step_name=self.name or self.description,
+                agent=agent,
+                agent_role=getattr(agent, "role", None),
+                task=self,
+                output=task_output,
+                payload=task_output,
+            )
+            dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
+            task_output = post_step_ctx.payload
+
             self.output = task_output
             self.end_time = datetime.datetime.now()
 
@@ -884,6 +938,32 @@ class Task(BaseModel):
             clear_task_files(self.id)
             reset_current_task_id(task_id_token)
 
+    def _dispatch_guardrail_retry_attempt(
+        self,
+        agent: BaseAgent | None,
+        context: str | None,
+        attempt: int,
+        error: Any,
+    ) -> str | None:
+        """Fire ``retry_attempt`` before re-executing a task after a guardrail failure.
+
+        Returns the (possibly hook-modified) retry context.
+        """
+        from crewai.hooks.contexts import RetryAttemptContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        retry_ctx = RetryAttemptContext(
+            agent=agent,
+            agent_role=getattr(agent, "role", None),
+            task=self,
+            attempt=attempt,
+            max_attempts=self.guardrail_max_retries,
+            error=error,
+            payload=context,
+        )
+        dispatch(InterceptionPoint.RETRY_ATTEMPT, retry_ctx)
+        return retry_ctx.payload
+
     def _post_agent_execution(self, agent: BaseAgent) -> None:
         pass
 
@@ -1317,6 +1397,13 @@ Follow these guidelines:
                     color="yellow",
                 )
 
+            context = self._dispatch_guardrail_retry_attempt(
+                agent=agent,
+                context=context,
+                attempt=current_retry_count,
+                error=guardrail_result.error,
+            )
+
             result = agent.execute_task(
                 task=self,
                 context=context,
@@ -1427,6 +1514,13 @@ Follow these guidelines:
                     color="yellow",
                 )
 
+            context = self._dispatch_guardrail_retry_attempt(
+                agent=agent,
+                context=context,
+                attempt=current_retry_count,
+                error=guardrail_result.error,
+            )
+
             result = await agent.aexecute_task(
                 task=self,
                 context=context,

lib/crewai/src/crewai/tools/agent_tools/base_agent_tools.py

@@ -108,6 +108,22 @@ class BaseAgentTool(BaseTool):
             )
 
         selected_agent = agent[0]
+
+        from crewai.hooks.contexts import PreDelegationContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        # Dispatched outside the try/except below so a HookAborted propagates
+        # instead of being swallowed into a tool-error string.
+        delegation_ctx = PreDelegationContext(
+            agent=selected_agent,
+            agent_role=getattr(selected_agent, "role", None),
+            coworker=sanitized_name,
+            delegate_to=selected_agent,
+            payload=task,
+        )
+        dispatch(InterceptionPoint.PRE_DELEGATION, delegation_ctx)
+        task = delegation_ctx.payload
+
         try:
             task_with_assigned_agent = Task(
                 description=task,

lib/crewai/src/crewai/tools/tool_usage.py

@@ -879,6 +879,22 @@ class ToolUsage:
                 return ToolUsageError(
                     f"{I18N_DEFAULT.errors('tool_usage_error').format(error=e)}\nMoving on then. {I18N_DEFAULT.slice('format').format(tool_names=self.tools_names)}"
                 )
+
+            from crewai.hooks.contexts import RetryAttemptContext
+            from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+            retry_ctx = RetryAttemptContext(
+                agent=self.agent,
+                agent_role=getattr(self.agent, "role", None),
+                task=self.task,
+                attempt=self._run_attempts,
+                max_attempts=self._max_parsing_attempts,
+                error=e,
+                payload=tool_string,
+            )
+            dispatch(InterceptionPoint.RETRY_ATTEMPT, retry_ctx)
+            tool_string = retry_ctx.payload
+
             return self._tool_calling(tool_string)
 
     def _validate_tool_input(self, tool_input: str | None) -> dict[str, Any]:

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:

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,24 @@ 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, 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 +133,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 +176,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 +216,24 @@ 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, False)
 
         tool_result = tool_usage.use(tool_calling, agent_action.text)
         raw_tool_result = tool_usage.get_last_raw_result(tool_result)
@@ -249,16 +249,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)
 

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.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"]
+
+    def test_agent_filter_falls_back_to_agent_role(self):
+        seen: list[str] = []
+
+        @on(InterceptionPoint.PRE_STEP, agents=["Researcher"])
+        def hook(ctx):
+            seen.append(ctx.agent_role)
+
+        # No agent object, only the agent_role string (e.g. flow seams).
+        dispatch(InterceptionPoint.PRE_STEP, _Ctx(agent_role="Writer"))
+        dispatch(InterceptionPoint.PRE_STEP, _Ctx(agent_role="Researcher"))
+        assert seen == ["Researcher"]
+
+    def test_unregister_resolves_filtered_wrapper(self):
+        @on(InterceptionPoint.PRE_TOOL_CALL, tools=["allowed_tool"])
+        def hook(ctx):
+            return None
+
+        assert len(get_hooks(InterceptionPoint.PRE_TOOL_CALL)) == 1
+        assert unregister_hook(InterceptionPoint.PRE_TOOL_CALL, hook) is True
+        assert get_hooks(InterceptionPoint.PRE_TOOL_CALL) == []
+
+
+class TestSharedQueueWithLegacyDialect:
+    """Legacy registrations and @on hooks compose in one ordered queue."""
+
+    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())
+            # Telemetry handlers run on the bus's thread pool; flush so the
+            # assertion doesn't race the emit.
+            crewai_event_bus.flush()
+
+        assert len(events) == 1
+        assert events[0].interception_point == "input"
+        assert events[0].outcome == "modified"
+        assert events[0].hook_count == 1
+
+    def test_event_reports_abort_outcome(self):
+        events: list[HookDispatchedEvent] = []
+
+        def blocker(ctx):
+            raise HookAborted(reason="blocked", source="policy")
+
+        register(InterceptionPoint.INPUT, blocker)
+
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(HookDispatchedEvent)
+            def _capture(_source, event):
+                events.append(event)
+
+            with pytest.raises(HookAborted):
+                dispatch(InterceptionPoint.INPUT, _Ctx())
+            crewai_event_bus.flush()
+
+        assert len(events) == 1
+        assert events[0].interception_point == "input"
+        assert events[0].outcome == "aborted"
+        assert events[0].abort_reason == "blocked"
+        assert events[0].abort_source == "policy"
+
+
+class TestNoOpOverhead:
+    """The no-op fast path must stay cheap (a single dict lookup)."""
+
+    def test_noop_dispatch_overhead_is_bounded(self):
+        # Relative (not absolute) budget: the no-op fast path is a dict lookup
+        # plus a guard, so it should stay within a wide multiple of a bare
+        # function call. This catches accidental O(n) regressions without
+        # depending on absolute timing on shared CI runners.
+        ctx = _Ctx()
+        iterations = 100_000
+
+        def _baseline(_c):
+            return _c
+
+        for _ in range(1000):  # warm up both paths
+            dispatch(InterceptionPoint.INPUT, ctx)
+            _baseline(ctx)
+
+        start = time.perf_counter()
+        for _ in range(iterations):
+            _baseline(ctx)
+        baseline = time.perf_counter() - start
+
+        start = time.perf_counter()
+        for _ in range(iterations):
+            dispatch(InterceptionPoint.INPUT, ctx)
+        noop = time.perf_counter() - start
+
+        assert noop < baseline * 50 + 5e-3

lib/crewai/tests/hooks/test_interception_conformance.py

@@ -0,0 +1,178 @@
+"""Conformance suite for the framework-native interception points.
+
+For each wired point this suite asserts the shared contract: the probe hook
+sees a well-shaped payload, an in-place/returned modification is honored, and a
+:class:`HookAborted` interrupts the step. Enterprise / ACS adapters build
+against these guarantees.
+"""
+
+from __future__ import annotations
+
+from crewai.flow.flow import Flow, listen, router, start
+from crewai.hooks.dispatch import (
+    HookAborted,
+    InterceptionPoint,
+    clear_all,
+    on,
+)
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def clear_dispatch_registry():
+    clear_all()
+    yield
+    clear_all()
+
+
+class _SimpleFlow(Flow):
+    @start()
+    def begin(self):
+        return "begin"
+
+    @listen(begin)
+    def finish(self, _):
+        return "flow-result"
+
+
+class TestFlowExecutionBoundaries:
+    """execution_start / input / output / execution_end on a flow."""
+
+    def test_all_boundary_points_fire_once(self):
+        fired: list[str] = []
+
+        for point in (
+            InterceptionPoint.EXECUTION_START,
+            InterceptionPoint.INPUT,
+            InterceptionPoint.OUTPUT,
+            InterceptionPoint.EXECUTION_END,
+        ):
+
+            @on(point)
+            def _probe(ctx, _point=point):
+                fired.append(_point.value)
+
+        _SimpleFlow().kickoff(inputs={"seed": 1})
+
+        assert fired == [
+            "execution_start",
+            "input",
+            "output",
+            "execution_end",
+        ]
+
+    def test_output_modification_is_honored(self):
+        @on(InterceptionPoint.OUTPUT)
+        def rewrite(ctx):
+            return "intercepted"
+
+        result = _SimpleFlow().kickoff()
+        assert result == "intercepted"
+
+    def test_input_payload_carries_inputs(self):
+        seen: dict = {}
+
+        @on(InterceptionPoint.INPUT)
+        def capture(ctx):
+            seen.update(ctx.payload or {})
+
+        _SimpleFlow().kickoff(inputs={"seed": 42})
+        assert seen == {"seed": 42}
+
+    def test_abort_at_execution_start_interrupts(self):
+        @on(InterceptionPoint.EXECUTION_START)
+        def block(ctx):
+            raise HookAborted(reason="not allowed", source="policy")
+
+        with pytest.raises(HookAborted) as exc:
+            _SimpleFlow().kickoff()
+        assert exc.value.reason == "not allowed"
+
+
+class TestFlowStepPoints:
+    """pre_step / post_step for flow methods (kind=flow_method)."""
+
+    def test_pre_and_post_step_fire_per_method(self):
+        kinds: list[tuple[str, str | None]] = []
+
+        @on(InterceptionPoint.PRE_STEP)
+        def pre(ctx):
+            kinds.append(("pre", ctx.step_name))
+
+        @on(InterceptionPoint.POST_STEP)
+        def post(ctx):
+            kinds.append(("post", ctx.step_name))
+
+        _SimpleFlow().kickoff()
+
+        assert ("pre", "begin") in kinds
+        assert ("post", "begin") in kinds
+        assert ("pre", "finish") in kinds
+        assert ("post", "finish") in kinds
+
+    def test_post_step_can_rewrite_method_output(self):
+        @on(InterceptionPoint.POST_STEP)
+        def rewrite(ctx):
+            if ctx.step_name == "finish":
+                return "rewritten"
+            return None
+
+        assert _SimpleFlow().kickoff() == "rewritten"
+
+
+class _RouterFlow(Flow):
+    @start()
+    def begin(self):
+        return "begin"
+
+    @router(begin)
+    def route(self):
+        return "go_left"
+
+    @listen("go_left")
+    def left(self):
+        return "left"
+
+    @listen("go_right")
+    def right(self):
+        return "right"
+
+
+class TestFlowTransitionAndRouter:
+    """flow_transition and router_decision on a routed flow."""
+
+    def test_transition_payload_carries_from_and_to(self):
+        seen: list[tuple[str | None, list[str]]] = []
+
+        @on(InterceptionPoint.FLOW_TRANSITION)
+        def capture(ctx):
+            seen.append((ctx.from_method, list(ctx.to_methods)))
+
+        _RouterFlow().kickoff()
+
+        assert any(to == ["left"] for _from, to in seen)
+
+    def test_router_decision_fires_with_route(self):
+        routes: list[object] = []
+
+        @on(InterceptionPoint.ROUTER_DECISION)
+        def capture(ctx):
+            routes.append(ctx.route)
+
+        _RouterFlow().kickoff()
+        assert "go_left" in routes
+
+    def test_router_decision_can_reroute(self):
+        @on(InterceptionPoint.ROUTER_DECISION)
+        def reroute(ctx):
+            return "go_right"
+
+        landed: list[str] = []
+
+        @on(InterceptionPoint.PRE_STEP)
+        def track(ctx):
+            landed.append(ctx.step_name)
+
+        _RouterFlow().kickoff()
+        assert "right" in landed
+        assert "left" not in landed