diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index fda456be40..3b5f8bb188 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -5,6 +5,7 @@ from concurrent.futures import Future from copy import copy as shallow_copy from hashlib import md5 +import inspect import json from pathlib import Path import re @@ -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..495b8f76c7 100644 --- a/lib/crewai/src/crewai/crews/utils.py +++ b/lib/crewai/src/crewai/crews/utils.py @@ -4,6 +4,7 @@ import asyncio from collections.abc import Callable, Coroutine, Iterable, Mapping +import inspect from typing import TYPE_CHECKING, Any from opentelemetry import baggage @@ -246,23 +247,40 @@ def _extract_files_from_inputs(inputs: dict[str, Any]) -> dict[str, Any]: return files -def prepare_kickoff( +def _run_before_callbacks( crew: Crew, - inputs: dict[str, Any] | None, - input_files: dict[str, FileInput] | None = None, + normalized: dict[str, Any] | None, ) -> dict[str, Any] | None: - """Prepare crew for kickoff execution. + """Run before_kickoff_callbacks synchronously. + + Args: + crew: The crew instance whose callbacks to run. + normalized: Current normalized inputs (may be None). + + Returns: + The potentially modified inputs dictionary. + """ + for before_callback in crew.before_kickoff_callbacks: + if normalized is None: + normalized = {} + normalized = before_callback(normalized) + return normalized - Handles before callbacks, event emission, task handler reset, input - interpolation, task callbacks, agent setup, and planning. + +def _prepare_kickoff_common( + crew: Crew, + normalized: dict[str, Any] | None, + input_files: dict[str, FileInput] | None, +) -> dict[str, Any] | None: + """Shared setup logic for kickoff: event emission, file storage, agent setup. Args: crew: The crew instance to prepare. - inputs: Optional input dictionary to pass to the crew. + normalized: Inputs after before callbacks have run. input_files: Optional dict of named file inputs for the crew. Returns: - The potentially modified inputs dictionary after before callbacks. + The normalized inputs dictionary. """ from crewai.events.base_events import reset_emission_counter from crewai.events.event_bus import crewai_event_bus @@ -278,19 +296,6 @@ def prepare_kickoff( reset_emission_counter() reset_last_event_id() - 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) - if resuming and crew._kickoff_event_id: if crew.verbose: from crewai.events.utils.console_formatter import ConsoleFormatter @@ -358,6 +363,84 @@ 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. + """ + normalized = _normalize_inputs(crew, inputs) + normalized = _run_before_callbacks(crew, normalized) + return _prepare_kickoff_common(crew, normalized, input_files) + + +def _normalize_inputs( + crew: Crew, inputs: dict[str, Any] | None +) -> dict[str, Any] | None: + """Normalize and validate inputs dict. + + Args: + crew: The crew instance. + inputs: Optional input dictionary. + + Returns: + Normalized inputs or None. + + Raises: + TypeError: If inputs is not a dict or Mapping. + """ + if inputs is not None: + if not isinstance(inputs, Mapping): + raise TypeError( + f"inputs must be a dict or Mapping, got {type(inputs).__name__}" + ) + return dict(inputs) + return None + + +async def aprepare_kickoff( + crew: Crew, + inputs: dict[str, Any] | None, + input_files: dict[str, FileInput] | None = None, +) -> dict[str, Any] | None: + """Async version of prepare_kickoff for use with akickoff(). + + Mirrors :func:`prepare_kickoff` but awaits async before-kickoff + callbacks so that coroutines are not silently discarded. + + 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. + """ + normalized = _normalize_inputs(crew, inputs) + # Run callbacks in async mode so awaitable results are resolved. + result: dict[str, Any] | None = None + for before_callback in crew.before_kickoff_callbacks: + if result is None: + result = normalized if normalized is not None else {} + result = before_callback(result) + if inspect.isawaitable(result): + result = await result + normalized = result + return _prepare_kickoff_common(crew, normalized, input_files) + + 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..1f4fab1ee4 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,135 @@ 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_awaits_async_after_callback( + self, mock_execute: AsyncMock, test_agent: Agent + ) -> None: + """Test that akickoff correctly awaits async after_kickoff_callbacks.""" + callback_executed = False + + async def async_after_callback(result: CrewOutput) -> CrewOutput: + nonlocal callback_executed + await asyncio.sleep(0) # simulate async work + callback_executed = 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_callback], + ) + + mock_output = TaskOutput( + description="Test task", + raw="Task result", + agent="Test Agent", + ) + mock_execute.return_value = mock_output + + await crew.akickoff() + + assert callback_executed + + @pytest.mark.asyncio + @patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock) + async def test_akickoff_awaits_async_before_callback( + self, mock_execute: AsyncMock, test_agent: Agent + ) -> None: + """Test that akickoff correctly awaits async before_kickoff_callbacks.""" + callback_executed = False + + async def async_before_callback(inputs: dict | None) -> dict: + nonlocal callback_executed + await asyncio.sleep(0) # simulate async work + callback_executed = True + return inputs or {} + + task = Task( + description="Test task", + expected_output="Test output", + agent=test_agent, + ) + crew = Crew( + agents=[test_agent], + tasks=[task], + verbose=False, + before_kickoff_callbacks=[async_before_callback], + ) + + mock_output = TaskOutput( + description="Test task", + raw="Task result", + agent="Test Agent", + ) + mock_execute.return_value = mock_output + + await crew.akickoff() + + assert callback_executed + + @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: + """Test akickoff with a mix of sync and async callbacks.""" + call_order = [] + + def sync_before(inputs: dict | None) -> dict: + call_order.append("sync_before") + return inputs or {} + + async def async_before(inputs: dict | None) -> dict: + call_order.append("async_before") + await asyncio.sleep(0) + return inputs or {} + + def sync_after(result: CrewOutput) -> CrewOutput: + call_order.append("sync_after") + return result + + async def async_after(result: CrewOutput) -> CrewOutput: + call_order.append("async_after") + await asyncio.sleep(0) + return result + + task = Task( + description="Test task", + 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=[sync_after, async_after], + ) + + mock_output = TaskOutput( + description="Test task", + raw="Task result", + agent="Test Agent", + ) + mock_execute.return_value = mock_output + + await crew.akickoff() + + assert call_order == [ + "sync_before", + "async_before", + "sync_after", + "async_after", + ] + class TestAsyncCrewKickoffForEach: """Tests for async crew kickoff_for_each methods."""