diff --git a/lib/crewai/src/crewai/agents/crew_agent_executor.py b/lib/crewai/src/crewai/agents/crew_agent_executor.py index bd7d313d35..7435c5247c 100644 --- a/lib/crewai/src/crewai/agents/crew_agent_executor.py +++ b/lib/crewai/src/crewai/agents/crew_agent_executor.py @@ -826,6 +826,138 @@ 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(). + 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: @@ -895,52 +1027,180 @@ def _execute_single_native_tool_call( original_tool: Any | None = None, should_execute: bool = True, ) -> dict[str, Any]: - from datetime import datetime - import json + """Execute a single native tool call synchronously. - from crewai.events.types.tool_usage_events import ( - ToolUsageErrorEvent, - ToolUsageFinishedEvent, - ToolUsageStartedEvent, + 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 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 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): + 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) + 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( + 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``. + + 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 self._tool_supports_native_async(tool): + 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( + 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, + ) + + def _prepare_single_native_tool_call( + self, + *, + func_name: str, + func_args: str | dict[str, Any], + call_id: str, + original_tool: Any | None, + should_execute: bool, + ) -> tuple[dict[str, Any] | None, dict[str, Any] | None, bool, Any | None]: + """Resolve and validate a native tool call before execution. + + 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 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 parse_error + return None, parse_error, False, 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 + return args_dict, None, (max_usage_reached if original_tool else False), 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" @@ -954,6 +1214,132 @@ def _execute_single_native_tool_call( 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, 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 + + 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() @@ -981,6 +1367,7 @@ def _execute_single_native_tool_call( ) 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 @@ -993,23 +1380,16 @@ def _execute_single_native_tool_call( and output_tool is not None ): try: - raw_result = available_functions[func_name](**(args_dict or {})) + raw_result = tool_runner(output_tool, 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 - ) + 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: @@ -1030,43 +1410,148 @@ def _execute_single_native_tool_call( ) 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, 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 + + 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 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 = 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] @@ -1351,7 +1836,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: 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"]