[BUG] before/after_kickoff_callbacks do not support async callables in akickoff

[magiccao]

Description

Summary

Crew.akickoff() is a native async execution path, but before_kickoff_callbacks and after_kickoff_callbacks are invoked without any awaitable check, silently ignoring async callables. This is inconsistent with how task_callback and step_callback are handled in the same code path.

There are two distinct problems:

1. after_kickoff_callbacks — simple oversight, easy to fix

In akickoff, the callbacks are called synchronously with no coroutine check:

# crew.py (akickoff path)
for after_callback in self.after_kickoff_callbacks:
    result = after_callback(result)   # if callback is async, result becomes a coroutine object

Compare this with how task_callback is correctly handled in the async task execution path:

# task.py (async path)
cb_result = self.callback(self.output)
if inspect.isawaitable(cb_result):
    await cb_result   # ✅ correctly awaited

If an async function is passed as after_kickoff_callbacks, result is silently replaced with an unawaited coroutine object. No error is raised, the result is lost, and the coroutine is never executed.

2. before_kickoff_callbacks — structural constraint

before_kickoff_callbacks are invoked inside prepare_kickoff(), which is a plain synchronous function shared by both kickoff() and akickoff():

# crews/utils.py
def prepare_kickoff(crew, inputs, input_files=None):
    ...
    for before_callback in crew.before_kickoff_callbacks:
        normalized = before_callback(normalized)   # no awaitable support

Since akickoff calls prepare_kickoff directly in the event loop (not via asyncio.to_thread), any synchronous I/O inside a before callback will block the event loop. Async callables are completely unsupported because prepare_kickoff has no async counterpart.

Note: kickoff_async is not affected by either issue. It wraps the entire synchronous kickoff in asyncio.to_thread, so all callbacks already run off the event loop.

Steps to Reproduce

import asyncio
from crewai import Agent, Task, Crew

async def async_after_callback(result):
    await asyncio.sleep(0)   # any async operation
    print("after callback executed")
    return result

agent = Agent(role="test", goal="test", backstory="test")
task = Task(description="Say hello", expected_output="A greeting")
crew = Crew(
    agents=[agent],
    tasks=[task],
    after_kickoff_callbacks=[async_after_callback],
)

result = asyncio.run(crew.akickoff())
# "after callback executed" is never printed
# result is a coroutine object, not CrewOutput

Expected behavior

• after_kickoff_callbacks in akickoff should detect and await async callables, consistent with task_callback behavior.
• before_kickoff_callbacks in akickoff should either support async callables (via an async version of prepare_kickoff) or document clearly that only synchronous, non-blocking functions are allowed.

Proposed Fix

For after_kickoff_callbacks in akickoff (minimal, low-risk):

for after_callback in self.after_kickoff_callbacks:
    result = after_callback(result)
    if inspect.isawaitable(result):
        result = await result

For before_kickoff_callbacks, introduce an async def aprepare_kickoff(...) that mirrors prepare_kickoff with await support, and call it from akickoff instead.

Operating System

macOS Sonoma

Python Version

3.12

crewAI Version

1.15.2

Virtual Environment

Venv

Additional context

• step_callback correctly uses _ainvoke_step_callback with await in the akickoff code path.
• task_callback correctly checks inspect.isawaitable in the async task execution path.
• The inconsistency suggests before/after_kickoff_callbacks were not updated when akickoff was introduced.
• SerializableCallable type alias (types/callback.py) accepts any Callable[..., Any], including coroutine functions, with no async-aware validation, making this failure mode completely silent.