mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-07-22 07:15:10 +00:00
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.
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user