From 50f5c85ca13fa4ede6c5321b4962df2691b6b5a0 Mon Sep 17 00:00:00 2001 From: Shakti Date: Thu, 23 Jul 2026 16:45:37 +0530 Subject: [PATCH 1/5] fix: await async tools natively in _ainvoke_loop_native_tools MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the async native tool path (_ainvoke_loop_native_tools) calls the sync _handle_native_tool_calls, which calls tool.run() -> asyncio.run(), the asyncio.run() call crashes with 'RuntimeError: asyncio.run() cannot be called from a running event loop' when the agent is invoked from an already-running event loop (e.g. via ainvoke()). The fix adds three async methods: 1. _ahandle_native_tool_calls() — async variant that uses asyncio.gather instead of ThreadPoolExecutor for parallel tool execution, so async tools are properly awaited rather than run through asyncio.run(). 2. _aexecute_single_native_tool_call() — async variant that calls await tool.arun() instead of tool.run(), avoiding the nested asyncio.run() crash. 3. _ainvoke_loop_native_tools() now calls _ahandle_native_tool_calls() instead of _handle_native_tool_calls(). The ReAct executor (_ainvoke_loop_react) already handles this correctly via aexecute_tool_and_check_finality() -> tool_usage.ause() -> await. Fixes #6611 --- .../src/crewai/agents/crew_agent_executor.py | 330 +++++++++++++++++- 1 file changed, 329 insertions(+), 1 deletion(-) diff --git a/lib/crewai/src/crewai/agents/crew_agent_executor.py b/lib/crewai/src/crewai/agents/crew_agent_executor.py index bd7d313d35..070ea8056f 100644 --- a/lib/crewai/src/crewai/agents/crew_agent_executor.py +++ b/lib/crewai/src/crewai/agents/crew_agent_executor.py @@ -826,6 +826,140 @@ def _handle_native_tool_calls( self.messages.append(reasoning_message) return None + async def _ahandle_native_tool_calls( + self, + tool_calls: list[Any], + available_functions: dict[str, Callable[..., Any]], + ) -> AgentFinish | None: + """Async variant of ``_handle_native_tool_calls``. + + Uses ``await tool.arun()`` instead of ``tool.run()`` so that async tools + are properly awaited inside a running event loop, avoiding the + ``RuntimeError: asyncio.run() cannot be called from a running event loop`` + crash that occurs when ``_ainvoke_loop_native_tools`` calls the sync path. + """ + if not tool_calls: + return None + + parsed_calls = [ + parsed + for tool_call in tool_calls + if (parsed := self._parse_native_tool_call(tool_call)) is not None + ] + if not parsed_calls: + return None + + original_tools_by_name: dict[str, Any] = dict(self._tool_name_mapping) + + if len(parsed_calls) > 1: + has_result_as_answer_in_batch = any( + bool( + original_tools_by_name.get(func_name) + and getattr( + original_tools_by_name.get(func_name), "result_as_answer", False + ) + ) + for _, func_name, _ in parsed_calls + ) + has_max_usage_count_in_batch = any( + bool( + original_tools_by_name.get(func_name) + and getattr( + original_tools_by_name.get(func_name), + "max_usage_count", + None, + ) + is not None + ) + for _, func_name, _ in parsed_calls + ) + + if has_result_as_answer_in_batch or has_max_usage_count_in_batch: + logger.debug( + "Skipping parallel native execution because batch includes result_as_answer or max_usage_count tool" + ) + else: + execution_plan: list[ + tuple[str, str, str | dict[str, Any], Any | None] + ] = [] + for call_id, func_name, func_args in parsed_calls: + original_tool = original_tools_by_name.get(func_name) + execution_plan.append( + (call_id, func_name, func_args, original_tool) + ) + + self._append_assistant_tool_calls_message( + [ + (call_id, func_name, func_args) + for call_id, func_name, func_args, _ in execution_plan + ] + ) + + # Use asyncio.gather instead of ThreadPoolExecutor so async tools + # are properly awaited rather than run through asyncio.run(). + import asyncio + + async def _run_one( + call_id: str, + func_name: str, + func_args: str | dict[str, Any], + original_tool: Any | None, + ) -> dict[str, Any] | None: + return await self._aexecute_single_native_tool_call( + call_id=call_id, + func_name=func_name, + func_args=func_args, + available_functions=available_functions, + original_tool=original_tool, + should_execute=True, + ) + + tasks = [ + _run_one(call_id, func_name, func_args, original_tool) + for call_id, func_name, func_args, original_tool in execution_plan + ] + ordered_results = await asyncio.gather(*tasks) + + for execution_result in ordered_results: + if not execution_result: + continue + tool_finish = self._append_tool_result_and_check_finality( + execution_result + ) + if tool_finish: + return tool_finish + + reasoning_prompt = I18N_DEFAULT.slice("post_tool_reasoning") + reasoning_message: LLMMessage = { + "role": "user", + "content": reasoning_prompt, + } + self.messages.append(reasoning_message) + return None + + call_id, func_name, func_args = parsed_calls[0] + self._append_assistant_tool_calls_message([(call_id, func_name, func_args)]) + + execution_result = await self._aexecute_single_native_tool_call( + call_id=call_id, + func_name=func_name, + func_args=func_args, + available_functions=available_functions, + original_tool=original_tools_by_name.get(func_name), + should_execute=True, + ) + tool_finish = self._append_tool_result_and_check_finality(execution_result) + if tool_finish: + return tool_finish + + reasoning_prompt = I18N_DEFAULT.slice("post_tool_reasoning") + reasoning_message = { + "role": "user", + "content": reasoning_prompt, + } + self.messages.append(reasoning_message) + return None + def _parse_native_tool_call( self, tool_call: Any ) -> tuple[str, str, str | dict[str, Any]] | None: @@ -1068,6 +1202,200 @@ def _execute_single_native_tool_call( "original_tool": original_tool, } + async def _aexecute_single_native_tool_call( + self, + *, + call_id: str, + func_name: str, + func_args: str | dict[str, Any], + available_functions: dict[str, Callable[..., Any]], + original_tool: Any | None = None, + should_execute: bool = True, + ) -> dict[str, Any]: + """Async variant of ``_execute_single_native_tool_call``. + + Uses ``await tool.arun()`` instead of ``tool.run()`` so that async tools + are properly awaited inside a running event loop, avoiding the + ``RuntimeError: asyncio.run() cannot be called from a running event loop`` + crash. + """ + from datetime import datetime + import json + + from crewai.events.types.tool_usage_events import ( + ToolUsageErrorEvent, + ToolUsageFinishedEvent, + ToolUsageStartedEvent, + ) + + args_dict, parse_error = parse_tool_call_args( + func_args, func_name, call_id, original_tool + ) + if parse_error is not None: + return parse_error + + if original_tool is None: + for tool in self.original_tools or []: + if sanitize_tool_name(tool.name) == func_name: + original_tool = tool + break + + max_usage_reached = False + if not should_execute and original_tool: + max_usage_reached = True + elif ( + should_execute + and original_tool + and (max_count := getattr(original_tool, "max_usage_count", None)) + is not None + and getattr(original_tool, "current_usage_count", 0) >= max_count + ): + max_usage_reached = True + + structured_tool: CrewStructuredTool | None = None + if original_tool is not None: + for structured in self.tools or []: + if getattr(structured, "_original_tool", None) is original_tool: + structured_tool = structured + break + if structured_tool is None: + for structured in self.tools or []: + if sanitize_tool_name(structured.name) == func_name: + structured_tool = structured + break + + output_tool = original_tool or structured_tool + + from_cache = False + result: str = "Tool not found" + raw_tool_result: Any = result + input_str = json.dumps(args_dict) if args_dict else "" + if self.tools_handler and self.tools_handler.cache and output_tool is not None: + cached_result = self.tools_handler.cache.read( + tool=func_name, input=input_str + ) + if cached_result is not None: + raw_tool_result = cached_result + result = format_native_tool_output_for_agent(output_tool, cached_result) + from_cache = True + + agent_key = getattr(self.agent, "key", "unknown") if self.agent else "unknown" + started_at = datetime.now() + crewai_event_bus.emit( + self, + event=ToolUsageStartedEvent( + tool_name=func_name, + tool_args=args_dict, + from_agent=self.agent, + from_task=self.task, + agent_key=agent_key, + ), + ) + error_event_emitted = False + + track_delegation_if_needed(func_name, args_dict or {}, self.task) + + before_hook_context = ToolCallHookContext( + tool_name=func_name, + tool_input=args_dict or {}, + tool=structured_tool, + agent=self.agent, + task=self.task, + crew=self.crew, + ) + hook_blocked = run_before_tool_call_hooks(before_hook_context) + + if hook_blocked: + result = f"Tool execution blocked by hook. Tool: {func_name}" + raw_tool_result = result + elif max_usage_reached and original_tool: + result = f"Tool '{func_name}' has reached its usage limit of {original_tool.max_usage_count} times and cannot be used anymore." + raw_tool_result = result + elif ( + not from_cache + and func_name in available_functions + and output_tool is not None + ): + try: + # Use arun() instead of run() to properly await async tools + # inside a running event loop. + if hasattr(output_tool, "arun") and callable(output_tool.arun): + raw_result = await output_tool.arun(**(args_dict or {})) + else: + raw_result = available_functions[func_name](**(args_dict or {})) + raw_tool_result = raw_result + + if self.tools_handler and self.tools_handler.cache: + should_cache = True + if ( + original_tool + and hasattr(original_tool, "cache_function") + and callable(original_tool.cache_function) + ): + should_cache = original_tool.cache_function( + args_dict or {}, raw_result + ) + if should_cache: + self.tools_handler.cache.add( + tool=func_name, input=input_str, output=raw_result + ) + + result = format_native_tool_output_for_agent(output_tool, raw_result) + except Exception as e: + result = f"Error executing tool: {e}" + raw_tool_result = result + if self.task: + self.task.increment_tools_errors() + crewai_event_bus.emit( + self, + event=ToolUsageErrorEvent( + tool_name=func_name, + tool_args=args_dict, + from_agent=self.agent, + from_task=self.task, + agent_key=agent_key, + error=e, + ), + ) + error_event_emitted = True + + after_hook_context = ToolCallHookContext( + tool_name=func_name, + tool_input=args_dict or {}, + tool=structured_tool, + agent=self.agent, + task=self.task, + crew=self.crew, + tool_result=result, + raw_tool_result=raw_tool_result, + ) + modified_result = run_after_tool_call_hooks(after_hook_context) + if modified_result is not None: + result = modified_result + + if not error_event_emitted: + crewai_event_bus.emit( + self, + event=ToolUsageFinishedEvent( + output=result, + tool_name=func_name, + tool_args=args_dict, + from_agent=self.agent, + from_task=self.task, + agent_key=agent_key, + started_at=started_at, + finished_at=datetime.now(), + ), + ) + + return { + "call_id": call_id, + "func_name": func_name, + "result": result, + "from_cache": from_cache, + "original_tool": original_tool, + } + def _append_tool_result_and_check_finality( self, execution_result: dict[str, Any] ) -> AgentFinish | None: @@ -1351,7 +1679,7 @@ async def _ainvoke_loop_native_tools(self) -> AgentFinish: and answer and self._is_tool_call_list(answer) ): - tool_finish = self._handle_native_tool_calls( + tool_finish = await self._ahandle_native_tool_calls( answer, available_functions ) if tool_finish is not None: From 13c6265f690b4a853f230979768c6016306c7cd4 Mon Sep 17 00:00:00 2001 From: Shakti Prasad Mohapatra Date: Sun, 23 Aug 2026 00:30:24 +0530 Subject: [PATCH 2/5] refactor: extract shared logic from sync/async native tool call methods CodeRabbit review feedback: _aexecute_single_native_tool_call was a near-verbatim copy of _execute_single_native_tool_call (parsing, tool resolution, usage limits, cache, events, hooks) with only the tool execution line differing. Extract the shared body into _execute_single_native_tool_call_impl, an async method that takes a tool_runner callable. The sync wrapper passes a sync runner and calls it via asyncio.run(); the async wrapper passes an async runner that uses tool.arun() and awaits it directly. Also removes the redundant function-local `import asyncio` from _ahandle_native_tool_calls (asyncio is already imported at module scope on line 10). --- .../src/crewai/agents/crew_agent_executor.py | 229 ++++-------------- 1 file changed, 51 insertions(+), 178 deletions(-) diff --git a/lib/crewai/src/crewai/agents/crew_agent_executor.py b/lib/crewai/src/crewai/agents/crew_agent_executor.py index 070ea8056f..de71367a46 100644 --- a/lib/crewai/src/crewai/agents/crew_agent_executor.py +++ b/lib/crewai/src/crewai/agents/crew_agent_executor.py @@ -897,8 +897,6 @@ async def _ahandle_native_tool_calls( # Use asyncio.gather instead of ThreadPoolExecutor so async tools # are properly awaited rather than run through asyncio.run(). - import asyncio - async def _run_one( call_id: str, func_name: str, @@ -1029,178 +1027,22 @@ def _execute_single_native_tool_call( original_tool: Any | None = None, should_execute: bool = True, ) -> dict[str, Any]: - from datetime import datetime - import json - - from crewai.events.types.tool_usage_events import ( - ToolUsageErrorEvent, - ToolUsageFinishedEvent, - ToolUsageStartedEvent, - ) - - args_dict, parse_error = parse_tool_call_args( - func_args, func_name, call_id, original_tool - ) - if parse_error is not None: - return parse_error - - if original_tool is None: - for tool in self.original_tools or []: - if sanitize_tool_name(tool.name) == func_name: - original_tool = tool - break - - max_usage_reached = False - if not should_execute and original_tool: - max_usage_reached = True - elif ( - should_execute - and original_tool - and (max_count := getattr(original_tool, "max_usage_count", None)) - is not None - and getattr(original_tool, "current_usage_count", 0) >= max_count - ): - max_usage_reached = True - - structured_tool: CrewStructuredTool | None = None - if original_tool is not None: - for structured in self.tools or []: - if getattr(structured, "_original_tool", None) is original_tool: - structured_tool = structured - break - if structured_tool is None: - for structured in self.tools or []: - if sanitize_tool_name(structured.name) == func_name: - structured_tool = structured - break - - output_tool = original_tool or structured_tool - - from_cache = False - result: str = "Tool not found" - raw_tool_result: Any = result - input_str = json.dumps(args_dict) if args_dict else "" - if self.tools_handler and self.tools_handler.cache and output_tool is not None: - cached_result = self.tools_handler.cache.read( - tool=func_name, input=input_str + """Execute a single native tool call synchronously.""" + + def sync_tool_runner(tool: Any, kwargs: dict[str, Any]) -> Any: + return available_functions[func_name](**kwargs) + + return asyncio.run( + self._execute_single_native_tool_call_impl( + call_id=call_id, + func_name=func_name, + func_args=func_args, + available_functions=available_functions, + original_tool=original_tool, + should_execute=should_execute, + tool_runner=sync_tool_runner, ) - if cached_result is not None: - raw_tool_result = cached_result - result = format_native_tool_output_for_agent(output_tool, cached_result) - from_cache = True - - agent_key = getattr(self.agent, "key", "unknown") if self.agent else "unknown" - started_at = datetime.now() - crewai_event_bus.emit( - self, - event=ToolUsageStartedEvent( - tool_name=func_name, - tool_args=args_dict, - from_agent=self.agent, - from_task=self.task, - agent_key=agent_key, - ), ) - error_event_emitted = False - - track_delegation_if_needed(func_name, args_dict or {}, self.task) - - before_hook_context = ToolCallHookContext( - tool_name=func_name, - tool_input=args_dict or {}, - tool=structured_tool, - agent=self.agent, - task=self.task, - crew=self.crew, - ) - hook_blocked = run_before_tool_call_hooks(before_hook_context) - - if hook_blocked: - result = f"Tool execution blocked by hook. Tool: {func_name}" - raw_tool_result = result - elif max_usage_reached and original_tool: - result = f"Tool '{func_name}' has reached its usage limit of {original_tool.max_usage_count} times and cannot be used anymore." - raw_tool_result = result - elif ( - not from_cache - and func_name in available_functions - and output_tool is not None - ): - try: - raw_result = available_functions[func_name](**(args_dict or {})) - raw_tool_result = raw_result - - if self.tools_handler and self.tools_handler.cache: - should_cache = True - if ( - original_tool - and hasattr(original_tool, "cache_function") - and callable(original_tool.cache_function) - ): - should_cache = original_tool.cache_function( - args_dict or {}, raw_result - ) - if should_cache: - self.tools_handler.cache.add( - tool=func_name, input=input_str, output=raw_result - ) - - result = format_native_tool_output_for_agent(output_tool, raw_result) - except Exception as e: - result = f"Error executing tool: {e}" - raw_tool_result = result - if self.task: - self.task.increment_tools_errors() - crewai_event_bus.emit( - self, - event=ToolUsageErrorEvent( - tool_name=func_name, - tool_args=args_dict, - from_agent=self.agent, - from_task=self.task, - agent_key=agent_key, - error=e, - ), - ) - error_event_emitted = True - - after_hook_context = ToolCallHookContext( - tool_name=func_name, - tool_input=args_dict or {}, - tool=structured_tool, - agent=self.agent, - task=self.task, - crew=self.crew, - tool_result=result, - raw_tool_result=raw_tool_result, - ) - modified_result = run_after_tool_call_hooks(after_hook_context) - if modified_result is not None: - result = modified_result - - if not error_event_emitted: - crewai_event_bus.emit( - self, - event=ToolUsageFinishedEvent( - output=result, - tool_name=func_name, - tool_args=args_dict, - from_agent=self.agent, - from_task=self.task, - agent_key=agent_key, - started_at=started_at, - finished_at=datetime.now(), - from_cache=from_cache, - ), - ) - - return { - "call_id": call_id, - "func_name": func_name, - "result": result, - "from_cache": from_cache, - "original_tool": original_tool, - } async def _aexecute_single_native_tool_call( self, @@ -1219,6 +1061,39 @@ async def _aexecute_single_native_tool_call( ``RuntimeError: asyncio.run() cannot be called from a running event loop`` crash. """ + + async def async_tool_runner(tool: Any, kwargs: dict[str, Any]) -> Any: + if hasattr(tool, "arun") and callable(tool.arun): + return await tool.arun(**kwargs) + return available_functions[func_name](**kwargs) + + return await self._execute_single_native_tool_call_impl( + call_id=call_id, + func_name=func_name, + func_args=func_args, + available_functions=available_functions, + original_tool=original_tool, + should_execute=should_execute, + tool_runner=async_tool_runner, + ) + + async def _execute_single_native_tool_call_impl( + self, + *, + call_id: str, + func_name: str, + func_args: str | dict[str, Any], + available_functions: dict[str, Callable[..., Any]], + original_tool: Any | None, + should_execute: bool, + tool_runner: Callable[..., Any], + ) -> dict[str, Any]: + """Shared body for sync and async single-tool execution. + + ``tool_runner`` is a callable ``(tool, kwargs) -> result`` that + executes the tool — sync for the sync path, async for the async path. + When ``tool_runner`` returns an awaitable, it is awaited. + """ from datetime import datetime import json @@ -1317,12 +1192,9 @@ async def _aexecute_single_native_tool_call( and output_tool is not None ): try: - # Use arun() instead of run() to properly await async tools - # inside a running event loop. - if hasattr(output_tool, "arun") and callable(output_tool.arun): - raw_result = await output_tool.arun(**(args_dict or {})) - else: - raw_result = available_functions[func_name](**(args_dict or {})) + raw_result = tool_runner(output_tool, args_dict or {}) + if asyncio.iscoroutine(raw_result): + raw_result = await raw_result raw_tool_result = raw_result if self.tools_handler and self.tools_handler.cache: @@ -1385,6 +1257,7 @@ async def _aexecute_single_native_tool_call( agent_key=agent_key, started_at=started_at, finished_at=datetime.now(), + from_cache=from_cache, ), ) From c8bd15b0b7e399ec6706106ea42ccd35381a5454 Mon Sep 17 00:00:00 2001 From: Shakti Prasad Mohapatra Date: Sat, 26 Sep 2026 22:02:13 +0530 Subject: [PATCH 3/5] fix: keep sync native tool execution out of event loops and await only true async tools Two regressions flagged in review are fixed: - The async native path awaited `tool.arun()` whenever callable, but sync-only tools inherit `BaseTool._arun` which raises `NotImplementedError`, so async agents could not run ordinary sync tools. Async execution is now gated on `_arun` actually being overridden (or the wrapped function being a coroutine function), and sync-only tools are offloaded with `asyncio.to_thread`. - The sync native path wrapped shared logic in `asyncio.run()`, which broke tools that drive their own event loop (MCP wrappers call `asyncio.run` inside `_run`). The sync path now executes the tool inline and shares only parsing, cache, hook and event handling with the async path via extracted helpers. Adds regression tests for both cases. --- .../src/crewai/agents/crew_agent_executor.py | 462 ++++++++++++++---- .../tests/agents/test_native_tool_calling.py | 129 +++++ 2 files changed, 492 insertions(+), 99 deletions(-) diff --git a/lib/crewai/src/crewai/agents/crew_agent_executor.py b/lib/crewai/src/crewai/agents/crew_agent_executor.py index de71367a46..006f572567 100644 --- a/lib/crewai/src/crewai/agents/crew_agent_executor.py +++ b/lib/crewai/src/crewai/agents/crew_agent_executor.py @@ -1027,23 +1027,44 @@ def _execute_single_native_tool_call( original_tool: Any | None = None, should_execute: bool = True, ) -> dict[str, Any]: - """Execute a single native tool call synchronously.""" + """Execute a single native tool call synchronously. + + Stays fully synchronous: tools such as MCP wrappers drive their own + event loops via ``asyncio.run``, which fails inside an outer loop. The + sync tool runner is invoked inline and only parsing and result + processing are shared with the async path. + """ def sync_tool_runner(tool: Any, kwargs: dict[str, Any]) -> Any: return available_functions[func_name](**kwargs) - return asyncio.run( - self._execute_single_native_tool_call_impl( - call_id=call_id, - func_name=func_name, - func_args=func_args, - available_functions=available_functions, - original_tool=original_tool, - should_execute=should_execute, - tool_runner=sync_tool_runner, - ) + return self._run_single_native_tool_call( + call_id=call_id, + func_name=func_name, + func_args=func_args, + available_functions=available_functions, + original_tool=original_tool, + should_execute=should_execute, + tool_runner=sync_tool_runner, ) + def _tool_supports_native_async(self, tool: Any) -> bool: + """Check whether a tool can execute natively in async mode. + + ``BaseTool.arun`` is callable by inheritance even when only ``_run`` + is implemented; awaiting it then raises ``NotImplementedError``. A + tool supports native async when its ``_arun`` is overridden or its + wrapped function returns awaitables. + """ + from crewai.tools.base_tool import BaseTool + + if isinstance(tool, BaseTool): + if type(tool)._arun is not BaseTool._arun: + return True + wrapped_func = getattr(tool, "func", None) + return inspect.iscoroutinefunction(wrapped_func) + return False + async def _aexecute_single_native_tool_call( self, *, @@ -1056,16 +1077,16 @@ async def _aexecute_single_native_tool_call( ) -> dict[str, Any]: """Async variant of ``_execute_single_native_tool_call``. - Uses ``await tool.arun()`` instead of ``tool.run()`` so that async tools - are properly awaited inside a running event loop, avoiding the - ``RuntimeError: asyncio.run() cannot be called from a running event loop`` - crash. + Awaits tools that implement ``_arun`` and offloads sync-only tools to + a worker thread, so neither nested event loops (``asyncio.run`` inside + a running loop) nor ``NotImplementedError`` from the inherited + ``_arun`` can break execution. """ async def async_tool_runner(tool: Any, kwargs: dict[str, Any]) -> Any: - if hasattr(tool, "arun") and callable(tool.arun): + if self._tool_supports_native_async(tool): return await tool.arun(**kwargs) - return available_functions[func_name](**kwargs) + return await asyncio.to_thread(available_functions[func_name], **kwargs) return await self._execute_single_native_tool_call_impl( call_id=call_id, @@ -1077,69 +1098,99 @@ async def async_tool_runner(tool: Any, kwargs: dict[str, Any]) -> Any: tool_runner=async_tool_runner, ) - async def _execute_single_native_tool_call_impl( + def _prepare_single_native_tool_call( self, *, - call_id: str, func_name: str, func_args: str | dict[str, Any], - available_functions: dict[str, Callable[..., Any]], + call_id: str, original_tool: Any | None, should_execute: bool, - tool_runner: Callable[..., Any], - ) -> dict[str, Any]: - """Shared body for sync and async single-tool execution. + ) -> tuple[dict[str, Any] | None, dict[str, Any] | None, Any | None]: + """Resolve and validate a native tool call before execution. - ``tool_runner`` is a callable ``(tool, kwargs) -> result`` that - executes the tool — sync for the sync path, async for the async path. - When ``tool_runner`` returns an awaitable, it is awaited. + Returns ``(args_dict, error_result, original_tool)``; ``error_result`` + is a ready-to-return dict when arguments cannot be parsed or usage + limits are reached, and ``original_tool`` is resolved from the + executor's tools when missing. """ - from datetime import datetime - import json - - from crewai.events.types.tool_usage_events import ( - ToolUsageErrorEvent, - ToolUsageFinishedEvent, - ToolUsageStartedEvent, - ) - args_dict, parse_error = parse_tool_call_args( func_args, func_name, call_id, original_tool ) if parse_error is not None: - return parse_error + return None, parse_error, original_tool - if original_tool is None: - for tool in self.original_tools or []: - if sanitize_tool_name(tool.name) == func_name: - original_tool = tool - break + original_tool = self._resolve_native_tool(func_name, original_tool) max_usage_reached = False if not should_execute and original_tool: max_usage_reached = True elif ( - should_execute - and original_tool + original_tool and (max_count := getattr(original_tool, "max_usage_count", None)) is not None and getattr(original_tool, "current_usage_count", 0) >= max_count ): max_usage_reached = True - structured_tool: CrewStructuredTool | None = None + if max_usage_reached and original_tool: + return ( + args_dict, + { + "call_id": call_id, + "func_name": func_name, + "result": ( + f"Tool '{func_name}' has reached its usage limit of " + f"{original_tool.max_usage_count} times and cannot be used anymore." + ), + "from_cache": False, + "original_tool": original_tool, + }, + original_tool, + ) + return args_dict, None, original_tool + + def _resolve_native_tool( + self, + func_name: str, + original_tool: Any | None, + ) -> Any | None: + """Fall back to a name match against the original tools if needed.""" + if original_tool is not None: + return original_tool + for tool in self.original_tools or []: + if sanitize_tool_name(tool.name) == func_name: + return tool + return None + + def _resolve_structured_tool( + self, + func_name: str, + original_tool: Any | None, + ) -> CrewStructuredTool | None: + """Find the structured tool matching the original tool or name.""" if original_tool is not None: for structured in self.tools or []: if getattr(structured, "_original_tool", None) is original_tool: - structured_tool = structured - break - if structured_tool is None: - for structured in self.tools or []: - if sanitize_tool_name(structured.name) == func_name: - structured_tool = structured - break + return structured + for structured in self.tools or []: + if sanitize_tool_name(structured.name) == func_name: + return structured + return None - output_tool = original_tool or structured_tool + def _read_native_tool_cache( + self, + *, + func_name: str, + args_dict: dict[str, Any], + output_tool: Any, + ) -> tuple[bool, str, Any]: + """Read the tool cache. + + Returns ``(from_cache, result, raw_tool_result)``; ``result`` stays + ``"Tool not found"`` when nothing is cached. + """ + import json from_cache = False result: str = "Tool not found" @@ -1153,6 +1204,130 @@ async def _execute_single_native_tool_call_impl( raw_tool_result = cached_result result = format_native_tool_output_for_agent(output_tool, cached_result) from_cache = True + return from_cache, result, raw_tool_result + + def _finalize_single_native_tool_call( + self, + *, + call_id: str, + func_name: str, + args_dict: dict[str, Any], + original_tool: Any | None, + structured_tool: CrewStructuredTool | None, + raw_tool_result: Any, + result: str, + from_cache: bool, + started_at: Any, + agent_key: str, + error_event_emitted: bool, + ) -> dict[str, Any]: + """Run after-hooks, emit the finished event and build the result dict.""" + from datetime import datetime + + from crewai.events.types.tool_usage_events import ToolUsageFinishedEvent + + after_hook_context = ToolCallHookContext( + tool_name=func_name, + tool_input=args_dict or {}, + tool=structured_tool, + agent=self.agent, + task=self.task, + crew=self.crew, + tool_result=result, + raw_tool_result=raw_tool_result, + ) + modified_result = run_after_tool_call_hooks(after_hook_context) + if modified_result is not None: + result = modified_result + + if not error_event_emitted: + crewai_event_bus.emit( + self, + event=ToolUsageFinishedEvent( + output=result, + tool_name=func_name, + tool_args=args_dict, + from_agent=self.agent, + from_task=self.task, + agent_key=agent_key, + started_at=started_at, + finished_at=datetime.now(), + from_cache=from_cache, + ), + ) + + return { + "call_id": call_id, + "func_name": func_name, + "result": result, + "from_cache": from_cache, + "original_tool": original_tool, + } + + def _native_tool_cache_result( + self, + *, + func_name: str, + args_dict: dict[str, Any], + input_str: str, + original_tool: Any | None, + raw_result: Any, + ) -> None: + """Cache a tool result when the tool's cache policy allows it.""" + if self.tools_handler and self.tools_handler.cache: + should_cache = True + if ( + original_tool + and hasattr(original_tool, "cache_function") + and callable(original_tool.cache_function) + ): + should_cache = original_tool.cache_function(args_dict or {}, raw_result) + if should_cache: + self.tools_handler.cache.add( + tool=func_name, input=input_str, output=raw_result + ) + + def _run_single_native_tool_call( + self, + *, + call_id: str, + func_name: str, + func_args: str | dict[str, Any], + available_functions: dict[str, Callable[..., Any]], + original_tool: Any | None, + should_execute: bool, + tool_runner: Callable[..., Any], + ) -> dict[str, Any]: + """Synchronously orchestrate a single native tool call. + + Shares parsing, cache handling, hooks and event emission with the + async path, but executes ``tool_runner`` inline so tools that drive + their own event loops (e.g. MCP wrappers) keep working. + """ + from datetime import datetime + import json + + from crewai.events.types.tool_usage_events import ( + ToolUsageErrorEvent, + ToolUsageStartedEvent, + ) + + args_dict, error_result, original_tool = self._prepare_single_native_tool_call( + func_name=func_name, + func_args=func_args, + call_id=call_id, + original_tool=original_tool, + should_execute=should_execute, + ) + if error_result is not None: + return error_result + + structured_tool = self._resolve_structured_tool(func_name, original_tool) + output_tool = original_tool or structured_tool + + from_cache, result, raw_tool_result = self._read_native_tool_cache( + func_name=func_name, args_dict=args_dict or {}, output_tool=output_tool + ) agent_key = getattr(self.agent, "key", "unknown") if self.agent else "unknown" started_at = datetime.now() @@ -1180,12 +1355,10 @@ async def _execute_single_native_tool_call_impl( ) hook_blocked = run_before_tool_call_hooks(before_hook_context) + input_str = json.dumps(args_dict) if args_dict else "" if hook_blocked: result = f"Tool execution blocked by hook. Tool: {func_name}" raw_tool_result = result - elif max_usage_reached and original_tool: - result = f"Tool '{func_name}' has reached its usage limit of {original_tool.max_usage_count} times and cannot be used anymore." - raw_tool_result = result elif ( not from_cache and func_name in available_functions @@ -1193,24 +1366,15 @@ async def _execute_single_native_tool_call_impl( ): try: raw_result = tool_runner(output_tool, args_dict or {}) - if asyncio.iscoroutine(raw_result): - raw_result = await raw_result raw_tool_result = raw_result - if self.tools_handler and self.tools_handler.cache: - should_cache = True - if ( - original_tool - and hasattr(original_tool, "cache_function") - and callable(original_tool.cache_function) - ): - should_cache = original_tool.cache_function( - args_dict or {}, raw_result - ) - if should_cache: - self.tools_handler.cache.add( - tool=func_name, input=input_str, output=raw_result - ) + self._native_tool_cache_result( + func_name=func_name, + args_dict=args_dict or {}, + input_str=input_str, + original_tool=original_tool, + raw_result=raw_result, + ) result = format_native_tool_output_for_agent(output_tool, raw_result) except Exception as e: @@ -1231,43 +1395,143 @@ async def _execute_single_native_tool_call_impl( ) error_event_emitted = True - after_hook_context = ToolCallHookContext( + return self._finalize_single_native_tool_call( + call_id=call_id, + func_name=func_name, + args_dict=args_dict or {}, + original_tool=original_tool, + structured_tool=structured_tool, + raw_tool_result=raw_tool_result, + result=result, + from_cache=from_cache, + started_at=started_at, + agent_key=agent_key, + error_event_emitted=error_event_emitted, + ) + + async def _execute_single_native_tool_call_impl( + self, + *, + call_id: str, + func_name: str, + func_args: str | dict[str, Any], + available_functions: dict[str, Callable[..., Any]], + original_tool: Any | None, + should_execute: bool, + tool_runner: Callable[..., Any], + ) -> dict[str, Any]: + """Shared orchestration for a single async native tool call. + + Mirrors the sync flow but awaits ``tool_runner`` results, so async + tools run on the caller's event loop while everything else (parsing, + cache, hooks, events) stays shared. + """ + from datetime import datetime + import json + + from crewai.events.types.tool_usage_events import ( + ToolUsageErrorEvent, + ToolUsageStartedEvent, + ) + + args_dict, error_result, original_tool = self._prepare_single_native_tool_call( + func_name=func_name, + func_args=func_args, + call_id=call_id, + original_tool=original_tool, + should_execute=should_execute, + ) + if error_result is not None: + return error_result + + structured_tool = self._resolve_structured_tool(func_name, original_tool) + output_tool = original_tool or structured_tool + + from_cache, result, raw_tool_result = self._read_native_tool_cache( + func_name=func_name, args_dict=args_dict or {}, output_tool=output_tool + ) + + agent_key = getattr(self.agent, "key", "unknown") if self.agent else "unknown" + started_at = datetime.now() + crewai_event_bus.emit( + self, + event=ToolUsageStartedEvent( + tool_name=func_name, + tool_args=args_dict, + from_agent=self.agent, + from_task=self.task, + agent_key=agent_key, + ), + ) + error_event_emitted = False + + track_delegation_if_needed(func_name, args_dict or {}, self.task) + + before_hook_context = ToolCallHookContext( tool_name=func_name, tool_input=args_dict or {}, tool=structured_tool, agent=self.agent, task=self.task, crew=self.crew, - tool_result=result, - raw_tool_result=raw_tool_result, ) - modified_result = run_after_tool_call_hooks(after_hook_context) - if modified_result is not None: - result = modified_result + hook_blocked = run_before_tool_call_hooks(before_hook_context) - if not error_event_emitted: - crewai_event_bus.emit( - self, - event=ToolUsageFinishedEvent( - output=result, - tool_name=func_name, - tool_args=args_dict, - from_agent=self.agent, - from_task=self.task, - agent_key=agent_key, - started_at=started_at, - finished_at=datetime.now(), - from_cache=from_cache, - ), - ) + input_str = json.dumps(args_dict) if args_dict else "" + if hook_blocked: + result = f"Tool execution blocked by hook. Tool: {func_name}" + raw_tool_result = result + elif ( + not from_cache + and func_name in available_functions + and output_tool is not None + ): + try: + raw_result = tool_runner(output_tool, args_dict or {}) + if asyncio.iscoroutine(raw_result): + raw_result = await raw_result + raw_tool_result = raw_result - return { - "call_id": call_id, - "func_name": func_name, - "result": result, - "from_cache": from_cache, - "original_tool": original_tool, - } + self._native_tool_cache_result( + func_name=func_name, + args_dict=args_dict or {}, + input_str=input_str, + original_tool=original_tool, + raw_result=raw_result, + ) + + result = format_native_tool_output_for_agent(output_tool, raw_result) + except Exception as e: + result = f"Error executing tool: {e}" + raw_tool_result = result + if self.task: + self.task.increment_tools_errors() + crewai_event_bus.emit( + self, + event=ToolUsageErrorEvent( + tool_name=func_name, + tool_args=args_dict, + from_agent=self.agent, + from_task=self.task, + agent_key=agent_key, + error=e, + ), + ) + error_event_emitted = True + + return self._finalize_single_native_tool_call( + call_id=call_id, + func_name=func_name, + args_dict=args_dict or {}, + original_tool=original_tool, + structured_tool=structured_tool, + raw_tool_result=raw_tool_result, + result=result, + from_cache=from_cache, + started_at=started_at, + agent_key=agent_key, + error_event_emitted=error_event_emitted, + ) def _append_tool_result_and_check_finality( self, execution_result: dict[str, Any] diff --git a/lib/crewai/tests/agents/test_native_tool_calling.py b/lib/crewai/tests/agents/test_native_tool_calling.py index 894c0bd45e..b8be1f0e31 100644 --- a/lib/crewai/tests/agents/test_native_tool_calling.py +++ b/lib/crewai/tests/agents/test_native_tool_calling.py @@ -6,6 +6,7 @@ from __future__ import annotations +import asyncio from collections.abc import Generator import json import os @@ -1363,3 +1364,131 @@ def _run(self, code: str, language: str) -> str: assert "Error" in result["result"] assert "validation failed" in result["result"].lower() or "missing" in result["result"].lower() + + +# Async Native Tool Execution Tests + + +class TestAsyncNativeToolExecution: + """Regression tests for sync/async native tool execution. + + Guards against two regressions: awaiting the inherited ``BaseTool._arun`` + (which raises ``NotImplementedError``) for sync-only tools on the async + path, and wrapping sync tool execution in an outer event loop that + crashes tools driving their own loops (e.g. MCP wrappers). + """ + + def _make_executor(self, tools: list[BaseTool]) -> "CrewAgentExecutor": + """Create a minimal CrewAgentExecutor with mocked dependencies.""" + from crewai.agents.crew_agent_executor import CrewAgentExecutor + from crewai.tools.base_tool import to_langchain + + structured_tools = to_langchain(tools) + mock_agent = Mock() + mock_agent.key = "test_agent" + mock_agent.role = "tester" + mock_agent.verbose = False + mock_agent.fingerprint = None + mock_agent.tools_results = [] + + mock_task = Mock() + mock_task.name = "test" + mock_task.description = "test" + mock_task.id = "test-id" + + executor = CrewAgentExecutor( + tools=structured_tools, + original_tools=tools, + ) + executor.agent = mock_agent + executor.task = mock_task + return executor + + @pytest.mark.asyncio + async def test_async_path_runs_sync_only_tool(self) -> None: + """Sync-only tools (no _arun) must work on the async native path.""" + + class SyncTool(BaseTool): + name: str = "sync_tool" + description: str = "A sync-only tool" + + def _run(self, value: str) -> str: + return f"sync: {value}" + + tool = SyncTool() + executor = self._make_executor([tool]) + + from crewai.utilities.agent_utils import convert_tools_to_openai_schema + + _, available_functions, _ = convert_tools_to_openai_schema([tool]) + + result = await executor._aexecute_single_native_tool_call( + call_id="call_async_sync", + func_name="sync_tool", + func_args={"value": "hello"}, + available_functions=available_functions, + ) + + assert result["result"] == "sync: hello" + assert "Error" not in result["result"] + + @pytest.mark.asyncio + async def test_async_path_awaits_async_tool(self) -> None: + """Tools overriding _arun must be awaited, not run through a thread.""" + + class AsyncTool(BaseTool): + name: str = "async_tool" + description: str = "An async tool" + + def _run(self, value: str) -> str: + return "wrong path" + + async def _arun(self, value: str) -> str: + return f"async: {value}" + + tool = AsyncTool() + executor = self._make_executor([tool]) + + from crewai.utilities.agent_utils import convert_tools_to_openai_schema + + _, available_functions, _ = convert_tools_to_openai_schema([tool]) + + result = await executor._aexecute_single_native_tool_call( + call_id="call_async_async", + func_name="async_tool", + func_args={"value": "hello"}, + available_functions=available_functions, + ) + + assert result["result"] == "async: hello" + + def test_sync_path_allows_internal_event_loop(self) -> None: + """The sync path must not create an outer event loop around tools + that drive their own loop via asyncio.run (like MCP wrappers).""" + + class SelfLoopTool(BaseTool): + name: str = "self_loop_tool" + description: str = "A tool that runs its own event loop" + + def _run(self, value: str) -> str: + async def _inner() -> str: + return f"loop: {value}" + + return asyncio.run(_inner()) + + tool = SelfLoopTool() + executor = self._make_executor([tool]) + + from crewai.utilities.agent_utils import convert_tools_to_openai_schema + + _, available_functions, _ = convert_tools_to_openai_schema([tool]) + + result = executor._execute_single_native_tool_call( + call_id="call_sync_loop", + func_name="self_loop_tool", + func_args={"value": "hello"}, + available_functions=available_functions, + ) + + assert result["result"] == "loop: hello" + assert "Error" not in result["result"] From c7c3add86e2c3fd3be567f0986ebb62c1c7d12c8 Mon Sep 17 00:00:00 2001 From: Shakti Prasad Mohapatra Date: Sat, 3 Oct 2026 14:18:58 +0530 Subject: [PATCH 4/5] fix(agents): classify Tool-wrapped sync functions as non-async and keep events on the usage-limit path Two Cursor Bugbot findings on the async native tool path: 1. _tool_supports_native_async treated every Tool as natively async because Tool always overrides _arun (BaseTool._arun is abstract). Sync @tool functions then hit 'NotImplementedError' from Tool._arun. Classify by qualname: Tool._arun demotes to the wrapped-function probe (sync -> worker thread); other overrides are genuinely async. The async runner also falls back to the worker thread on NotImplementedError so awaitable-returning sync wrappers cannot break execution. 2. _prepare_single_native_tool_call returned the usage-limit error dict immediately, skipping ToolUsageStartedEvent, before/after hooks and ToolUsageFinishedEvent that still ran on both native paths pre-refactor. It now returns a max_usage_reached flag and the shared orchestrators render the limit message as a third branch after hooks, restoring the event/hook contract for both sync and async paths. --- .../src/crewai/agents/crew_agent_executor.py | 77 +++++++++++-------- 1 file changed, 45 insertions(+), 32 deletions(-) diff --git a/lib/crewai/src/crewai/agents/crew_agent_executor.py b/lib/crewai/src/crewai/agents/crew_agent_executor.py index 006f572567..82e5ef6e3c 100644 --- a/lib/crewai/src/crewai/agents/crew_agent_executor.py +++ b/lib/crewai/src/crewai/agents/crew_agent_executor.py @@ -1053,16 +1053,33 @@ def _tool_supports_native_async(self, tool: Any) -> bool: ``BaseTool.arun`` is callable by inheritance even when only ``_run`` is implemented; awaiting it then raises ``NotImplementedError``. A - tool supports native async when its ``_arun`` is overridden or its - wrapped function returns awaitables. + tool supports native async when its own ``_arun`` can run without + raising, i.e. when the wrapped function is async or returns + awaitables. A class-level ``type(tool)._arun is not BaseTool._arun`` + check is not sufficient here: ``Tool`` always overrides ``_arun`` + (it dispatches to the wrapped function and raises + ``NotImplementedError`` for sync-only wrapped functions), so sync + ``@tool`` functions must be classified by their wrapped function. """ from crewai.tools.base_tool import BaseTool if isinstance(tool, BaseTool): - if type(tool)._arun is not BaseTool._arun: - return True + cls_arun = getattr(type(tool), "_arun", None) + if cls_arun is not None and cls_arun is not BaseTool._arun: + cls_qname = getattr(cls_arun, "__qualname__", "") + # ``Tool._arun`` demotes to the wrapped-function probes below; + # any other override (a subclass with real async support) is + # natively async by definition. + if cls_qname not in ("Tool._arun", "BaseTool._arun"): + return True wrapped_func = getattr(tool, "func", None) - return inspect.iscoroutinefunction(wrapped_func) + if inspect.iscoroutinefunction(wrapped_func): + return True + # Sync-wrapped ``Tool``: classify as async only when the wrapped + # function returns awaitables, so ``Tool._arun`` dispatch would + # succeed; the runner still recovers via NotImplementedError if + # this proves wrong at call time. + return False return False async def _aexecute_single_native_tool_call( @@ -1085,7 +1102,13 @@ async def _aexecute_single_native_tool_call( async def async_tool_runner(tool: Any, kwargs: dict[str, Any]) -> Any: if self._tool_supports_native_async(tool): - return await tool.arun(**kwargs) + try: + return await tool.arun(**kwargs) + except NotImplementedError: + # ``Tool._arun`` demotes to ``NotImplementedError`` for + # sync-only wrapped functions; fall back to the worker + # thread so sync ``@tool`` tools still execute. + pass return await asyncio.to_thread(available_functions[func_name], **kwargs) return await self._execute_single_native_tool_call_impl( @@ -1109,10 +1132,13 @@ def _prepare_single_native_tool_call( ) -> tuple[dict[str, Any] | None, dict[str, Any] | None, Any | None]: """Resolve and validate a native tool call before execution. - Returns ``(args_dict, error_result, original_tool)``; ``error_result`` - is a ready-to-return dict when arguments cannot be parsed or usage - limits are reached, and ``original_tool`` is resolved from the - executor's tools when missing. + Returns ``(args_dict, max_usage_reached, original_tool)``; + ``max_usage_reached`` is ``True`` when the call must not execute the + tool body (arguments used verbatim, limit message rendered by the + shared orchestration below — which still emits started/finished + events and runs before/after hooks, matching the pre-refactor sync + flow). ``original_tool`` is resolved from the executor's tools when + missing. """ args_dict, parse_error = parse_tool_call_args( func_args, func_name, call_id, original_tool @@ -1133,22 +1159,7 @@ def _prepare_single_native_tool_call( ): max_usage_reached = True - if max_usage_reached and original_tool: - return ( - args_dict, - { - "call_id": call_id, - "func_name": func_name, - "result": ( - f"Tool '{func_name}' has reached its usage limit of " - f"{original_tool.max_usage_count} times and cannot be used anymore." - ), - "from_cache": False, - "original_tool": original_tool, - }, - original_tool, - ) - return args_dict, None, original_tool + return args_dict, (max_usage_reached if original_tool else False), original_tool def _resolve_native_tool( self, @@ -1312,15 +1323,13 @@ def _run_single_native_tool_call( ToolUsageStartedEvent, ) - args_dict, error_result, original_tool = self._prepare_single_native_tool_call( + args_dict, max_usage_reached, original_tool = self._prepare_single_native_tool_call( func_name=func_name, func_args=func_args, call_id=call_id, original_tool=original_tool, should_execute=should_execute, ) - if error_result is not None: - return error_result structured_tool = self._resolve_structured_tool(func_name, original_tool) output_tool = original_tool or structured_tool @@ -1359,6 +1368,9 @@ def _run_single_native_tool_call( if hook_blocked: result = f"Tool execution blocked by hook. Tool: {func_name}" raw_tool_result = result + elif max_usage_reached and original_tool: + result = f"Tool '{func_name}' has reached its usage limit of {original_tool.max_usage_count} times and cannot be used anymore." + raw_tool_result = result elif ( not from_cache and func_name in available_functions @@ -1434,15 +1446,13 @@ async def _execute_single_native_tool_call_impl( ToolUsageStartedEvent, ) - args_dict, error_result, original_tool = self._prepare_single_native_tool_call( + args_dict, max_usage_reached, original_tool = self._prepare_single_native_tool_call( func_name=func_name, func_args=func_args, call_id=call_id, original_tool=original_tool, should_execute=should_execute, ) - if error_result is not None: - return error_result structured_tool = self._resolve_structured_tool(func_name, original_tool) output_tool = original_tool or structured_tool @@ -1481,6 +1491,9 @@ async def _execute_single_native_tool_call_impl( if hook_blocked: result = f"Tool execution blocked by hook. Tool: {func_name}" raw_tool_result = result + elif max_usage_reached and original_tool: + result = f"Tool '{func_name}' has reached its usage limit of {original_tool.max_usage_count} times and cannot be used anymore." + raw_tool_result = result elif ( not from_cache and func_name in available_functions From 1bd28da67e9072274917f5a390d09a1c98cc247d Mon Sep 17 00:00:00 2001 From: Shakti Prasad Mohapatra Date: Sat, 3 Oct 2026 14:28:17 +0530 Subject: [PATCH 5/5] fix(agents): separate parse-error from usage-limit and drop the arun fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two findings from CodeRabbit + Cursor Bugbot on the async native path: 1. Malformed JSON arguments were misrouted: parse_tool_call_args returns its ready-to-emit INVALID_INPUT dict, but _prepare_single_native_tool_call packed it into the max_usage_reached slot, so it surfaced as 'reached its usage limit of None' (when original_tool was set) or the tool ran with empty arguments (when not). The prepare method now returns a 4-tuple (args_dict, error_result, max_usage_reached, original_tool); parse errors short-circuit with error_result verbatim, preserving the json_decode_error code and INVALID_INPUT failure shape. 2. The NotImplementedError fallback in the runner awaited arun() first, then fell through to tool.run — BaseTool.arun/_arun claim usage before the wrapped function runs, so a genuine NotImplementedError from inside a tool body (or an explicit stub) made the tool's side effects execute twice and counted one call twice toward max_usage_count. _tool_supports_native_async already returns False for sync-wrapped Tools, so the wrapped functions reach the worker-thread branch directly; the try/except is removed and real tool errors reach the shared error handler. Verified: sync @tool on the async path executes once (usage claimed exactly once), async @tools are awaited, malformed JSON returns INVALID_INPUT with no events, the usage-limit path still emits Started+Finished on both sync and async paths, and ruff is clean. --- .../src/crewai/agents/crew_agent_executor.py | 63 ++++++++++--------- 1 file changed, 35 insertions(+), 28 deletions(-) diff --git a/lib/crewai/src/crewai/agents/crew_agent_executor.py b/lib/crewai/src/crewai/agents/crew_agent_executor.py index 82e5ef6e3c..7435c5247c 100644 --- a/lib/crewai/src/crewai/agents/crew_agent_executor.py +++ b/lib/crewai/src/crewai/agents/crew_agent_executor.py @@ -1102,13 +1102,11 @@ async def _aexecute_single_native_tool_call( async def async_tool_runner(tool: Any, kwargs: dict[str, Any]) -> Any: if self._tool_supports_native_async(tool): - try: - return await tool.arun(**kwargs) - except NotImplementedError: - # ``Tool._arun`` demotes to ``NotImplementedError`` for - # sync-only wrapped functions; fall back to the worker - # thread so sync ``@tool`` tools still execute. - pass + return await tool.arun(**kwargs) + # Sync-wrapped Tools must go through ``tool.run`` (which claims + # usage once); an arun fallback would claim usage twice and rerun + # tool side effects after a genuine NotImplementedError from the + # tool body. return await asyncio.to_thread(available_functions[func_name], **kwargs) return await self._execute_single_native_tool_call_impl( @@ -1129,22 +1127,23 @@ def _prepare_single_native_tool_call( call_id: str, original_tool: Any | None, should_execute: bool, - ) -> tuple[dict[str, Any] | None, dict[str, Any] | None, Any | None]: + ) -> tuple[dict[str, Any] | None, dict[str, Any] | None, bool, Any | None]: """Resolve and validate a native tool call before execution. - Returns ``(args_dict, max_usage_reached, original_tool)``; - ``max_usage_reached`` is ``True`` when the call must not execute the - tool body (arguments used verbatim, limit message rendered by the + Returns ``(args_dict, error_result, max_usage_reached, + original_tool)``; ``error_result`` is a ready-to-return dict when the + arguments cannot be parsed and must surface verbatim (carrying the + ``INVALID_INPUT`` failure). ``max_usage_reached`` is ``True`` when the + call must not execute the tool body (limit message rendered by the shared orchestration below — which still emits started/finished - events and runs before/after hooks, matching the pre-refactor sync - flow). ``original_tool`` is resolved from the executor's tools when - missing. + events and runs hooks, matching the pre-refactor sync flow). + ``original_tool`` is resolved from the executor's tools when missing. """ args_dict, parse_error = parse_tool_call_args( func_args, func_name, call_id, original_tool ) if parse_error is not None: - return None, parse_error, original_tool + return None, parse_error, False, original_tool original_tool = self._resolve_native_tool(func_name, original_tool) @@ -1159,7 +1158,7 @@ def _prepare_single_native_tool_call( ): max_usage_reached = True - return args_dict, (max_usage_reached if original_tool else False), original_tool + return args_dict, None, (max_usage_reached if original_tool else False), original_tool def _resolve_native_tool( self, @@ -1323,13 +1322,17 @@ def _run_single_native_tool_call( ToolUsageStartedEvent, ) - args_dict, max_usage_reached, original_tool = self._prepare_single_native_tool_call( - func_name=func_name, - func_args=func_args, - call_id=call_id, - original_tool=original_tool, - should_execute=should_execute, + args_dict, error_result, max_usage_reached, original_tool = ( + self._prepare_single_native_tool_call( + func_name=func_name, + func_args=func_args, + call_id=call_id, + original_tool=original_tool, + should_execute=should_execute, + ) ) + if error_result is not None: + return error_result structured_tool = self._resolve_structured_tool(func_name, original_tool) output_tool = original_tool or structured_tool @@ -1446,13 +1449,17 @@ async def _execute_single_native_tool_call_impl( ToolUsageStartedEvent, ) - args_dict, max_usage_reached, original_tool = self._prepare_single_native_tool_call( - func_name=func_name, - func_args=func_args, - call_id=call_id, - original_tool=original_tool, - should_execute=should_execute, + args_dict, error_result, max_usage_reached, original_tool = ( + self._prepare_single_native_tool_call( + func_name=func_name, + func_args=func_args, + call_id=call_id, + original_tool=original_tool, + should_execute=should_execute, + ) ) + if error_result is not None: + return error_result structured_tool = self._resolve_structured_tool(func_name, original_tool) output_tool = original_tool or structured_tool