fix(agent): keep guardrail retries on the async path in kickoff_async

Author: bunnysayzz

lib/crewai/src/crewai/agent/core.py

@@ -1803,6 +1803,42 @@ def _finalize_kickoff(
 
         return output
 
+    async def _finalize_kickoff_async(
+        self,
+        output: LiteAgentOutput,
+        executor: AgentExecutor,
+        inputs: dict[str, str],
+        response_format: type[Any] | None,
+        messages: str | list[LLMMessage],
+        agent_info: dict[str, Any],
+        usage_baseline: UsageMetrics | None = None,
+    ) -> LiteAgentOutput:
+        """Async variant of _finalize_kickoff for kickoff_async.
+
+        Identical except guardrail retries stay on the async execution path;
+        see _process_kickoff_guardrail_async.
+        """
+        if self.guardrail is not None:
+            output = await self._process_kickoff_guardrail_async(
+                output=output,
+                executor=executor,
+                inputs=inputs,
+                response_format=response_format,
+                usage_baseline=usage_baseline,
+            )
+
+        self._save_kickoff_to_memory(messages, output.raw)
+
+        crewai_event_bus.emit(
+            self,
+            event=LiteAgentExecutionCompletedEvent(
+                agent_info=agent_info,
+                output=output.raw,
+            ),
+        )
+
+        return output
+
     def _emit_kickoff_error(self, agent_info: dict[str, Any], e: Exception) -> NoReturn:
         """Emit a kickoff error event and re-raise."""
         crewai_event_bus.emit(
@@ -1972,6 +2008,31 @@ async def _execute_and_build_output_async(
             result, executor, response_format, usage_baseline, kickoff_failures
         )
 
+    def _resolve_guardrail_callable(self) -> GuardrailCallable | None:
+        """Return the configured guardrail as a callable, or None if unset."""
+        if isinstance(self.guardrail, str):
+            from crewai.tasks.llm_guardrail import LLMGuardrail
+
+            return cast(
+                GuardrailCallable,
+                LLMGuardrail(description=self.guardrail, llm=cast(BaseLLM, self.llm)),
+            )
+        if callable(self.guardrail):
+            return self.guardrail
+        return None
+
+    @staticmethod
+    def _apply_guardrail_result(
+        output: LiteAgentOutput, guardrail_result: Any
+    ) -> LiteAgentOutput:
+        """Fold an accepted guardrail result into the output (shared sync/async)."""
+        if guardrail_result.result is not None:
+            if isinstance(guardrail_result.result, str):
+                output.raw = guardrail_result.result
+            elif isinstance(guardrail_result.result, BaseModel):
+                output.pydantic = guardrail_result.result
+        return output
+
     def _process_kickoff_guardrail(
         self,
         output: LiteAgentOutput,
@@ -1996,17 +2057,8 @@ def _process_kickoff_guardrail(
         Returns:
             Validated/updated output.
         """
-        guardrail_callable: GuardrailCallable
-        if isinstance(self.guardrail, str):
-            from crewai.tasks.llm_guardrail import LLMGuardrail
-
-            guardrail_callable = cast(
-                GuardrailCallable,
-                LLMGuardrail(description=self.guardrail, llm=cast(BaseLLM, self.llm)),
-            )
-        elif callable(self.guardrail):
-            guardrail_callable = self.guardrail
-        else:
+        guardrail_callable = self._resolve_guardrail_callable()
+        if guardrail_callable is None:
             return output
 
         guardrail_result = process_guardrail(
@@ -2048,13 +2100,68 @@ def _process_kickoff_guardrail(
                 usage_baseline=usage_baseline,
             )
 
-        if guardrail_result.result is not None:
-            if isinstance(guardrail_result.result, str):
-                output.raw = guardrail_result.result
-            elif isinstance(guardrail_result.result, BaseModel):
-                output.pydantic = guardrail_result.result
+        return self._apply_guardrail_result(output, guardrail_result)
 
-        return output
+    async def _process_kickoff_guardrail_async(
+        self,
+        output: LiteAgentOutput,
+        executor: AgentExecutor,
+        inputs: dict[str, str],
+        response_format: type[Any] | None = None,
+        retry_count: int = 0,
+        usage_baseline: UsageMetrics | None = None,
+    ) -> LiteAgentOutput:
+        """Async variant of _process_kickoff_guardrail for kickoff_async.
+
+        Identical except the retry re-executes through
+        _execute_and_build_output_async: the sync executor.invoke() detects a
+        running event loop and hands back an unawaited coroutine instead of a
+        result dict, which then crashes in _build_output_from_result.
+        """
+        guardrail_callable = self._resolve_guardrail_callable()
+        if guardrail_callable is None:
+            return output
+
+        guardrail_result = process_guardrail(
+            output=output,
+            guardrail=guardrail_callable,
+            retry_count=retry_count,
+            event_source=self,
+            from_agent=self,
+        )
+
+        if not guardrail_result.success:
+            if retry_count >= self.guardrail_max_retries:
+                raise ValueError(
+                    f"Agent's guardrail failed validation after {self.guardrail_max_retries} retries. "
+                    f"Last error: {guardrail_result.error}"
+                )
+
+            executor._append_message_to_state(
+                guardrail_result.error or "Guardrail validation failed",
+                role="user",
+            )
+
+            retried = await self._execute_and_build_output_async(
+                executor, inputs, response_format, usage_baseline
+            )
+            # The retry opens its own collector, so carry the blocked attempt's
+            # failures forward or they vanish from the final output.
+            retried.tool_failures = merge_tool_failures(
+                output.tool_failures, retried.tool_failures
+            )
+            output = retried
+
+            return await self._process_kickoff_guardrail_async(
+                output=output,
+                executor=executor,
+                inputs=inputs,
+                response_format=response_format,
+                retry_count=retry_count + 1,
+                usage_baseline=usage_baseline,
+            )
+
+        return self._apply_guardrail_result(output, guardrail_result)
 
     async def kickoff_async(
         self,
@@ -2118,7 +2225,7 @@ async def kickoff_async(
             output = await self._execute_and_build_output_async(
                 executor, inputs, response_format, usage_baseline
             )
-            return self._finalize_kickoff(
+            return await self._finalize_kickoff_async(
                 output,
                 executor,
                 inputs,

lib/crewai/tests/agents/test_agent_kickoff_guardrail_async.py

@@ -0,0 +1,85 @@
+"""Guardrail retries inside kickoff_async must stay on the async path (#7252)."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from crewai import Agent
+from crewai.lite_agent_output import LiteAgentOutput
+
+
+def _agent_with_fail_once_guardrail():
+    calls = {"n": 0}
+
+    def fail_once(output):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            return (False, "not good enough, retry")
+        return (True, output)
+
+    agent = Agent(
+        role="Test Agent",
+        goal="Answer",
+        backstory="Test backstory.",
+        guardrail=fail_once,
+        guardrail_max_retries=2,
+    )
+    return agent, calls
+
+
+def _canned_output(raw="final answer"):
+    return LiteAgentOutput(raw=raw, agent_role="Test Agent")
+
+
+@pytest.mark.asyncio
+async def test_kickoff_async_guardrail_retry_uses_async_execution():
+    """A guardrail failure on kickoff_async must retry via invoke_async.
+
+    Before the fix the retry went through the sync executor.invoke(), which
+    under a running loop hands back an unawaited coroutine instead of a result
+    dict and crashes on .get("output"). The sync execute path must not run at
+    all here.
+    """
+    agent, calls = _agent_with_fail_once_guardrail()
+
+    with (
+        patch.object(
+            Agent, "_prepare_kickoff", return_value=(MagicMock(), {}, {}, [])
+        ),
+        patch.object(Agent, "_current_usage_summary", return_value=MagicMock()),
+        patch.object(
+            Agent,
+            "_execute_and_build_output_async",
+            new=AsyncMock(side_effect=[_canned_output("first"), _canned_output("final answer")]),
+        ),
+        patch.object(
+            Agent,
+            "_execute_and_build_output",
+            side_effect=AssertionError("sync execution path taken in async context"),
+        ),
+    ):
+        result = await agent.kickoff_async("answer me")
+
+    assert calls["n"] == 2
+    assert result.raw == "final answer"
+
+
+@pytest.mark.asyncio
+async def test_kickoff_async_guardrail_retry_exhaustion_still_raises():
+    """Retry limits are honored on the async path too."""
+    agent, _ = _agent_with_fail_once_guardrail()
+    agent.guardrail_max_retries = 0
+
+    with (
+        patch.object(
+            Agent, "_prepare_kickoff", return_value=(MagicMock(), {}, {}, [])
+        ),
+        patch.object(Agent, "_current_usage_summary", return_value=MagicMock()),
+        patch.object(
+            Agent,
+            "_execute_and_build_output_async",
+            new=AsyncMock(return_value=_canned_output("first")),
+        ),
+    ):
+        with pytest.raises(ValueError, match="guardrail failed validation"):
+            await agent.kickoff_async("answer me")