feat: add step interception points and rework execution hooks docs around @on (#6518)

* feat: add pre_step and post_step interception points on task execution

Introduces `StepContext` and the two step points in the dispatcher, and
wires them around agent execution in `task.py` (sync and async paths):
`pre_step` fires after `TaskStartedEvent` with the task context as
payload, `post_step` fires before `TaskCompletedEvent` with the
`TaskOutput`, and hook replacements are rebound in both directions.

* feat: wire pre_step and post_step on flow method execution

Dispatches the step points around each flow method with
kind="flow_method": `pre_step` receives the dumped call params and maps
returned edits back onto args/kwargs, `post_step` can rewrite the method
result before it is recorded. Conformance tests cover per-method firing
and output rewriting.

* docs: rework execution hooks page around the @on api

Replaces the standalone interception hooks catalog with a single
`execution-hooks.mdx` page that teaches `@on` as the primary way to
write hooks, covering the full ten-point catalog across task, flow, and
LLM execution. The legacy per-point decorators stay documented in a
closing section, and the `docs.json` navigation drops the removed page.

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

@@ -2626,6 +2626,37 @@ 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.
@@ -2653,6 +2684,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

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

@@ -56,3 +56,16 @@ 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

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

@@ -56,6 +56,10 @@ class InterceptionPoint(str, Enum):
     PRE_TOOL_CALL = "pre_tool_call"
     POST_TOOL_CALL = "post_tool_call"
 
+    # Step points
+    PRE_STEP = "pre_step"
+    POST_STEP = "post_step"
+
 
 class HookAborted(Exception):  # noqa: N818 - public contract name from OSS-86
     """Raised by a hook (or a legacy adapter) to abort the intercepted operation.

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 = cast(TaskOutput, post_step_ctx.payload)
+
             self.output = task_output
             self.end_time = datetime.datetime.now()
 
@@ -739,10 +766,12 @@ class Task(BaseModel):
 
             if self.output_file:
                 content = (
-                    json_output
-                    if json_output
+                    task_output.json_dict
+                    if task_output.json_dict
                     else (
-                        pydantic_output.model_dump_json() if pydantic_output else result
+                        task_output.pydantic.model_dump_json()
+                        if task_output.pydantic
+                        else task_output.raw
                     )
                 )
                 self._save_file(content)
@@ -787,6 +816,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 +887,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 = cast(TaskOutput, post_step_ctx.payload)
+
             self.output = task_output
             self.end_time = datetime.datetime.now()
 
@@ -864,10 +920,12 @@ class Task(BaseModel):
 
             if self.output_file:
                 content = (
-                    json_output
-                    if json_output
+                    task_output.json_dict
+                    if task_output.json_dict
                     else (
-                        pydantic_output.model_dump_json() if pydantic_output else result
+                        task_output.pydantic.model_dump_json()
+                        if task_output.pydantic
+                        else task_output.raw
                     )
                 )
                 self._save_file(content)
@@ -1316,7 +1374,6 @@ Follow these guidelines:
                     content=f"Guardrail {guardrail_index if guardrail_index is not None else ''} blocked (attempt {attempt + 1}/{max_attempts}), retrying due to: {guardrail_result.error}\n",
                     color="yellow",
                 )
-
             result = agent.execute_task(
                 task=self,
                 context=context,
@@ -1426,7 +1483,6 @@ Follow these guidelines:
                     content=f"Guardrail {guardrail_index if guardrail_index is not None else ''} blocked (attempt {attempt + 1}/{max_attempts}), retrying due to: {guardrail_result.error}\n",
                     color="yellow",
                 )
-
             result = await agent.aexecute_task(
                 task=self,
                 context=context,

lib/crewai/tests/hooks/test_interception_conformance.py

@@ -7,6 +7,9 @@ sees a well-shaped payload, an in-place/returned modification is honored, and a
 
 from __future__ import annotations
 
+from unittest.mock import patch
+
+from crewai.agent import Agent
 from crewai.flow.flow import Flow, listen, start
 from crewai.hooks.dispatch import (
     HookAborted,
@@ -14,6 +17,7 @@ from crewai.hooks.dispatch import (
     clear_all,
     on,
 )
+from crewai.task import Task
 import pytest
 
 
@@ -86,3 +90,60 @@ class TestFlowExecutionBoundaries:
         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 TestTaskStepPoints:
+    """pre_step / post_step for task execution (kind=task)."""
+
+    def test_post_step_rewrite_is_persisted_to_output_file(
+        self, tmp_path, monkeypatch
+    ):
+        @on(InterceptionPoint.POST_STEP)
+        def sanitize(ctx):
+            return ctx.payload.model_copy(update={"raw": "sanitized output"})
+
+        monkeypatch.chdir(tmp_path)
+        agent = Agent(role="Writer", goal="Write", backstory="Writes things.")
+        task = Task(
+            description="Write something",
+            expected_output="Some text",
+            output_file="output.txt",
+            agent=agent,
+        )
+
+        with patch.object(Agent, "execute_task", return_value="original output"):
+            result = task.execute_sync(agent=agent)
+
+        assert result.raw == "sanitized output"
+        assert (tmp_path / "output.txt").read_text() == "sanitized output"