fix: #6481 - Add async support for 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
@@ -1266,7 +1267,8 @@ async def run_crew() -> None:
 
         runtime_scope = crewai_event_bus._enter_runtime_scope()
         try:
-            inputs = prepare_kickoff(self, inputs, input_files)
+            from crewai.crews.utils import aprepare_kickoff
+            inputs = await aprepare_kickoff(self, inputs, input_files)
 
             if self.process == Process.sequential:
                 result = await self._arun_sequential_process()
@@ -1279,6 +1281,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
 
@@ -379,6 +380,142 @@ def prepare_kickoff(
     return normalized
 
 
+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 native async kickoff (akickoff).
+
+    Handles before callbacks, event emission, task handler reset, input
+    interpolation, task callbacks, agent setup, and planning. Supports
+    async awaitable 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
+
+    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()
+
+    from crewai.hooks.contexts import ExecutionStartContext, InputContext
+    from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+    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)
+
+    # ``inputs`` aliases the same object as ``payload`` (not a fresh ``{}`` from
+    # ``or``) so in-place edits to either survive read-back, per the context
+    # contract. ``None`` inputs are preserved rather than coerced to ``{}``.
+    start_ctx = ExecutionStartContext(
+        crew=crew,
+        inputs=normalized if normalized is not None else {},
+        payload=normalized,
+    )
+    # Pairing flags: EXECUTION_END fires (once) only for executions whose
+    # EXECUTION_START actually dispatched, including on the failure path.
+    crew._execution_start_dispatched = False
+    crew._execution_end_dispatched = False
+    dispatch(InterceptionPoint.EXECUTION_START, start_ctx)
+    crew._execution_start_dispatched = True
+    normalized = start_ctx.payload
+
+    for before_callback in crew.before_kickoff_callbacks:
+        if normalized is None:
+            normalized = {}
+        cb_result = before_callback(normalized)
+        if inspect.isawaitable(cb_result):
+            normalized = await cb_result
+        else:
+            normalized = cb_result
+
+    input_ctx = InputContext(
+        crew=crew,
+        inputs=normalized if normalized is not None else {},
+        payload=normalized,
+    )
+    dispatch(InterceptionPoint.INPUT, input_ctx)
+    normalized = input_ctx.payload
+
+    if resuming and crew._kickoff_event_id:
+        if crew.verbose:
+            from crewai.events.utils.console_formatter import ConsoleFormatter
+
+            fmt = ConsoleFormatter(verbose=True)
+            content = fmt.create_status_content(
+                "Resuming from Checkpoint",
+                crew.name or "Crew",
+                "bright_magenta",
+                ID=str(crew.id),
+            )
+            fmt.print_panel(
+                content, "\U0001f504 Resuming from Checkpoint", "bright_magenta"
+            )
+    else:
+        started_event = CrewKickoffStartedEvent(crew_name=crew.name, inputs=normalized)
+        crew._kickoff_event_id = started_event.event_id
+        future = crewai_event_bus.emit(crew, started_event)
+        if future is not None:
+            try:
+                future.result()
+            except Exception:  # noqa: S110
+                pass
+
+    crew._task_output_handler.reset()
+    crew._logging_color = "bold_purple"
+
+    _flow_files = baggage.get_baggage("flow_input_files")
+    flow_files: dict[str, Any] = _flow_files if isinstance(_flow_files, dict) else {}
+
+    if normalized is not None:
+        unpacked_files = _extract_files_from_inputs(normalized)
+
+        all_files = {**flow_files, **(input_files or {}), **unpacked_files}
+        if all_files:
+            store_files(crew.id, all_files)
+
+        crew._inputs = normalized
+        crew._interpolate_inputs(normalized)
+    else:
+        all_files = {**flow_files, **(input_files or {})}
+        if all_files:
+            store_files(crew.id, all_files)
+    crew._set_tasks_callbacks()
+    crew._set_allow_crewai_trigger_context_for_first_task()
+
+    agents_to_setup: list[BaseAgent] = list(crew.agents)
+    seen_agent_ids: set[int] = {id(agent) for agent in agents_to_setup}
+    for task in crew.tasks:
+        if task.agent is not None and id(task.agent) not in seen_agent_ids:
+            agents_to_setup.append(task.agent)
+            seen_agent_ids.add(id(task.agent))
+
+    setup_agents(
+        crew,
+        agents_to_setup,
+        crew.embedder,
+        crew.function_calling_llm,
+        crew.step_callback,
+    )
+
+    if crew.planning:
+        crew._handle_crew_planning()
+
+    return normalized
+
+
 class StreamingContext:
     """Container for streaming state and holders used during crew execution."""
 

lib/crewai/src/crewai/tools/tool_usage.py

@@ -896,12 +896,12 @@ def _original_tool_calling(
 
         except Exception:
             if raise_error:
-                raise
+                raise ToolUsageError(f"{I18N_DEFAULT.errors('tool_arguments_error')}")
             return ToolUsageError(f"{I18N_DEFAULT.errors('tool_arguments_error')}")
 
         if not isinstance(arguments, dict):
             if raise_error:
-                raise
+                raise ToolUsageError(f"{I18N_DEFAULT.errors('tool_arguments_error')}")
             return ToolUsageError(f"{I18N_DEFAULT.errors('tool_arguments_error')}")
 
         return ToolCalling(

lib/crewai/tests/crew/test_async_callbacks.py

@@ -0,0 +1,194 @@
+"""Tests for async callbacks support in akickoff."""
+
+import asyncio
+import pytest
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from crewai.agent import Agent
+from crewai.crew import Crew
+from crewai.task import Task
+from crewai.crews.crew_output import CrewOutput
+from crewai.tasks.task_output import TaskOutput
+
+
+@pytest.fixture
+def test_agent() -> Agent:
+    """Create a test agent."""
+    return Agent(
+        role="Test Agent",
+        goal="Test goal",
+        backstory="Test backstory",
+        llm="gpt-4o-mini",
+        verbose=False,
+    )
+
+
+@pytest.fixture
+def test_task(test_agent: Agent) -> Task:
+    """Create a test task."""
+    return Task(
+        description="Test task description",
+        expected_output="Test expected output",
+        agent=test_agent,
+    )
+
+
+class TestAsyncCallbacksSupport:
+    """Tests for async callback support in akickoff."""
+
+    @pytest.mark.asyncio
+    @patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock)
+    async def test_akickoff_calls_async_before_callback(
+        self, mock_execute: AsyncMock, test_agent: Agent
+    ) -> None:
+        """Test that async before_callback is awaited in aprepare_kickoff."""
+        callback_result = {"called": False}
+
+        async def async_before_callback(inputs: dict | None) -> dict[str, Any]:
+            callback_result["called"] = True
+            await asyncio.sleep(0.01)
+            return inputs or {}
+
+        task = Task(
+            description="Test task for {topic}",
+            expected_output="Expected output for {topic}",
+            agent=test_agent,
+        )
+        crew = Crew(
+            agents=[test_agent],
+            tasks=[task],
+            verbose=False,
+            before_kickoff_callbacks=[async_before_callback],
+        )
+
+        mock_output = TaskOutput(
+            description="Test task for AI",
+            raw="Task result about AI",
+            agent="Test Agent",
+        )
+        mock_execute.return_value = mock_output
+
+        result = await crew.akickoff(inputs={"topic": "AI"})
+
+        assert callback_result["called"], "Async before callback was not called"
+        assert result is not None
+        assert isinstance(result, CrewOutput)
+
+    @pytest.mark.asyncio
+    @patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock)
+    async def test_akickoff_calls_async_after_callback(
+        self, mock_execute: AsyncMock, test_agent: Agent
+    ) -> None:
+        """Test that async after_callback is awaited in akickoff."""
+        callback_result = {"called": False, "received": None}
+
+        async def async_after_callback(result: CrewOutput) -> CrewOutput:
+            nonlocal callback_result
+            callback_result["called"] = True
+            callback_result["received"] = result
+            await asyncio.sleep(0.01)
+            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
+
+        result = await crew.akickoff()
+
+        assert callback_result["called"], "Async after callback was not called"
+        assert callback_result["received"] is not None
+        assert isinstance(callback_result["received"], CrewOutput)
+
+    @pytest.mark.asyncio
+    @patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock)
+    async def test_akickoff_mixed_sync_and_async_callbacks(
+        self, mock_execute: AsyncMock, test_agent: Agent
+    ) -> None:
+        """Test that mixed sync and async callbacks work together."""
+        sync_result = {"called": False}
+        async_result = {"called": False, "received": None}
+
+        def sync_before_callback(inputs: dict | None) -> dict:
+            sync_result["called"] = True
+            return inputs or {}
+
+        async def async_after_callback(result: CrewOutput) -> CrewOutput:
+            nonlocal async_result
+            async_result["called"] = True
+            async_result["received"] = result
+            await asyncio.sleep(0.01)
+            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_callback],
+            after_kickoff_callbacks=[async_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 sync_result["called"], "Sync before callback was not called"
+        assert async_result["called"], "Async after callback was not called"
+        assert async_result["received"] is not None
+
+    @pytest.mark.asyncio
+    @patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock)
+    async def test_akickoff_empty_callbacks(
+        self, mock_execute: AsyncMock, test_agent: Agent
+    ) -> None:
+        """Test that empty callbacks list still works normally."""
+        task = Task(
+            description="Test task",
+            expected_output="Test output",
+            agent=test_agent,
+        )
+        crew = Crew(
+            agents=[test_agent],
+            tasks=[task],
+            verbose=False,
+            before_kickoff_callbacks=[],
+            after_kickoff_callbacks=[],
+        )
+
+        mock_output = TaskOutput(
+            description="Test task",
+            raw="Task result",
+            agent="Test Agent",
+        )
+        mock_execute.return_value = mock_output
+
+        result = await crew.akickoff()
+
+        assert result is not None
+        assert isinstance(result, CrewOutput)
+        assert result.raw == "Task result"