From a6deaed50c27b4675f3f321bf1743665837f21cd Mon Sep 17 00:00:00 2001 From: dongcao Date: Wed, 8 Jul 2026 12:33:15 +0800 Subject: [PATCH] fix: support async before/after_kickoff_callbacks in akickoff before_kickoff_callbacks and after_kickoff_callbacks were invoked synchronously in akickoff, silently dropping async callables and blocking the event loop on IO-bound sync callbacks. - Extract _begin_prepare_kickoff, _normalize_inputs, and _finish_prepare_kickoff helpers from prepare_kickoff to eliminate code duplication between the sync and async paths. - Add aprepare_kickoff (async counterpart of prepare_kickoff) that awaits coroutine before-callbacks via inspect.isawaitable, consistent with the existing task_callback pattern in task.py. - Use aprepare_kickoff in akickoff instead of prepare_kickoff. - Add inspect.isawaitable check for after_kickoff_callbacks in akickoff. - Add tests covering async before, async after, and mixed sync+async callback pipelines. kickoff_async is unaffected: it wraps the entire sync kickoff in asyncio.to_thread, so before/after callbacks already run off the event loop. --- lib/crewai/src/crewai/crew.py | 6 +- lib/crewai/src/crewai/crews/utils.py | 118 +++++++++++++++------ lib/crewai/tests/crew/test_async_crew.py | 127 +++++++++++++++++++++++ 3 files changed, 218 insertions(+), 33 deletions(-) diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index fda456be40..783ac3edef 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import inspect from collections.abc import Callable, Sequence from concurrent.futures import Future from copy import copy as shallow_copy @@ -67,6 +68,7 @@ def get_supported_content_types(provider: str, api: str | None = None) -> list[s from crewai.crews.crew_output import CrewOutput from crewai.crews.utils import ( StreamingContext, + aprepare_kickoff, check_conditional_skip, enable_agent_streaming, prepare_kickoff, @@ -1230,7 +1232,7 @@ async def run_crew() -> None: runtime_scope = crewai_event_bus._enter_runtime_scope() try: - inputs = prepare_kickoff(self, inputs, input_files) + inputs = await aprepare_kickoff(self, inputs, input_files) if self.process == Process.sequential: result = await self._arun_sequential_process() @@ -1243,6 +1245,8 @@ async def run_crew() -> None: for after_callback in self.after_kickoff_callbacks: result = after_callback(result) + if inspect.isawaitable(result): + result = await result result = self._post_kickoff(result) diff --git a/lib/crewai/src/crewai/crews/utils.py b/lib/crewai/src/crewai/crews/utils.py index 6faa0e8c28..c2f2083d83 100644 --- a/lib/crewai/src/crewai/crews/utils.py +++ b/lib/crewai/src/crewai/crews/utils.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import inspect from collections.abc import Callable, Coroutine, Iterable, Mapping from typing import TYPE_CHECKING, Any @@ -246,50 +247,41 @@ def _extract_files_from_inputs(inputs: dict[str, Any]) -> dict[str, Any]: return files -def prepare_kickoff( - crew: Crew, - inputs: dict[str, Any] | None, - input_files: dict[str, FileInput] | None = None, -) -> dict[str, Any] | None: - """Prepare crew for kickoff execution. - - Handles before callbacks, event emission, task handler reset, input - interpolation, task callbacks, agent setup, and planning. - - Args: - crew: The crew instance to prepare. - inputs: Optional input dictionary to pass to the crew. - input_files: Optional dict of named file inputs for the crew. - - Returns: - The potentially modified inputs dictionary after before callbacks. - """ +def _begin_prepare_kickoff(crew: Crew) -> bool: + """Reset emission counters and return whether the crew is resuming from a checkpoint.""" from crewai.events.base_events import reset_emission_counter - from crewai.events.event_bus import crewai_event_bus from crewai.events.event_context import ( get_current_parent_id, reset_last_event_id, ) - from crewai.events.types.crew_events import CrewKickoffStartedEvent resuming = crew.checkpoint_kickoff_event_id is not None - if not resuming and get_current_parent_id() is None: reset_emission_counter() reset_last_event_id() + return resuming - normalized: dict[str, Any] | None = None - if inputs is not None: - if not isinstance(inputs, Mapping): - raise TypeError( - f"inputs must be a dict or Mapping, got {type(inputs).__name__}" - ) - normalized = dict(inputs) - for before_callback in crew.before_kickoff_callbacks: - if normalized is None: - normalized = {} - normalized = before_callback(normalized) +def _normalize_inputs(inputs: dict[str, Any] | None) -> dict[str, Any] | None: + """Validate and convert inputs to a plain dict.""" + if inputs is None: + return None + if not isinstance(inputs, Mapping): + raise TypeError( + f"inputs must be a dict or Mapping, got {type(inputs).__name__}" + ) + return dict(inputs) + + +def _finish_prepare_kickoff( + crew: Crew, + normalized: dict[str, Any] | None, + input_files: dict[str, FileInput] | None, + resuming: bool, +) -> dict[str, Any] | None: + """Shared post-callback kickoff preparation: events, inputs, agents, planning.""" + from crewai.events.event_bus import crewai_event_bus + from crewai.events.types.crew_events import CrewKickoffStartedEvent if resuming and crew._kickoff_event_id: if crew.verbose: @@ -358,6 +350,68 @@ def prepare_kickoff( return normalized +def prepare_kickoff( + crew: Crew, + inputs: dict[str, Any] | None, + input_files: dict[str, FileInput] | None = None, +) -> dict[str, Any] | None: + """Prepare crew for kickoff execution. + + Handles before callbacks, event emission, task handler reset, input + interpolation, task callbacks, agent setup, and planning. + + Args: + crew: The crew instance to prepare. + inputs: Optional input dictionary to pass to the crew. + input_files: Optional dict of named file inputs for the crew. + + Returns: + The potentially modified inputs dictionary after before callbacks. + """ + resuming = _begin_prepare_kickoff(crew) + normalized = _normalize_inputs(inputs) + + for before_callback in crew.before_kickoff_callbacks: + if normalized is None: + normalized = {} + normalized = before_callback(normalized) + + return _finish_prepare_kickoff(crew, normalized, input_files, resuming) + + +async def aprepare_kickoff( + crew: Crew, + inputs: dict[str, Any] | None, + input_files: dict[str, FileInput] | None = None, +) -> dict[str, Any] | None: + """Async variant of prepare_kickoff that awaits coroutine before-callbacks. + + Used by akickoff to support both sync and async before_kickoff_callbacks + without blocking the event loop. + + Args: + crew: The crew instance to prepare. + inputs: Optional input dictionary to pass to the crew. + input_files: Optional dict of named file inputs for the crew. + + Returns: + The potentially modified inputs dictionary after before callbacks. + """ + resuming = _begin_prepare_kickoff(crew) + normalized = _normalize_inputs(inputs) + + for before_callback in crew.before_kickoff_callbacks: + if normalized is None: + normalized = {} + result = before_callback(normalized) + if inspect.isawaitable(result): + normalized = await result + else: + normalized = result + + return _finish_prepare_kickoff(crew, normalized, input_files, resuming) + + class StreamingContext: """Container for streaming state and holders used during crew execution.""" diff --git a/lib/crewai/tests/crew/test_async_crew.py b/lib/crewai/tests/crew/test_async_crew.py index aaaffa64fb..c92b58a763 100644 --- a/lib/crewai/tests/crew/test_async_crew.py +++ b/lib/crewai/tests/crew/test_async_crew.py @@ -1,5 +1,6 @@ """Tests for async crew execution.""" +import asyncio import pytest from unittest.mock import AsyncMock, MagicMock, patch @@ -221,6 +222,132 @@ def after_callback(result: CrewOutput) -> CrewOutput: assert callback_called + @pytest.mark.asyncio + @patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock) + async def test_akickoff_calls_async_before_callbacks( + self, mock_execute: AsyncMock, test_agent: Agent + ) -> None: + """async before_kickoff_callback should be awaited and its return value used.""" + async def async_before(inputs: dict) -> dict: + await asyncio.sleep(0) + return {**inputs, "injected": True} + + task = Task( + description="Test task for {topic}", + expected_output="Test output", + agent=test_agent, + ) + crew = Crew( + agents=[test_agent], + tasks=[task], + verbose=False, + before_kickoff_callbacks=[async_before], + ) + + mock_output = TaskOutput( + description="Test task for async", + raw="Task result", + agent="Test Agent", + ) + mock_execute.return_value = mock_output + + await crew.akickoff(inputs={"topic": "async"}) + + assert crew._inputs is not None + assert crew._inputs.get("injected") is True + + @pytest.mark.asyncio + @patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock) + async def test_akickoff_calls_async_after_callbacks( + self, mock_execute: AsyncMock, test_agent: Agent + ) -> None: + """async after_kickoff_callback should be awaited; result must remain CrewOutput.""" + callback_called = False + + async def async_after(result: CrewOutput) -> CrewOutput: + nonlocal callback_called + await asyncio.sleep(0) + callback_called = True + return result + + task = Task( + description="Test task", + expected_output="Test output", + agent=test_agent, + ) + crew = Crew( + agents=[test_agent], + tasks=[task], + verbose=False, + after_kickoff_callbacks=[async_after], + ) + + mock_output = TaskOutput( + description="Test task", + raw="Task result", + agent="Test Agent", + ) + mock_execute.return_value = mock_output + + result = await crew.akickoff() + + assert callback_called + assert isinstance(result, CrewOutput) + + @pytest.mark.asyncio + @patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock) + async def test_akickoff_mixed_sync_async_callbacks( + self, mock_execute: AsyncMock, test_agent: Agent + ) -> None: + """Mixed sync+async callbacks should execute in order, preserving the pipeline.""" + call_order: list[str] = [] + + def sync_before(inputs: dict) -> dict: + call_order.append("sync_before") + return {**inputs, "sync_before": True} + + async def async_before(inputs: dict) -> dict: + await asyncio.sleep(0) + call_order.append("async_before") + return {**inputs, "async_before": True} + + async def async_after(result: CrewOutput) -> CrewOutput: + await asyncio.sleep(0) + call_order.append("async_after") + return result + + def sync_after(result: CrewOutput) -> CrewOutput: + call_order.append("sync_after") + return result + + task = Task( + description="Test task for {topic}", + expected_output="Test output", + agent=test_agent, + ) + crew = Crew( + agents=[test_agent], + tasks=[task], + verbose=False, + before_kickoff_callbacks=[sync_before, async_before], + after_kickoff_callbacks=[async_after, sync_after], + ) + + mock_output = TaskOutput( + description="Test task for mixed", + raw="Task result", + agent="Test Agent", + ) + mock_execute.return_value = mock_output + + result = await crew.akickoff(inputs={"topic": "mixed"}) + + assert call_order == ["sync_before", "async_before", "async_after", "sync_after"] + assert crew._inputs is not None + assert crew._inputs.get("sync_before") is True + assert crew._inputs.get("async_before") is True + assert isinstance(result, CrewOutput) + class TestAsyncCrewKickoffForEach: """Tests for async crew kickoff_for_each methods."""