From 356215a9f7e4aeda756dff75cba6bbb31b88b215 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Tue, 14 Jul 2026 03:14:45 -0300 Subject: [PATCH] refactor: drop speculative interception points from the catalog MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removes `pre_delegation`, `retry_attempt`, `flow_transition`, and `router_decision` — none has a hard use case yet, and `retry_attempt` was wired at three unrelated seams (agent task retry, tool parse retry, guardrail retry) with a different payload at each, so its semantics were ambiguous. Also casts the `post_step` payload back to `TaskOutput` in `task.py` so the seam keeps mypy green. --- docs/edge/en/learn/interception-hooks.mdx | 13 +--- lib/crewai/src/crewai/agent/core.py | 18 ------ .../src/crewai/flow/runtime/__init__.py | 32 ++-------- lib/crewai/src/crewai/hooks/contexts.py | 34 ----------- lib/crewai/src/crewai/hooks/dispatch.py | 6 -- lib/crewai/src/crewai/task.py | 46 +------------- .../tools/agent_tools/base_agent_tools.py | 16 ----- lib/crewai/src/crewai/tools/tool_usage.py | 16 ----- .../hooks/test_interception_conformance.py | 60 +------------------ 9 files changed, 10 insertions(+), 231 deletions(-) diff --git a/docs/edge/en/learn/interception-hooks.mdx b/docs/edge/en/learn/interception-hooks.mdx index 3377e7023..952f18139 100644 --- a/docs/edge/en/learn/interception-hooks.mdx +++ b/docs/edge/en/learn/interception-hooks.mdx @@ -6,8 +6,8 @@ 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. +every model call, tool call, and memory read, 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 @@ -98,8 +98,6 @@ behavior. | `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`. @@ -114,13 +112,6 @@ behavior. | `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 diff --git a/lib/crewai/src/crewai/agent/core.py b/lib/crewai/src/crewai/agent/core.py index 3e0d7f7c3..4f0c36f51 100644 --- a/lib/crewai/src/crewai/agent/core.py +++ b/lib/crewai/src/crewai/agent/core.py @@ -656,22 +656,6 @@ 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. @@ -725,7 +709,6 @@ 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( @@ -747,7 +730,6 @@ 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: diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index 5ca8fc325..eaf2b8b38 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -2643,7 +2643,11 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): 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()), + ( + 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) @@ -2886,19 +2890,6 @@ 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) @@ -2927,19 +2918,6 @@ 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 ) diff --git a/lib/crewai/src/crewai/hooks/contexts.py b/lib/crewai/src/crewai/hooks/contexts.py index 276ac890a..c0063b14c 100644 --- a/lib/crewai/src/crewai/hooks/contexts.py +++ b/lib/crewai/src/crewai/hooks/contexts.py @@ -78,23 +78,6 @@ class ToolSelectionContext(InterceptionContext): 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.""" @@ -132,20 +115,3 @@ class MCPConnectContext(InterceptionContext): 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 a20059a54..75f7bcfb8 100644 --- a/lib/crewai/src/crewai/hooks/dispatch.py +++ b/lib/crewai/src/crewai/hooks/dispatch.py @@ -60,8 +60,6 @@ class InterceptionPoint(str, Enum): 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" @@ -70,10 +68,6 @@ class InterceptionPoint(str, Enum): 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/task.py b/lib/crewai/src/crewai/task.py index c1d19f233..8cdc8b4da 100644 --- a/lib/crewai/src/crewai/task.py +++ b/lib/crewai/src/crewai/task.py @@ -743,7 +743,7 @@ class Task(BaseModel): payload=task_output, ) dispatch(InterceptionPoint.POST_STEP, post_step_ctx) - task_output = post_step_ctx.payload + task_output = cast(TaskOutput, post_step_ctx.payload) self.output = task_output self.end_time = datetime.datetime.now() @@ -895,7 +895,7 @@ class Task(BaseModel): payload=task_output, ) dispatch(InterceptionPoint.POST_STEP, post_step_ctx) - task_output = post_step_ctx.payload + task_output = cast(TaskOutput, post_step_ctx.payload) self.output = task_output self.end_time = datetime.datetime.now() @@ -938,32 +938,6 @@ 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 @@ -1396,14 +1370,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", ) - - 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, @@ -1513,14 +1479,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", ) - - 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 61725de06..19c19914c 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,22 +108,6 @@ 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, diff --git a/lib/crewai/src/crewai/tools/tool_usage.py b/lib/crewai/src/crewai/tools/tool_usage.py index 112d35e7e..4a973add9 100644 --- a/lib/crewai/src/crewai/tools/tool_usage.py +++ b/lib/crewai/src/crewai/tools/tool_usage.py @@ -879,22 +879,6 @@ 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]: diff --git a/lib/crewai/tests/hooks/test_interception_conformance.py b/lib/crewai/tests/hooks/test_interception_conformance.py index 23b897903..0ac33e6ea 100644 --- a/lib/crewai/tests/hooks/test_interception_conformance.py +++ b/lib/crewai/tests/hooks/test_interception_conformance.py @@ -7,7 +7,7 @@ sees a well-shaped payload, an in-place/returned modification is honored, and a from __future__ import annotations -from crewai.flow.flow import Flow, listen, router, start +from crewai.flow.flow import Flow, listen, start from crewai.hooks.dispatch import ( HookAborted, InterceptionPoint, @@ -117,61 +117,3 @@ class TestFlowStepPoints: 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