fix(crew): await async before/after_kickoff_callbacks in akickoff

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
@@ -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:

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