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..286c5dab0f 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 @@ -264,20 +265,29 @@ def prepare_kickoff( Returns: The potentially modified inputs dictionary after before callbacks. """ - 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 + return _prepare_kickoff_impl(crew, inputs, input_files, normalized_inputs=None) - 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() +async def aprepare_kickoff( + crew: Crew, + inputs: dict[str, Any] | None, + input_files: dict[str, FileInput] | None = None, +) -> dict[str, Any] | None: + """Async counterpart of :func:`prepare_kickoff`. + + Used by ``Crew.akickoff`` so that async ``before_kickoff_callbacks`` are + awaited instead of silently dropped, and any blocking work inside them + does not stall the event loop. Sync before-callbacks continue to work + unchanged. + 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: dict[str, Any] | None = None if inputs is not None: if not isinstance(inputs, Mapping): @@ -286,10 +296,82 @@ def prepare_kickoff( ) normalized = dict(inputs) + normalized = await _arun_before_kickoff_callbacks(crew, normalized) + + return _prepare_kickoff_impl( + crew, inputs, input_files, normalized_inputs=normalized + ) + + +def _run_before_kickoff_callbacks( + crew: Crew, normalized: dict[str, Any] | None +) -> dict[str, Any]: + """Run sync ``before_kickoff_callbacks``, returning the (possibly) new inputs.""" for before_callback in crew.before_kickoff_callbacks: if normalized is None: normalized = {} normalized = before_callback(normalized) + return normalized # type: ignore[return-value] + + +async def _arun_before_kickoff_callbacks( + crew: Crew, normalized: dict[str, Any] | None +) -> dict[str, Any] | None: + """Run ``before_kickoff_callbacks`` with async support. + + Awaits callbacks that return an awaitable (coroutine functions), matching + how ``task_callback`` and ``step_callback`` are handled in the async path. + Sync callbacks run unchanged. + """ + for before_callback in crew.before_kickoff_callbacks: + if normalized is None: + normalized = {} + result = before_callback(normalized) + if inspect.isawaitable(result): + result = await result + normalized = result + return normalized + + +def _prepare_kickoff_impl( + crew: Crew, + inputs: dict[str, Any] | None, + input_files: dict[str, FileInput] | None, + *, + normalized_inputs: dict[str, Any] | None, +) -> dict[str, Any] | None: + """Shared body of :func:`prepare_kickoff` and :func:`aprepare_kickoff`. + + When ``normalized_inputs`` is ``None`` the sync before-callbacks are run + here; otherwise the caller (the async path) has already applied them and + they are skipped to avoid double-execution. + """ + 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() + + if normalized_inputs is None: + 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) + + normalized = _run_before_kickoff_callbacks(crew, normalized) + else: + normalized = normalized_inputs if resuming and crew._kickoff_event_id: if crew.verbose: diff --git a/lib/crewai/tests/crew/test_async_crew.py b/lib/crewai/tests/crew/test_async_crew.py index aaaffa64fb..8ddd32f4ab 100644 --- a/lib/crewai/tests/crew/test_async_crew.py +++ b/lib/crewai/tests/crew/test_async_crew.py @@ -1,5 +1,7 @@ """Tests for async crew execution.""" +from typing import Any + import pytest from unittest.mock import AsyncMock, MagicMock, patch @@ -221,6 +223,160 @@ 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_callbacks( + self, mock_execute: AsyncMock, test_agent: Agent + ) -> None: + """akickoff must await async after_kickoff_callbacks. + + Regression test: previously an async after-callback was called without + an awaitable check, so its coroutine was never awaited and the crew + result was silently replaced with the coroutine object. + """ + callback_awaited = False + + async def after_callback(result: CrewOutput) -> CrewOutput: + nonlocal callback_awaited + callback_awaited = 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=[after_callback], + ) + + mock_output = TaskOutput( + description="Test task", + raw="Task result", + agent="Test Agent", + ) + mock_execute.return_value = mock_output + + result = await crew.akickoff() + + assert callback_awaited + assert isinstance(result, CrewOutput) + + @pytest.mark.asyncio + @patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock) + async def test_akickoff_applies_async_after_callback_result( + self, mock_execute: AsyncMock, test_agent: Agent + ) -> None: + """An async after-callback's returned CrewOutput replaces the result.""" + marker = TaskOutput( + description="replaced", + raw="async callback result", + agent="Test Agent", + ) + + async def after_callback(_result: CrewOutput) -> CrewOutput: + return marker + + task = Task( + description="Test task", + expected_output="Test output", + agent=test_agent, + ) + crew = Crew( + agents=[test_agent], + tasks=[task], + verbose=False, + after_kickoff_callbacks=[after_callback], + ) + + mock_execute.return_value = TaskOutput( + description="original", + raw="original result", + agent="Test Agent", + ) + + result = await crew.akickoff() + + assert result.raw == marker.raw + + @pytest.mark.asyncio + @patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock) + async def test_akickoff_awaits_async_before_callbacks( + self, mock_execute: AsyncMock, test_agent: Agent + ) -> None: + """akickoff must await async before_kickoff_callbacks.""" + callback_awaited = False + + async def before_callback(inputs: dict | None) -> dict: + nonlocal callback_awaited + callback_awaited = 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=[before_callback], + ) + + mock_execute.return_value = TaskOutput( + description="Test task", + raw="Task result", + agent="Test Agent", + ) + + await crew.akickoff() + + assert callback_awaited + + @pytest.mark.asyncio + @patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock) + async def test_akickoff_applies_async_before_callback_inputs( + self, mock_execute: AsyncMock, test_agent: Agent + ) -> None: + """An async before-callback's returned inputs flow into the crew.""" + captured: dict[str, Any] = {} + + async def before_callback(inputs: dict | None) -> dict[str, Any]: + merged = dict(inputs or {}) + merged["injected"] = "from-async-before" + return merged + + task = Task( + description="Test task for {injected}", + expected_output="Test output", + agent=test_agent, + ) + crew = Crew( + agents=[test_agent], + tasks=[task], + verbose=False, + before_kickoff_callbacks=[before_callback], + ) + + async def capture_inputs(*args: Any, **kwargs: Any) -> TaskOutput: + captured.update(crew._inputs or {}) + return TaskOutput( + description="Test task", + raw="Task result", + agent="Test Agent", + ) + + mock_execute.side_effect = capture_inputs + + await crew.akickoff(inputs={"base": "value"}) + + assert captured.get("injected") == "from-async-before" + assert captured.get("base") == "value" + class TestAsyncCrewKickoffForEach: """Tests for async crew kickoff_for_each methods."""