fix: support async callables in akickoff before/after callbacks

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)
 

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

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