Skip to content

Commit 3cfba59

Browse files
committed
fix: support async callables in akickoff before/after callbacks
Closes crewAIInc#6481 Crew.akickoff() is a native async execution path, but before_kickoff_callbacks and after_kickoff_callbacks were invoked synchronously with no awaitable check. Async callables were silently discarded — their coroutine objects were never awaited. - crew.py: Add inspect.isawaitable check in akickoff after_callback loop - crews/utils.py: Add aprepare_kickoff() async variant that awaits async before_kickoff_callbacks using inspect.isawaitable - crew.py: akickoff() now calls aprepare_kickoff() instead of prepare_kickoff() Follows the same pattern used by task_callback and step_callback. Tests: test_akickoff_awaits_async_after_callback, test_akickoff_awaits_async_before_callback, test_akickoff_mixed_sync_async_callbacks. All 14 async crew tests pass. Lint and mypy clean.
1 parent 289686a commit 3cfba59

3 files changed

Lines changed: 239 additions & 22 deletions

File tree

lib/crewai/src/crewai/crew.py

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
from concurrent.futures import Future
66
from copy import copy as shallow_copy
77
from hashlib import md5
8+
import inspect
89
import json
910
from pathlib import Path
1011
import re
@@ -67,6 +68,7 @@ def get_supported_content_types(provider: str, api: str | None = None) -> list[s
6768
from crewai.crews.crew_output import CrewOutput
6869
from crewai.crews.utils import (
6970
StreamingContext,
71+
aprepare_kickoff,
7072
check_conditional_skip,
7173
enable_agent_streaming,
7274
prepare_kickoff,
@@ -1230,7 +1232,7 @@ async def run_crew() -> None:
12301232

12311233
runtime_scope = crewai_event_bus._enter_runtime_scope()
12321234
try:
1233-
inputs = prepare_kickoff(self, inputs, input_files)
1235+
inputs = await aprepare_kickoff(self, inputs, input_files)
12341236

12351237
if self.process == Process.sequential:
12361238
result = await self._arun_sequential_process()
@@ -1243,6 +1245,8 @@ async def run_crew() -> None:
12431245

12441246
for after_callback in self.after_kickoff_callbacks:
12451247
result = after_callback(result)
1248+
if inspect.isawaitable(result):
1249+
result = await result
12461250

12471251
result = self._post_kickoff(result)
12481252

lib/crewai/src/crewai/crews/utils.py

Lines changed: 104 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44

55
import asyncio
66
from collections.abc import Callable, Coroutine, Iterable, Mapping
7+
import inspect
78
from typing import TYPE_CHECKING, Any
89

910
from opentelemetry import baggage
@@ -246,23 +247,40 @@ def _extract_files_from_inputs(inputs: dict[str, Any]) -> dict[str, Any]:
246247
return files
247248

248249

249-
def prepare_kickoff(
250+
def _run_before_callbacks(
250251
crew: Crew,
251-
inputs: dict[str, Any] | None,
252-
input_files: dict[str, FileInput] | None = None,
252+
normalized: dict[str, Any] | None,
253253
) -> dict[str, Any] | None:
254-
"""Prepare crew for kickoff execution.
254+
"""Run before_kickoff_callbacks synchronously.
255+
256+
Args:
257+
crew: The crew instance whose callbacks to run.
258+
normalized: Current normalized inputs (may be None).
259+
260+
Returns:
261+
The potentially modified inputs dictionary.
262+
"""
263+
for before_callback in crew.before_kickoff_callbacks:
264+
if normalized is None:
265+
normalized = {}
266+
normalized = before_callback(normalized)
267+
return normalized
255268

256-
Handles before callbacks, event emission, task handler reset, input
257-
interpolation, task callbacks, agent setup, and planning.
269+
270+
def _prepare_kickoff_common(
271+
crew: Crew,
272+
normalized: dict[str, Any] | None,
273+
input_files: dict[str, FileInput] | None,
274+
) -> dict[str, Any] | None:
275+
"""Shared setup logic for kickoff: event emission, file storage, agent setup.
258276
259277
Args:
260278
crew: The crew instance to prepare.
261-
inputs: Optional input dictionary to pass to the crew.
279+
normalized: Inputs after before callbacks have run.
262280
input_files: Optional dict of named file inputs for the crew.
263281
264282
Returns:
265-
The potentially modified inputs dictionary after before callbacks.
283+
The normalized inputs dictionary.
266284
"""
267285
from crewai.events.base_events import reset_emission_counter
268286
from crewai.events.event_bus import crewai_event_bus
@@ -278,19 +296,6 @@ def prepare_kickoff(
278296
reset_emission_counter()
279297
reset_last_event_id()
280298

281-
normalized: dict[str, Any] | None = None
282-
if inputs is not None:
283-
if not isinstance(inputs, Mapping):
284-
raise TypeError(
285-
f"inputs must be a dict or Mapping, got {type(inputs).__name__}"
286-
)
287-
normalized = dict(inputs)
288-
289-
for before_callback in crew.before_kickoff_callbacks:
290-
if normalized is None:
291-
normalized = {}
292-
normalized = before_callback(normalized)
293-
294299
if resuming and crew._kickoff_event_id:
295300
if crew.verbose:
296301
from crewai.events.utils.console_formatter import ConsoleFormatter
@@ -358,6 +363,84 @@ def prepare_kickoff(
358363
return normalized
359364

360365

366+
def prepare_kickoff(
367+
crew: Crew,
368+
inputs: dict[str, Any] | None,
369+
input_files: dict[str, FileInput] | None = None,
370+
) -> dict[str, Any] | None:
371+
"""Prepare crew for kickoff execution.
372+
373+
Handles before callbacks, event emission, task handler reset, input
374+
interpolation, task callbacks, agent setup, and planning.
375+
376+
Args:
377+
crew: The crew instance to prepare.
378+
inputs: Optional input dictionary to pass to the crew.
379+
input_files: Optional dict of named file inputs for the crew.
380+
381+
Returns:
382+
The potentially modified inputs dictionary after before callbacks.
383+
"""
384+
normalized = _normalize_inputs(crew, inputs)
385+
normalized = _run_before_callbacks(crew, normalized)
386+
return _prepare_kickoff_common(crew, normalized, input_files)
387+
388+
389+
def _normalize_inputs(
390+
crew: Crew, inputs: dict[str, Any] | None
391+
) -> dict[str, Any] | None:
392+
"""Normalize and validate inputs dict.
393+
394+
Args:
395+
crew: The crew instance.
396+
inputs: Optional input dictionary.
397+
398+
Returns:
399+
Normalized inputs or None.
400+
401+
Raises:
402+
TypeError: If inputs is not a dict or Mapping.
403+
"""
404+
if inputs is not None:
405+
if not isinstance(inputs, Mapping):
406+
raise TypeError(
407+
f"inputs must be a dict or Mapping, got {type(inputs).__name__}"
408+
)
409+
return dict(inputs)
410+
return None
411+
412+
413+
async def aprepare_kickoff(
414+
crew: Crew,
415+
inputs: dict[str, Any] | None,
416+
input_files: dict[str, FileInput] | None = None,
417+
) -> dict[str, Any] | None:
418+
"""Async version of prepare_kickoff for use with akickoff().
419+
420+
Mirrors :func:`prepare_kickoff` but awaits async before-kickoff
421+
callbacks so that coroutines are not silently discarded.
422+
423+
Args:
424+
crew: The crew instance to prepare.
425+
inputs: Optional input dictionary to pass to the crew.
426+
input_files: Optional dict of named file inputs for the crew.
427+
428+
Returns:
429+
The potentially modified inputs dictionary after before callbacks.
430+
"""
431+
normalized = _normalize_inputs(crew, inputs)
432+
# Run callbacks in async mode so awaitable results are resolved.
433+
result: dict[str, Any] | None = None
434+
for before_callback in crew.before_kickoff_callbacks:
435+
if result is None:
436+
result = normalized if normalized is not None else {}
437+
result = before_callback(result)
438+
if inspect.isawaitable(result):
439+
result = await result
440+
normalized = result
441+
return _prepare_kickoff_common(crew, normalized, input_files)
442+
443+
361444
class StreamingContext:
362445
"""Container for streaming state and holders used during crew execution."""
363446

lib/crewai/tests/crew/test_async_crew.py

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
"""Tests for async crew execution."""
22

3+
import asyncio
34
import pytest
45
from unittest.mock import AsyncMock, MagicMock, patch
56

@@ -221,6 +222,135 @@ def after_callback(result: CrewOutput) -> CrewOutput:
221222

222223
assert callback_called
223224

225+
@pytest.mark.asyncio
226+
@patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock)
227+
async def test_akickoff_awaits_async_after_callback(
228+
self, mock_execute: AsyncMock, test_agent: Agent
229+
) -> None:
230+
"""Test that akickoff correctly awaits async after_kickoff_callbacks."""
231+
callback_executed = False
232+
233+
async def async_after_callback(result: CrewOutput) -> CrewOutput:
234+
nonlocal callback_executed
235+
await asyncio.sleep(0) # simulate async work
236+
callback_executed = True
237+
return result
238+
239+
task = Task(
240+
description="Test task",
241+
expected_output="Test output",
242+
agent=test_agent,
243+
)
244+
crew = Crew(
245+
agents=[test_agent],
246+
tasks=[task],
247+
verbose=False,
248+
after_kickoff_callbacks=[async_after_callback],
249+
)
250+
251+
mock_output = TaskOutput(
252+
description="Test task",
253+
raw="Task result",
254+
agent="Test Agent",
255+
)
256+
mock_execute.return_value = mock_output
257+
258+
await crew.akickoff()
259+
260+
assert callback_executed
261+
262+
@pytest.mark.asyncio
263+
@patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock)
264+
async def test_akickoff_awaits_async_before_callback(
265+
self, mock_execute: AsyncMock, test_agent: Agent
266+
) -> None:
267+
"""Test that akickoff correctly awaits async before_kickoff_callbacks."""
268+
callback_executed = False
269+
270+
async def async_before_callback(inputs: dict | None) -> dict:
271+
nonlocal callback_executed
272+
await asyncio.sleep(0) # simulate async work
273+
callback_executed = True
274+
return inputs or {}
275+
276+
task = Task(
277+
description="Test task",
278+
expected_output="Test output",
279+
agent=test_agent,
280+
)
281+
crew = Crew(
282+
agents=[test_agent],
283+
tasks=[task],
284+
verbose=False,
285+
before_kickoff_callbacks=[async_before_callback],
286+
)
287+
288+
mock_output = TaskOutput(
289+
description="Test task",
290+
raw="Task result",
291+
agent="Test Agent",
292+
)
293+
mock_execute.return_value = mock_output
294+
295+
await crew.akickoff()
296+
297+
assert callback_executed
298+
299+
@pytest.mark.asyncio
300+
@patch("crewai.task.Task.aexecute_sync", new_callable=AsyncMock)
301+
async def test_akickoff_mixed_sync_async_callbacks(
302+
self, mock_execute: AsyncMock, test_agent: Agent
303+
) -> None:
304+
"""Test akickoff with a mix of sync and async callbacks."""
305+
call_order = []
306+
307+
def sync_before(inputs: dict | None) -> dict:
308+
call_order.append("sync_before")
309+
return inputs or {}
310+
311+
async def async_before(inputs: dict | None) -> dict:
312+
call_order.append("async_before")
313+
await asyncio.sleep(0)
314+
return inputs or {}
315+
316+
def sync_after(result: CrewOutput) -> CrewOutput:
317+
call_order.append("sync_after")
318+
return result
319+
320+
async def async_after(result: CrewOutput) -> CrewOutput:
321+
call_order.append("async_after")
322+
await asyncio.sleep(0)
323+
return result
324+
325+
task = Task(
326+
description="Test task",
327+
expected_output="Test output",
328+
agent=test_agent,
329+
)
330+
crew = Crew(
331+
agents=[test_agent],
332+
tasks=[task],
333+
verbose=False,
334+
before_kickoff_callbacks=[sync_before, async_before],
335+
after_kickoff_callbacks=[sync_after, async_after],
336+
)
337+
338+
mock_output = TaskOutput(
339+
description="Test task",
340+
raw="Task result",
341+
agent="Test Agent",
342+
)
343+
mock_execute.return_value = mock_output
344+
345+
await crew.akickoff()
346+
347+
assert call_order == [
348+
"sync_before",
349+
"async_before",
350+
"sync_after",
351+
"async_after",
352+
]
353+
224354

225355
class TestAsyncCrewKickoffForEach:
226356
"""Tests for async crew kickoff_for_each methods."""

0 commit comments

Comments
 (0)