From 10f54e0ebea50a573b244546c42b38d658ed59c9 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Mon, 13 Jul 2026 15:18:57 -0300 Subject: [PATCH 1/2] feat: wire remaining interception points and document the catalog Wires the step, agent, subsystem, and flow points onto the dispatcher: `pre_step`/`post_step`, `tool_selection`, `pre_delegation`, `retry_attempt`, `memory_write`/`memory_read`, `knowledge_retrieval`, `pre_code_execution`, `mcp_connect`, `flow_transition`, and `router_decision`, extending the typed- context module with their contexts. Each seam passes a typed context whose `payload` a hook may observe, mutate, or replace. Adds the conformance suite for these points and a new `interception-hooks` doc page with the full point/payload catalog and the interceptor contract. --- docs/docs.json | 3 +- docs/edge/en/learn/execution-hooks.mdx | 8 + docs/edge/en/learn/interception-hooks.mdx | 162 ++++++++++++++++++ lib/crewai/src/crewai/agent/core.py | 33 ++++ lib/crewai/src/crewai/crew.py | 15 +- .../src/crewai/flow/runtime/__init__.py | 47 +++++ .../src/crewai/flow/runtime/_actions.py | 12 ++ lib/crewai/src/crewai/hooks/contexts.py | 93 ++++++++++ lib/crewai/src/crewai/hooks/dispatch.py | 18 ++ lib/crewai/src/crewai/knowledge/knowledge.py | 14 ++ lib/crewai/src/crewai/mcp/client.py | 10 ++ .../src/crewai/memory/unified_memory.py | 35 ++++ lib/crewai/src/crewai/task.py | 94 ++++++++++ .../tools/agent_tools/base_agent_tools.py | 13 ++ lib/crewai/src/crewai/tools/tool_usage.py | 16 ++ .../hooks/test_interception_conformance.py | 91 +++++++++- 16 files changed, 661 insertions(+), 3 deletions(-) create mode 100644 docs/edge/en/learn/interception-hooks.mdx diff --git a/docs/docs.json b/docs/docs.json index f6d122bb3a..e2e349c001 100644 --- a/docs/docs.json +++ b/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" ] }, { diff --git a/docs/edge/en/learn/execution-hooks.mdx b/docs/edge/en/learn/execution-hooks.mdx index f9f667e4fb..11b0b3ef3d 100644 --- a/docs/edge/en/learn/execution-hooks.mdx +++ b/docs/edge/en/learn/execution-hooks.mdx @@ -42,6 +42,14 @@ Control and monitor tool execution: [View Tool Hooks Documentation →](/learn/tool-hooks) + +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. + + ## Hook Registration Methods ### 1. Decorator-Based Hooks (Recommended) diff --git a/docs/edge/en/learn/interception-hooks.mdx b/docs/edge/en/learn/interception-hooks.mdx new file mode 100644 index 0000000000..3377e70230 --- /dev/null +++ b/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) diff --git a/lib/crewai/src/crewai/agent/core.py b/lib/crewai/src/crewai/agent/core.py index 4c20693a8e..3e0d7f7c3d 100644 --- a/lib/crewai/src/crewai/agent/core.py +++ b/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) diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index db49b3a1e0..f2b0bf688a 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -1687,7 +1687,20 @@ def is_auto_injected(content_type: str) -> bool: if files_needing_tool: tools = self._add_file_tools(tools, files_needing_tool) - return tools + from crewai.hooks.contexts import ToolSelectionContext + from crewai.hooks.dispatch import InterceptionPoint, dispatch + + selection_ctx = ToolSelectionContext( + agent=agent, + agent_role=getattr(agent, "role", None), + task=task, + crew=self, + tools=tools, + payload=tools, + ) + dispatch(InterceptionPoint.TOOL_SELECTION, selection_ctx) + + return selection_ctx.payload def _get_agent_to_use(self, task: Task) -> BaseAgent | None: if self.process == Process.hierarchical: diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index 812d83268f..dd552e5ff2 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -2629,6 +2629,17 @@ 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) + # 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 +2667,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 +2873,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 +2914,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 ) diff --git a/lib/crewai/src/crewai/flow/runtime/_actions.py b/lib/crewai/src/crewai/flow/runtime/_actions.py index 885a2f2261..c21bc219a6 100644 --- a/lib/crewai/src/crewai/flow/runtime/_actions.py +++ b/lib/crewai/src/crewai/flow/runtime/_actions.py @@ -224,6 +224,18 @@ def __init__(self, flow: Flow[Any], definition: FlowScriptActionDefinition) -> N def run(self, *args: Any, **kwargs: Any) -> Any: local_context = _pop_local_context(kwargs) + + 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) + return self.handler( state=self.flow.state, outputs=outputs_by_name( diff --git a/lib/crewai/src/crewai/hooks/contexts.py b/lib/crewai/src/crewai/hooks/contexts.py index d718903b13..276ac890aa 100644 --- a/lib/crewai/src/crewai/hooks/contexts.py +++ b/lib/crewai/src/crewai/hooks/contexts.py @@ -56,3 +56,96 @@ 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 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 diff --git a/lib/crewai/src/crewai/hooks/dispatch.py b/lib/crewai/src/crewai/hooks/dispatch.py index 095484220e..a20059a54e 100644 --- a/lib/crewai/src/crewai/hooks/dispatch.py +++ b/lib/crewai/src/crewai/hooks/dispatch.py @@ -56,6 +56,24 @@ class InterceptionPoint(str, Enum): 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" + + # 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. diff --git a/lib/crewai/src/crewai/knowledge/knowledge.py b/lib/crewai/src/crewai/knowledge/knowledge.py index 76198fec97..77b956b584 100644 --- a/lib/crewai/src/crewai/knowledge/knowledge.py +++ b/lib/crewai/src/crewai/knowledge/knowledge.py @@ -145,6 +145,13 @@ def query( 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 @@ async def aquery( 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, diff --git a/lib/crewai/src/crewai/mcp/client.py b/lib/crewai/src/crewai/mcp/client.py index adf1afb8ca..7a1a77aeb0 100644 --- a/lib/crewai/src/crewai/mcp/client.py +++ b/lib/crewai/src/crewai/mcp/client.py @@ -152,6 +152,16 @@ async def connect(self) -> Self: 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) + started_at = datetime.now() crewai_event_bus.emit( self, diff --git a/lib/crewai/src/crewai/memory/unified_memory.py b/lib/crewai/src/crewai/memory/unified_memory.py index dcd5383ceb..47ee7b870e 100644 --- a/lib/crewai/src/crewai/memory/unified_memory.py +++ b/lib/crewai/src/crewai/memory/unified_memory.py @@ -466,6 +466,18 @@ def remember( 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 @@ def remember_many( 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 @@ def recall( # 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 diff --git a/lib/crewai/src/crewai/task.py b/lib/crewai/src/crewai/task.py index c63cfe8666..c1d19f2335 100644 --- a/lib/crewai/src/crewai/task.py +++ b/lib/crewai/src/crewai/task.py @@ -662,6 +662,21 @@ async def _aexecute_core( 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 @@ async def _aexecute_core( 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 @@ def _execute_core( 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 @@ def _execute_core( 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 @@ def _execute_core( 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 @@ def _invoke_guardrail_function( 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 @@ async def _ainvoke_guardrail_function( 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, diff --git a/lib/crewai/src/crewai/tools/agent_tools/base_agent_tools.py b/lib/crewai/src/crewai/tools/agent_tools/base_agent_tools.py index 19c19914c9..a5b69d5669 100644 --- a/lib/crewai/src/crewai/tools/agent_tools/base_agent_tools.py +++ b/lib/crewai/src/crewai/tools/agent_tools/base_agent_tools.py @@ -109,6 +109,19 @@ def _execute( selected_agent = agent[0] try: + from crewai.hooks.contexts import PreDelegationContext + from crewai.hooks.dispatch import InterceptionPoint, dispatch + + 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 + task_with_assigned_agent = Task( description=task, agent=selected_agent, diff --git a/lib/crewai/src/crewai/tools/tool_usage.py b/lib/crewai/src/crewai/tools/tool_usage.py index 4a973add9f..112d35e7e1 100644 --- a/lib/crewai/src/crewai/tools/tool_usage.py +++ b/lib/crewai/src/crewai/tools/tool_usage.py @@ -879,6 +879,22 @@ def _tool_calling( 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]: diff --git a/lib/crewai/tests/hooks/test_interception_conformance.py b/lib/crewai/tests/hooks/test_interception_conformance.py index 3593699d52..329961d89c 100644 --- a/lib/crewai/tests/hooks/test_interception_conformance.py +++ b/lib/crewai/tests/hooks/test_interception_conformance.py @@ -8,7 +8,7 @@ from __future__ import annotations -from crewai.flow.flow import Flow, listen, start +from crewai.flow.flow import Flow, listen, router, start from crewai.hooks.dispatch import ( HookAborted, InterceptionPoint, @@ -87,3 +87,92 @@ def block(ctx): 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 From ae9306d4a31c30d011a73d16814db9186fbd2ae4 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Mon, 13 Jul 2026 15:19:05 -0300 Subject: [PATCH 2/2] 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. --- lib/crewai/src/crewai/crew.py | 18 +++---------- .../src/crewai/flow/runtime/__init__.py | 16 +++++++++++ .../src/crewai/flow/runtime/_actions.py | 22 ++++++++++++--- lib/crewai/src/crewai/mcp/client.py | 4 +++ .../tools/agent_tools/base_agent_tools.py | 27 ++++++++++--------- 5 files changed, 58 insertions(+), 29 deletions(-) diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index f2b0bf688a..74a0015f8c 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -1687,20 +1687,10 @@ def is_auto_injected(content_type: str) -> bool: if files_needing_tool: tools = self._add_file_tools(tools, files_needing_tool) - from crewai.hooks.contexts import ToolSelectionContext - from crewai.hooks.dispatch import InterceptionPoint, dispatch - - selection_ctx = ToolSelectionContext( - agent=agent, - agent_role=getattr(agent, "role", None), - task=task, - crew=self, - tools=tools, - payload=tools, - ) - dispatch(InterceptionPoint.TOOL_SELECTION, selection_ctx) - - return selection_ctx.payload + # 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: if self.process == Process.hierarchical: diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index dd552e5ff2..81b8f0a1d2 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -2640,6 +2640,22 @@ async def _execute_method( ) 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. diff --git a/lib/crewai/src/crewai/flow/runtime/_actions.py b/lib/crewai/src/crewai/flow/runtime/_actions.py index c21bc219a6..47482f1064 100644 --- a/lib/crewai/src/crewai/flow/runtime/_actions.py +++ b/lib/crewai/src/crewai/flow/runtime/_actions.py @@ -236,7 +236,22 @@ def run(self, *args: Any, **kwargs: Any) -> Any: ) dispatch(InterceptionPoint.PRE_CODE_EXECUTION, code_ctx) - return self.handler( + # 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, @@ -246,7 +261,7 @@ def run(self, *args: Any, **kwargs: Any) -> Any: 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( @@ -255,8 +270,9 @@ def _compile_handler(self) -> Callable[..., Any]: "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( diff --git a/lib/crewai/src/crewai/mcp/client.py b/lib/crewai/src/crewai/mcp/client.py index 7a1a77aeb0..402f3a125d 100644 --- a/lib/crewai/src/crewai/mcp/client.py +++ b/lib/crewai/src/crewai/mcp/client.py @@ -161,6 +161,10 @@ async def connect(self) -> Self: 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( diff --git a/lib/crewai/src/crewai/tools/agent_tools/base_agent_tools.py b/lib/crewai/src/crewai/tools/agent_tools/base_agent_tools.py index a5b69d5669..61725de066 100644 --- a/lib/crewai/src/crewai/tools/agent_tools/base_agent_tools.py +++ b/lib/crewai/src/crewai/tools/agent_tools/base_agent_tools.py @@ -108,20 +108,23 @@ def _execute( ) selected_agent = agent[0] - try: - from crewai.hooks.contexts import PreDelegationContext - from crewai.hooks.dispatch import InterceptionPoint, dispatch - 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 + 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, agent=selected_agent,