feat: wire remaining interception points and document the catalog

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,162 @@
+---
+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 |
+
+### 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 @@ def _finalize_task_execution(self, task: Task, result: Any) -> Any:
 
         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 @@ def _handle_execution_error(
             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 @@ async def _handle_execution_error_async(
             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 @@ def create_agent_executor(
             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/crew.py

@@ -1687,6 +1687,9 @@ def is_auto_injected(content_type: str) -> bool:
             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:

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

@@ -2629,6 +2629,33 @@ async def _execute_method(
                 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.
@@ -2656,6 +2683,16 @@ async def _execute_method(
                     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
@@ -2852,6 +2889,19 @@ async def _execute_listeners(
                     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)
@@ -2880,6 +2930,19 @@ async def _execute_listeners(
                     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
                     )