[rkfshakti]

Fixes #8904.

Follow-up to #7630 (the closed #7634 was the pre-#8192 fix for the same family).

The bug (reproduced on main)

After #8192, every sync Flow method runs in a worker thread (asyncio.to_thread, flow runtime _execute_method). An async native tool's coroutine therefore goes through the bare asyncio.run() inside Tool.run — on a fresh, unrelated event loop per call. Anything bound to the kickoff caller's loop fails deterministically:

RuntimeError: Task ... got Future <Future pending> attached to a different loop

Minimal repro on unpatched main: a Crew with one @tool async def; the tool awaits a Future created on the kickoff_async caller's loop after asyncio.run(crew.kickoff_async()) → the tool result the LLM sees is Error executing tool: ... attached to a different loop. Same shape for any caller-loop-bound resource (httpx.AsyncClient created before kickoff, sockets).

The fix

• The async kickoffs capture the caller's event loop into a ContextVar (Crew.kickoff_async non-stream branch, AgentExecutor.invoke_async). Crew.kickoff_async is await asyncio.to_thread(self.kickoff, ...), so capturing before the thread boundary is what makes the loop visible inside every worker lane — ContextVars propagate through to_thread and the Flow runtime's copy_context (including the ThreadPoolExecutor in execute_native_tool).
• agent_utils.run_native_tool_result(...) executes the tool: when a caller loop is present and the tool has a working async implementation, bridge tool.arun(**args) onto the caller loop via run_coroutine_threadsafe(...).result() (the worker thread blocks; the coroutine runs with caller-loop affinity). Otherwise the exact historical sync behavior — the available_functions mapping stays authoritative, so test doubles and cache wrappers are untouched.
• Async-capability detection: Tool-wrapped functions are async iff the wrapped function is a coroutine function (Tool always overrides _arun, even for sync funcs); class-based BaseTools qualify when their type overrides the NotImplementedError-raising BaseTool._arun.

Tests

New lib/crewai/tests/agents/test_native_async_tool_loop_affinity.py:

• worker-thread bridge onto a caller-loop Future (unit, with and without caller loop),
• full Crew + scripted-LLM kickoff_async path asserting the caller-loop Future resolves instead of the cross-loop error.

Verification

• ruff check clean on all touched files.
• A/B vs pristine main: pytest lib/crewai/tests/agents -k "native or agent_executor or invoke" — identical failure sets (3 pre-existing TestDeferredNativeToolResults / TestResponseFormatWithKickoff failures reproduce on pristine main in this environment; 172 pass on both), plus 4 new green tests.
• With the patch, the repro above returns got:expected-item instead of the error.

---

Note

Medium Risk
Changes core tool execution and cross-thread asyncio bridging on every async kickoff; fallbacks and tests limit blast radius but blocking future.result() in worker threads can stall the caller loop if async tools block.

Overview
Fixes "attached to a different loop" failures when async native tools run under kickoff_async / invoke_async, where sync Flow work executes in asyncio.to_thread worker threads and tools previously ran via a fresh loop inside Tool.run.

run_native_tool_result in agent_utils captures the caller loop in a ContextVar at async entry points (Crew.kickoff_async, AgentExecutor.invoke_async) and, for tools with a real async implementation, schedules tool.arun on that loop with run_coroutine_threadsafe. Sync tools, missing caller loop, deadlock-prone cases (already on target loop), and closed loops keep the prior tool_func(**args) path.

Native tool execution in agent_executor and shared agent_utils helpers now routes through this helper instead of calling the mapped function directly. Adds regression tests in test_native_async_tool_loop_affinity.py (unit bridge, error propagation, E2E kickoff_async).

^{Reviewed by Cursor Bugbot for commit f1afd0d. Bugbot is set up for automated code reviews on this repo. Configure here.}

[strix-security]

Strix Security Review

Warning

This pull request has 94 commits after the last Strix review (39f754e). Strix has not reviewed these changes.
Automatic review on push is off for this repository. To review the latest changes, tag @strix-security in a comment, or turn on re-review on push.

1 open security finding on this PR:

• 🟢 LOW Async native tool is silently executed twice when its coroutine raises RuntimeError — lib/crewai/src/crewai/utilities/agent_utils.py:1814-1820

Review summary

Reviewed all four changed files of the async-native-tool caller-loop fix, tracing the new bridge from Crew.kickoff_async and AgentExecutor.invoke_async through the ContextVar capture, the flow runtime's asyncio.to_thread dispatch, the native-tool thread pool, and both call sites in agent_executor.py / agent_utils.py, plus the tool classes (Tool, BaseTool, deferred-result reader, MCP wrappers, bedrock code-interpreter family) whose run/arun semantics the new path depends on. The caller-loop affinity mechanism itself is sound and not exploitable, but the error handling around it is not: the finding "Async native tool is silently executed twice when its coroutine raises RuntimeError" documents a demonstrated duplicate side effect, a masked tool failure, and skewed usage accounting whenever a tool's own RuntimeError is mistaken for a closed caller loop. The same block also lacks a loop-identity guard (a permanent hang if the bridge is ever invoked on the caller loop's own thread) and moves blocking async tool bodies onto the application's shared loop, which is recorded as a secondary operational consequence with its measured evidence.

_{Fixed the findings? re-run the review, or tag @strix-security in a PR comment to run a fresh review.}

_{Updated for 39f754e.}

---

Reviewed by Strix
_{Re-run review · Configure security review settings}

[strix-security]

Strix flagged a new security finding below. See the pinned summary comment for the full PR status.

[strix-security]

🟢 Async native tool is silently executed twice when its coroutine raises RuntimeError

Severity: LOW · CWE-754

Summary

The new helper run_native_tool_result wraps asyncio.run_coroutine_threadsafe(tool.arun(**args_dict), loop).result() in except RuntimeError: pass, then falls through to the legacy invocation tool_func(**args_dict). Because concurrent.futures.Future.result() re-raises whatever exception the coroutine raised, a RuntimeError raised by the tool itself is indistinguishable from "the caller's event loop is already closed". The handler therefore swallows the tool's exception and invokes the tool a second time through the legacy path, within a single logical tool call.

Consequence: duplicated side effect, masked failure

For any async native tool that raises RuntimeError after performing a side effect, the side effect happens twice while the caller only observes the second attempt:

• the genuine failure never reaches the call site's except Exception, failure_from_exception(), task.increment_tools_errors(), the error event or the tool-failure policy, so failure accounting and alerting miss it;
• when the retry succeeds, the agent sees success and the failure is silently masked;
• current_usage_count is incremented once per attempt, so max_usage_count accounting is wrong (two claims for one call).

Same block: unguarded re-entrancy and shared-loop starvation

Two further properties of the same unconditional bridge were measured on the patched tree:

• Re-entrancy: the helper never checks whether loop is the loop already executing the current thread. Invoked on the caller loop's own thread, future.result() blocks that very loop, so the submission can never be serviced and the call never returns — the process must be killed, and even an outer asyncio.wait_for timeout cannot fire because the loop itself is blocked. No code path in the framework was found that reaches the helper on the loop thread, so this is a latent hazard in a public helper rather than a demonstrated framework defect; it becomes reachable the moment any caller stops offloading tool execution to a worker thread.
• Starvation of the application's loop: because the coroutine now runs on the loop the application awaits on, an async tool whose body performs blocking work stalls every other task on that loop for the tool's whole duration. Measured with a 10 ms heartbeat on the caller loop and a 1.0 s blocking tool body: maximum loop gap 1.001 s through the new bridge versus 0.012 s when the same tool runs the previous way on its private loop; end-to-end through kickoff_async, 1.005 s versus 0.015 s. This is reachable with shipped tools — several class-based tools declared as async def _arun simply forward to their blocking synchronous implementation, and are now classified as async-capable and executed on the shared loop. No availability metric is claimed for the framework defect itself, because the stall requires a tool implementation that blocks inside a coroutine (a defect in the tool, and the remediation for it lives outside this change); it is recorded here so the operational consequence of the new bridge is visible.

Demonstrated

A dynamic repro against the patched tree runs an async native tool through the async kickoff path (kickoff_async → worker thread → run_native_tool_result) where the tool records a side effect and raises RuntimeError on its first attempt:

• tool body executed 2 times (attempt 1 on the kickoff caller's loop, attempt 2 on a fresh loop);
• current_usage_count == 2 for one call;
• end-to-end kickoff_async run: 2 side effects recorded and the model saw the successful value — the call was reported as a success;
• pre-patch baseline, tool.run(...) called directly: 1 execution and the RuntimeError propagates.

Exceptions other than RuntimeError, sync tools and sync-kickoff paths are unaffected, which bounds the defect to async native tools reached through the caller-loop bridge.

Suggested change

	loop = caller_loop if caller_loop is not None else _caller_loop_context.get()
	if loop is not None and tool is not None and _has_async_implementation(tool):
	try:
	return asyncio.run_coroutine_threadsafe(tool.arun(**args_dict), loop).result()
	except RuntimeError:
	pass # caller loop is already closed — run() fresh-loop is the fallback
	return tool_func(**args_dict)
	loop = caller_loop if caller_loop is not None else _caller_loop_context.get()
	if loop is not None and tool is not None and _has_async_implementation(tool):
	try:
	running_loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop()
	except RuntimeError:
	running_loop = None
	# Bridging onto the loop that is already executing this call would block that
	# loop inside future.result() forever, so fall back to the sync path instead.
	if running_loop is not loop and not loop.is_closed():
	coro = tool.arun(**args_dict)
	try:
	future = asyncio.run_coroutine_threadsafe(coro, loop)
	except RuntimeError:
	coro.close() # loop shut down mid-flight; use the run() fallback
	else:
	# Tool exceptions (RuntimeError included) must propagate: re-running
	# the tool here would repeat its side effects and hide the failure.
	return future.result()
	return tool_func(**args_dict)

Prompt to fix with AI

This is a security vulnerability found during a code review.

Vulnerability: Async native tool is silently executed twice when its coroutine raises RuntimeError
Severity: LOW
CWE: CWE-754

## Summary

The new helper `run_native_tool_result` wraps `asyncio.run_coroutine_threadsafe(tool.arun(**args_dict), loop).result()` in `except RuntimeError: pass`, then falls through to the legacy invocation `tool_func(**args_dict)`. Because `concurrent.futures.Future.result()` re-raises whatever exception the coroutine raised, a `RuntimeError` raised **by the tool itself** is indistinguishable from "the caller's event loop is already closed". The handler therefore swallows the tool's exception and **invokes the tool a second time** through the legacy path, within a single logical tool call.

## Consequence: duplicated side effect, masked failure

For any async native tool that raises `RuntimeError` after performing a side effect, the side effect happens twice while the caller only observes the second attempt:

- the genuine failure never reaches the call site's `except Exception`, `failure_from_exception()`, `task.increment_tools_errors()`, the error event or the tool-failure policy, so failure accounting and alerting miss it;
- when the retry succeeds, the agent sees success and the failure is silently masked;
- `current_usage_count` is incremented once per attempt, so `max_usage_count` accounting is wrong (two claims for one call).

## Same block: unguarded re-entrancy and shared-loop starvation

Two further properties of the same unconditional bridge were measured on the patched tree:

- **Re-entrancy**: the helper never checks whether `loop` is the loop already executing the current thread. Invoked on the caller loop's own thread, `future.result()` blocks that very loop, so the submission can never be serviced and the call never returns — the process must be killed, and even an outer `asyncio.wait_for` timeout cannot fire because the loop itself is blocked. No code path in the framework was found that reaches the helper on the loop thread, so this is a latent hazard in a public helper rather than a demonstrated framework defect; it becomes reachable the moment any caller stops offloading tool execution to a worker thread.
- **Starvation of the application's loop**: because the coroutine now runs on the loop the application awaits on, an async tool whose body performs blocking work stalls every other task on that loop for the tool's whole duration. Measured with a 10 ms heartbeat on the caller loop and a 1.0 s blocking tool body: maximum loop gap `1.001 s` through the new bridge versus `0.012 s` when the same tool runs the previous way on its private loop; end-to-end through `kickoff_async`, `1.005 s` versus `0.015 s`. This is reachable with shipped tools — several class-based tools declared as `async def _arun` simply forward to their blocking synchronous implementation, and are now classified as async-capable and executed on the shared loop. No availability metric is claimed for the framework defect itself, because the stall requires a tool implementation that blocks inside a coroutine (a defect in the tool, and the remediation for it lives outside this change); it is recorded here so the operational consequence of the new bridge is visible.

## Demonstrated

A dynamic repro against the patched tree runs an async native tool through the async kickoff path (`kickoff_async` → worker thread → `run_native_tool_result`) where the tool records a side effect and raises `RuntimeError` on its first attempt:

- tool body executed **2 times** (attempt 1 on the kickoff caller's loop, attempt 2 on a fresh loop);
- `current_usage_count == 2` for one call;
- end-to-end `kickoff_async` run: **2 side effects** recorded and the model saw the successful value — the call was reported as a success;
- pre-patch baseline, `tool.run(...)` called directly: 1 execution and the `RuntimeError` propagates.

Exceptions other than `RuntimeError`, sync tools and sync-kickoff paths are unaffected, which bounds the defect to async native tools reached through the caller-loop bridge.

Location: lib/crewai/src/crewai/utilities/agent_utils.py:1814-1820
Context: Caller-loop bridge swallows the tool's own RuntimeError, re-invokes the tool, and never checks that the target loop is not the one already executing the call
```
// Before:
    loop = caller_loop if caller_loop is not None else _caller_loop_context.get()
    if loop is not None and tool is not None and _has_async_implementation(tool):
        try:
            return asyncio.run_coroutine_threadsafe(tool.arun(**args_dict), loop).result()
        except RuntimeError:
            pass  # caller loop is already closed — run() fresh-loop is the fallback
    return tool_func(**args_dict)
// After:
    loop = caller_loop if caller_loop is not None else _caller_loop_context.get()
    if loop is not None and tool is not None and _has_async_implementation(tool):
        try:
            running_loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop()
        except RuntimeError:
            running_loop = None
        # Bridging onto the loop that is already executing this call would block that
        # loop inside future.result() forever, so fall back to the sync path instead.
        if running_loop is not loop and not loop.is_closed():
            coro = tool.arun(**args_dict)
            try:
                future = asyncio.run_coroutine_threadsafe(coro, loop)
            except RuntimeError:
                coro.close()  # loop shut down mid-flight; use the run() fallback
            else:
                # Tool exceptions (RuntimeError included) must propagate: re-running
                # the tool here would repeat its side effects and hide the failure.
                return future.result()
    return tool_func(**args_dict)
```

How to fix:
Distinguish "cannot schedule onto the caller loop" from "the tool failed", and never run the bridge on the loop that is already executing the call. Check `loop.is_closed()` and compare against the thread's currently running loop before scheduling, treat only a scheduling failure as the signal to fall back, and let every exception raised by the coroutine propagate unchanged out of the helper. Never re-invoke a tool after it has started executing: at that point its side effects are already in flight, so a second invocation converts one failure into a duplicated state change. When a fallback is required, close the unscheduled coroutine so it is not left un-awaited.

Separately, keep asynchronous tool implementations non-blocking (offload blocking client calls to a worker thread) so that executing them on the caller's loop does not stall unrelated work, and document that requirement for the new caller-loop affinity behaviour. Add a regression test asserting that an async native tool raising `RuntimeError` executes exactly once, that the error reaches the agent as a tool failure, and that the usage counter increments once.

Please fix this vulnerability. If you propose a fix, make it concise and minimal.

_{React 👍 / 👎 to tune Strix for this repo. A repo collaborator (or the PR author) can resolve this thread to dismiss the finding.}

[rkfshakti]

Fixed in 8bb5045d8 — thanks, the analysis was spot on, including the double _claim_usage().

Change (helper only, same 3-branch shape you suggested):

• Scheduling vs execution failures are now separated. Only a RuntimeError raised while scheduling (run_coroutine_threadsafe on a shut-down loop) falls back; the coroutine is close()d first so it isn't left un-awaited. Exceptions from future.result() — RuntimeError included — propagate unchanged. The fallback no longer re-invokes a tool that has already started executing, so no duplicated side effects and no masked failure.
• Re-entrancy guard. If the helper is called on the target loop's own thread, the bridge is skipped and the provided sync callable is used — future.result() can never block the loop it would need to service itself.
• Blocking contract documented (your third point): bridged coroutines run on the application's loop, so an async implementation that blocks stalls unrelated tasks; that remediation belongs in the tool (offload blocking calls), not in the bridge — abandoning loop affinity here would reintroduce the affinity regression this PR exists to fix. The docstring now says so explicitly.

Regression tests added (tests/agents/test_native_async_tool_loop_affinity.py):

• async tool raising RuntimeError after a side effect → exception propagates, executed exactly once, current_usage_count == 1;
• helper invoked on the target loop's own thread → sync fallback used, async path never scheduled (no deadlock).

Verification: 6/6 tests in the file green, ruff clean, and the tests/agents -k "native or agent_executor or invoke" A/B against pristine main still shows an identical failure set (3 pre-existing environment failures on both trees).

[rkfshakti]

Good catch — that one's on me. The except RuntimeError was meant for a loop that's already closed, but .result() re-raises whatever the tool itself threw, so an async tool raising RuntimeError after a side effect could run twice and the real failure was masked. Fixed in 8bb5045: only scheduling failures fall back now (and I close the unscheduled coroutine), plus a guard so the helper never tries to schedule onto the loop it's already running on — that case would have deadlocked. Added two regression tests: exactly-once execution when the tool raises, and the no-deadlock path. The blocking-inside-a-coroutine point is fair as well — I documented that as a contract on the helper rather than papering over it in the bridge, since fixing it here would mean giving up the caller-loop affinity this PR is about.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 2 potential issues.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

Want higher recall? High effort reviews run extra passes and find more bugs. A team admin can switch effort levels in the Cursor dashboard.

^{Reviewed by Cursor Bugbot for commit 773a81c. Configure here.}

[cursor]

Streaming kickoff skips caller loop

Medium Severity

The streaming branch of kickoff_async still offloads via asyncio.to_thread(self.kickoff) without capturing the caller loop in _caller_loop_context. Async native tools on that path keep running on a fresh loop, so caller-loop-bound resources fail with the same cross-loop RuntimeError this change aims to fix.

 

^{Reviewed by Cursor Bugbot for commit 773a81c. Configure here.}

[cursor]

Inherited async tools not detected

Medium Severity

_has_async_implementation only inspects type(tool).__dict__ for _arun, so a class-based tool that inherits its async implementation from a parent or mixin is treated as sync. Those tools skip the caller-loop bridge and still run through a fresh asyncio.run loop.

 

^{Reviewed by Cursor Bugbot for commit 773a81c. Configure here.}

[rkfshakti]

Rebased onto current main (was 10 behind → 0; head 939884ce6, clean replay, no conflicts).

• Focused regression suite 6/6 green: the 4 loop-affinity tests plus the 2 security-review regressions (no re-invocation after a bridge failure / exactly-once, and the loop-affinity no-deadlock guard).
• The 14 failing tests in tests/agents/test_native_tool_calling.py (anthropic/gemini function-calling) reproduce identically on pristine main at 38db11e33 in the same local environment — pre-existing, unrelated to this change.

The fork-workflow runs need maintainer approval again after the force-push — re-requesting when convenient. 🙏

[github-actions]

Thanks for the pull request.

First-time contributors need an associated open issue before we can review a PR.

1. Open an issue with a template, or pick an existing open one.
2. Open a new PR (or reopen this one) whose title or body mentions that issue, for example #123.

See the contributing guide.

[rkfshakti]

Superseded by #8905 (same commits, rebased onto current main @ 5f948e386). This PR was auto-closed by the needs-issue gate because its only candidate issue references were #7630 (closed) and #8192/#7634 (PRs); #8904 is now open and linked.