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

@@ -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:

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.

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(

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(

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,