From 3444a2c1902ca7da1df6ded42f5531e3cd42c56e Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Sat, 11 Jul 2026 19:25:37 -0300 Subject: [PATCH] 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 | 29 ++++++++++--------- 5 files changed, 59 insertions(+), 30 deletions(-) diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index f2b0bf688..74a0015f8 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -1687,20 +1687,10 @@ class Crew(FlowTrackable, BaseModel): 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 dd552e5ff..81b8f0a1d 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -2640,6 +2640,22 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): ) 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 c21bc219a..47482f106 100644 --- a/lib/crewai/src/crewai/flow/runtime/_actions.py +++ b/lib/crewai/src/crewai/flow/runtime/_actions.py @@ -236,7 +236,22 @@ class ScriptAction: ) 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 @@ 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( @@ -255,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( diff --git a/lib/crewai/src/crewai/mcp/client.py b/lib/crewai/src/crewai/mcp/client.py index 7a1a77aeb..402f3a125 100644 --- a/lib/crewai/src/crewai/mcp/client.py +++ b/lib/crewai/src/crewai/mcp/client.py @@ -161,6 +161,10 @@ class MCPClient: 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 a5b69d566..61725de06 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 @@ 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: - 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,