fix: #6611 - Add async native tool execution to experimental executor - #6699
fix: #6611 - Add async native tool execution to experimental executor#6699Diwak4r wants to merge 3 commits into
Conversation
…_original_tool_calling for consistent error handling
… executor - Added async _execute_single_native_tool_call_async method using CrewStructuredTool.ainvoke for proper async tool support - Changed execute_native_tool to async method using asyncio.gather for parallel execution - Sync tools still work via fallback to _available_functions in thread pool - Async tools now run on caller's event loop (MainThread) instead of via asyncio.run() in worker thread - Updated test_execute_native_tool tests to await the async method - Matches ReAct executor's async-aware dispatch pattern
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
📝 WalkthroughWalkthroughNative tool execution now uses async orchestration with awaited structured tools, concurrent and sequential paths, preserved lifecycle behavior, standardized argument errors, and updated tests. Uploader factory failure paths now raise ChangesNative tool async execution
Uploader error standardization
Sequence Diagram(s)sequenceDiagram
participant AgentExecutor
participant CrewStructuredTool
participant ToolFunction
AgentExecutor->>CrewStructuredTool: invoke native tool asynchronously
CrewStructuredTool->>ToolFunction: await async tool function
ToolFunction-->>AgentExecutor: return execution result
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
lib/crewai/tests/agents/test_agent_executor.py (1)
1243-1277: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winNo test exercises an actual async tool, and the parallelism assertion is too loose to detect serialization.
The PR's core behavior — awaiting coroutine tools natively on the running loop (issue
#6611) — is untested. Every tool here is a plaindefwithtime.sleep, which goes down the sync fallback branch.Separately,
assert elapsed < 0.5passes even when the two 0.2s calls run strictly sequentially (~0.4s), so it no longer proves concurrency; see the blocking-fallback concern atlib/crewai/src/crewai/experimental/agent_executor.pyLine 2203. Tightening to< 0.35would make the assertion meaningful.Suggest adding a case with an
async deftool asserting it is awaited on the caller's loop (e.g. captureasyncio.get_running_loop()inside the tool and compare), plus one asserting twoasyncio.sleep(0.2)tools complete in ~0.2s.As per coding guidelines, "Write unit tests for new functionality that focus on behavior rather than implementation details."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/tests/agents/test_agent_executor.py` around lines 1243 - 1277, Expand test_execute_native_tool_runs_parallel_for_multiple_calls to use async tools with asyncio.sleep and assert both complete below 0.35 seconds, proving concurrent execution rather than serialization. Add a separate async-tool test that captures asyncio.get_running_loop() inside the tool and verifies it matches the caller’s loop, while preserving result and tool-message assertions.Source: Coding guidelines
🧹 Nitpick comments (3)
lib/crewai/src/crewai/tools/tool_usage.py (1)
846-854: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winSwallowing the original exception loses argument-parse diagnostics.
_tool_calling(Line 878) prints and formats the caught exception into the agent-visibletool_usage_errormessage. Since this path now always raises the generictool_arguments_errorstring, the concrete cause (malformed JSON, repair failure, schema mismatch) is no longer visible to either the agent or the logs — that detail is what lets the LLM correct its own arguments on retry. Consider chaining the original.♻️ Preserve the cause
- except Exception: + except Exception as e: if raise_error: - raise ToolUsageError(f"{I18N_DEFAULT.errors('tool_arguments_error')}") + raise ToolUsageError( + f"{I18N_DEFAULT.errors('tool_arguments_error')}" + ) from e return ToolUsageError(f"{I18N_DEFAULT.errors('tool_arguments_error')}")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/src/crewai/tools/tool_usage.py` around lines 846 - 854, Preserve the original argument-parsing exception in the exception branch of _tool_calling by chaining it when raising or returning ToolUsageError, while retaining the existing generic tool_arguments_error message and raise_error behavior. Ensure callers can inspect the concrete cause for malformed JSON, repair, or schema failures.lib/crewai/src/crewai/experimental/agent_executor.py (2)
1736-1749: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winConsider
return_exceptions=Truefor the parallel batch.
_execute_single_native_tool_call_asynconly guards the tool invocation itself; failures in hooks, event emission, orformat_native_tool_output_for_agentpropagate and will abort the whole batch, discarding results from sibling calls that already completed.execute_todos_parallel(Line 1240) already usesreturn_exceptions=Truefor this reason.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/src/crewai/experimental/agent_executor.py` around lines 1736 - 1749, The parallel branch in the native tool execution flow should preserve sibling results when one task fails outside the tool-invocation guard. Update the asyncio.gather call in the should_parallelize path to use return_exceptions=True, then handle gathered exceptions consistently with execute_todos_parallel while retaining successful results.
2051-2058: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRefactor the duplicated native tool invocation flow.
_execute_single_native_tool_call_asynccopies the sync call path plus most hook/cache/event logic from_execute_single_native_tool_call; only the structured-tool invocation differs (await structured_tool.ainvoke(...)vstool_func(**args_dict)). Extract the shared resolution/cache/hooks into helpers so cache-key/hook-order fixes are applied in one place, and drop the private experimental sync method if it has no remaining callers.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai/src/crewai/experimental/agent_executor.py` around lines 2051 - 2058, Refactor _execute_single_native_tool_call_async and _execute_single_native_tool_call to share helpers for tool resolution, cache access, hooks, and event handling, leaving only the invocation operation different (ainvoke asynchronously versus direct synchronous execution). Preserve the existing cache-key and hook ordering in the shared flow, and remove the private synchronous method if no callers remain after consolidation.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@lib/crewai/src/crewai/experimental/agent_executor.py`:
- Around line 2203-2224: Update the synchronous fallback in the native tool
execution path identified by _available_functions and tool_func so
tool_func(**args_dict) runs via asyncio.to_thread instead of directly on the
event loop. Await the offloaded call while preserving the existing raw-result,
caching, and formatting flow for synchronous tools.
---
Outside diff comments:
In `@lib/crewai/tests/agents/test_agent_executor.py`:
- Around line 1243-1277: Expand
test_execute_native_tool_runs_parallel_for_multiple_calls to use async tools
with asyncio.sleep and assert both complete below 0.35 seconds, proving
concurrent execution rather than serialization. Add a separate async-tool test
that captures asyncio.get_running_loop() inside the tool and verifies it matches
the caller’s loop, while preserving result and tool-message assertions.
---
Nitpick comments:
In `@lib/crewai/src/crewai/experimental/agent_executor.py`:
- Around line 1736-1749: The parallel branch in the native tool execution flow
should preserve sibling results when one task fails outside the tool-invocation
guard. Update the asyncio.gather call in the should_parallelize path to use
return_exceptions=True, then handle gathered exceptions consistently with
execute_todos_parallel while retaining successful results.
- Around line 2051-2058: Refactor _execute_single_native_tool_call_async and
_execute_single_native_tool_call to share helpers for tool resolution, cache
access, hooks, and event handling, leaving only the invocation operation
different (ainvoke asynchronously versus direct synchronous execution). Preserve
the existing cache-key and hook ordering in the shared flow, and remove the
private synchronous method if no callers remain after consolidation.
In `@lib/crewai/src/crewai/tools/tool_usage.py`:
- Around line 846-854: Preserve the original argument-parsing exception in the
exception branch of _tool_calling by chaining it when raising or returning
ToolUsageError, while retaining the existing generic tool_arguments_error
message and raise_error behavior. Ensure callers can inspect the concrete cause
for malformed JSON, repair, or schema failures.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 58fa3175-1660-49ae-b1c4-68b4465c9c5b
📒 Files selected for processing (3)
lib/crewai/src/crewai/experimental/agent_executor.pylib/crewai/src/crewai/tools/tool_usage.pylib/crewai/tests/agents/test_agent_executor.py
| elif func_name in self._available_functions: | ||
| # Fallback to sync function for tools without structured_tool | ||
| try: | ||
| tool_func = self._available_functions[func_name] | ||
| raw_result = tool_func(**args_dict) | ||
| raw_tool_result = raw_result | ||
|
|
||
| # Add to cache after successful execution (before string conversion) | ||
| if self.tools_handler and self.tools_handler.cache: | ||
| should_cache = True | ||
| if original_tool: | ||
| should_cache = original_tool.cache_function( | ||
| args_dict, 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 | ||
| ) |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
Sync fallback blocks the event loop — wrap in asyncio.to_thread.
The removed ThreadPoolExecutor path let genuinely synchronous tools run off the loop. Calling tool_func(**args_dict) directly here blocks the running loop for the duration of the tool, which (a) serializes the asyncio.gather batch at Line 1743 so parallel native tool calls are no longer parallel, and (b) stalls every other coroutine on the agent's loop. The PR objective explicitly states thread-pool execution should be preserved for synchronous tools.
Note the existing parallel test (test_execute_native_tool_runs_parallel_for_multiple_calls) does not catch this: two 0.2s sleeps serialize to ~0.4s, still under the < 0.5 assertion. Tightening that bound would surface the regression.
🐛 Proposed fix
elif func_name in self._available_functions:
- # Fallback to sync function for tools without structured_tool
+ # Fallback for tools without a structured tool: run sync callables
+ # off the event loop, await coroutine functions natively.
try:
tool_func = self._available_functions[func_name]
- raw_result = tool_func(**args_dict)
+ if inspect.iscoroutinefunction(tool_func):
+ raw_result = await tool_func(**args_dict)
+ else:
+ raw_result = await asyncio.to_thread(
+ functools.partial(tool_func, **args_dict)
+ )
raw_tool_result = raw_result📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| elif func_name in self._available_functions: | |
| # Fallback to sync function for tools without structured_tool | |
| try: | |
| tool_func = self._available_functions[func_name] | |
| raw_result = tool_func(**args_dict) | |
| raw_tool_result = raw_result | |
| # Add to cache after successful execution (before string conversion) | |
| if self.tools_handler and self.tools_handler.cache: | |
| should_cache = True | |
| if original_tool: | |
| should_cache = original_tool.cache_function( | |
| args_dict, 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 | |
| ) | |
| elif func_name in self._available_functions: | |
| # Fallback for tools without a structured tool: run sync callables | |
| # off the event loop, await coroutine functions natively. | |
| try: | |
| tool_func = self._available_functions[func_name] | |
| if inspect.iscoroutinefunction(tool_func): | |
| raw_result = await tool_func(**args_dict) | |
| else: | |
| raw_result = await asyncio.to_thread( | |
| functools.partial(tool_func, **args_dict) | |
| ) | |
| raw_tool_result = raw_result | |
| # Add to cache after successful execution (before string conversion) | |
| if self.tools_handler and self.tools_handler.cache: | |
| should_cache = True | |
| if original_tool: | |
| should_cache = original_tool.cache_function( | |
| args_dict, 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 | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/crewai/src/crewai/experimental/agent_executor.py` around lines 2203 -
2224, Update the synchronous fallback in the native tool execution path
identified by _available_functions and tool_func so tool_func(**args_dict) runs
via asyncio.to_thread instead of directly on the event loop. Await the offloaded
call while preserving the existing raw-result, caching, and formatting flow for
synchronous tools.
… factory.py - Line 199: Bedrock missing bucket config now raises PermanentUploadError - Line 216: Unknown provider now raises PermanentUploadError - All ImportError cases now raise PermanentUploadError with helpful messages - Matches pattern from crewAIInc#6430 fix for consistent error handling
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
lib/crewai-files/src/crewai_files/uploaders/factory.py (1)
150-154: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winDo not classify every
ImportErrorincreate_file_uploaderas a missing SDK.These
tryblocks cover both the uploader import and the constructor, so unrelated transitive import failures or constructor bugs inGeminiFileUploader,AnthropicFileUploader,OpenAIFileUploader, orBedrockFileUploaderare logged as “package not installed” and converted into non-retryablePermanentUploadError. Catch only the expected dependency failure by name, or re-raise unexpectedImportErrorinstances while preserving the original exception.Also applies to lines 165-169, 185-187, and 214-216.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@lib/crewai-files/src/crewai_files/uploaders/factory.py` around lines 150 - 154, Update create_file_uploader’s provider-specific exception handling so only the expected missing SDK dependency is converted to PermanentUploadError and logged as not installed. Preserve the original exception and re-raise unrelated ImportError instances from uploader imports or constructors for GeminiFileUploader, AnthropicFileUploader, OpenAIFileUploader, and BedrockFileUploader; apply the same behavior to all four try blocks.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@lib/crewai-files/src/crewai_files/uploaders/factory.py`:
- Around line 218-219: Update the get_uploader() docstring to state that
unsupported providers raise PermanentUploadError instead of returning None. Keep
the documented behavior for supported providers accurate and aligned with the
existing exception path.
---
Outside diff comments:
In `@lib/crewai-files/src/crewai_files/uploaders/factory.py`:
- Around line 150-154: Update create_file_uploader’s provider-specific exception
handling so only the expected missing SDK dependency is converted to
PermanentUploadError and logged as not installed. Preserve the original
exception and re-raise unrelated ImportError instances from uploader imports or
constructors for GeminiFileUploader, AnthropicFileUploader, OpenAIFileUploader,
and BedrockFileUploader; apply the same behavior to all four try blocks.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: a5442dce-6ad5-4dcb-87ca-f9f570d4c4bd
📒 Files selected for processing (1)
lib/crewai-files/src/crewai_files/uploaders/factory.py
| logger.debug(f"No file uploader available for provider: {provider}") | ||
| raise | ||
| raise PermanentUploadError(f"No file uploader available for provider: {provider}") |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Update the get_uploader() API documentation.
The docstring says unsupported providers return None, but this path now raises PermanentUploadError. Document the exception behavior so callers do not rely on an invalid return contract.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/crewai-files/src/crewai_files/uploaders/factory.py` around lines 218 -
219, Update the get_uploader() docstring to state that unsupported providers
raise PermanentUploadError instead of returning None. Keep the documented
behavior for supported providers accurate and aligned with the existing
exception path.
Source: Coding guidelines
Summary
Fixes #6611 - Async tools are now awaited natively on the running event loop in the experimental native function calling executor, consistent with the ReAct executor.
Changes
_execute_single_native_tool_call_asyncmethod that usesCrewStructuredTool.ainvoke()for proper async tool executionexecute_native_toolto async method usingasyncio.gather()for parallel execution_available_functions(thread pool)ThreadPoolExecutorimport, now uses pure asyncasyncio.run()the async methodRoot Cause
The experimental native function calling executor called tools synchronously via
tool.run()which bridged coroutines withasyncio.run()- creating a fresh nested event loop per call in a worker thread. The ReAct executor correctly usesCrewStructuredTool.ainvoke()which detects coroutine functions and awaits them natively on the running loop.Fix
The fix routes native tool calls through the same async-aware dispatch pattern:
structured_tool.ainvoke(args_dict)- awaits coroutine functions natively on the current event looprun_in_executor)asyncio.gather()instead ofThreadPoolExecutorTesting
_runnow executes on MainThread (is_main=True) instead of worker thread