fix: await async tools natively in _ainvoke_loop_native_tools#6622
Open
rkfshakti wants to merge 1 commit into
Open
fix: await async tools natively in _ainvoke_loop_native_tools#6622rkfshakti wants to merge 1 commit into
rkfshakti wants to merge 1 commit into
Conversation
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 crewAIInc#6611
📝 WalkthroughWalkthrough
ChangesAsync native tool execution
Sequence Diagram(s)sequenceDiagram
participant NativeToolsLoop
participant CrewAgentExecutor
participant Tool
NativeToolsLoop->>CrewAgentExecutor: await _ahandle_native_tool_calls(tool_calls)
CrewAgentExecutor->>Tool: await _aexecute_single_native_tool_call(...)
Tool-->>CrewAgentExecutor: tool result
CrewAgentExecutor-->>NativeToolsLoop: AgentFinish or updated tool messages
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 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
🧹 Nitpick comments (1)
lib/crewai/src/crewai/agents/crew_agent_executor.py (1)
1184-1377: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftHeavy duplication with
_execute_single_native_tool_call.
_aexecute_single_native_tool_callis a near-verbatim copy of the sync method (parsing, tool resolution, usage limits, cache, events, hooks) with only the invocation block (1301-1304) differing. Same applies to_ahandle_native_tool_callsvs_handle_native_tool_calls. This ~190-line duplication will drift as either path changes. Consider extracting the shared pre/post logic into helpers that accept an invocation callable, keeping only the sync-vs-async invocation distinct.🤖 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/agents/crew_agent_executor.py` around lines 1184 - 1377, Refactor _aexecute_single_native_tool_call and _execute_single_native_tool_call to share their common parsing, tool resolution, usage-limit, cache, event, hook, and result-processing logic through a helper that accepts the tool invocation operation. Keep only synchronous versus awaited invocation in separate callables, and apply the same deduplication to _ahandle_native_tool_calls and _handle_native_tool_calls while preserving existing behavior.
🤖 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/agents/crew_agent_executor.py`:
- Around line 1298-1304: Update the synchronous fallback in the tool execution
branch of the crew agent executor to run available_functions[func_name] via the
existing ThreadPoolExecutor or equivalent worker-thread mechanism, preserving
the current arguments and awaited result handling. Keep the output_tool.arun
path unchanged so synchronous tools no longer block the running event loop or
serialize gathered executions.
---
Nitpick comments:
In `@lib/crewai/src/crewai/agents/crew_agent_executor.py`:
- Around line 1184-1377: Refactor _aexecute_single_native_tool_call and
_execute_single_native_tool_call to share their common parsing, tool resolution,
usage-limit, cache, event, hook, and result-processing logic through a helper
that accepts the tool invocation operation. Keep only synchronous versus awaited
invocation in separate callables, and apply the same deduplication to
_ahandle_native_tool_calls and _handle_native_tool_calls while preserving
existing behavior.
🪄 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: 7e1f60bd-6faa-43c5-80fd-84b52fe84664
📒 Files selected for processing (1)
lib/crewai/src/crewai/agents/crew_agent_executor.py
Comment on lines
+1298
to
+1304
| 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 {})) |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
Sync tool fallback blocks the event loop and regresses batch parallelism.
When output_tool has no arun, the synchronous callable is invoked directly on the running loop. Inside asyncio.gather, this serializes sync tools and blocks the loop for their full duration — a regression from the previous sync path, which offloaded these calls to a ThreadPoolExecutor (true parallelism for I/O-bound sync tools). Offload to a worker thread instead.
♻️ Offload the sync fallback
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 = await asyncio.to_thread(
+ available_functions[func_name], **(args_dict or {})
+ )📝 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.
Suggested change
| 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 {})) | |
| 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 = await asyncio.to_thread( | |
| available_functions[func_name], **(args_dict or {}) | |
| ) |
🤖 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/agents/crew_agent_executor.py` around lines 1298 -
1304, Update the synchronous fallback in the tool execution branch of the crew
agent executor to run available_functions[func_name] via the existing
ThreadPoolExecutor or equivalent worker-thread mechanism, preserving the current
arguments and awaited result handling. Keep the output_tool.arun path unchanged
so synchronous tools no longer block the running event loop or serialize
gathered executions.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What this PR does
Adds async variants of
_handle_native_tool_callsand_execute_single_native_tool_callthat useawait tool.arun()instead oftool.run(), and updates_ainvoke_loop_native_toolsto call them.Why it's needed
When the async native tool path (
_ainvoke_loop_native_tools) calls the sync_handle_native_tool_calls, which callstool.run()→asyncio.run(), theasyncio.run()call crashes withRuntimeError: asyncio.run() cannot be called from a running event loopwhen the agent is invoked from an already-running event loop (e.g. viaainvoke()).The ReAct executor (
_ainvoke_loop_react) already handles this correctly viaaexecute_tool_and_check_finality()→tool_usage.ause()→await.What changed
_ahandle_native_tool_calls()— async variant that usesasyncio.gatherinstead ofThreadPoolExecutorfor parallel tool execution, so async tools are properly awaited rather than run throughasyncio.run()._aexecute_single_native_tool_call()— async variant that callsawait tool.arun()instead oftool.run(), avoiding the nestedasyncio.run()crash._ainvoke_loop_native_tools()now calls_ahandle_native_tool_calls()instead of_handle_native_tool_calls().Reviewer Test Plan
How to verify
_runmethod is a coroutine)crew.kickoff_async()withfunction_calling_llmset to a model that supports native function callingRuntimeError: asyncio.run() cannot be called from a running event loopTested on
Risk & Scope
asyncio.gatherinstead ofThreadPoolExecutorfor parallel execution. This is the correct approach for async code but means parallel tool execution is cooperative rather than preemptive.AgentExecutor(Flow-based) has the same bug but is in a separate code path.Linked Issues
Fixes #6611