fix: support async before/after_kickoff_callbacks in akickoff

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,
@@ -1243,7 +1245,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()
@@ -1256,6 +1258,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)
 

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."""
 

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."""