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

[nolanchic]

Description

Crew.akickoff() is a native async path, but before_kickoff_callbacks and after_kickoff_callbacks were invoked without any awaitable check — silently dropping async callables:

• after_kickoff_callbacks: result = after_callback(result) returned an unawaited coroutine when the callback was async, so the crew result was silently replaced with a coroutine object (never executed, never awaited).
• before_kickoff_callbacks: invoked inside the synchronous prepare_kickoff(), which has no async support — async before-callbacks never ran.

This is inconsistent with task_callback and step_callback, which correctly check inspect.isawaitable in the async path.

Note: this change is scoped to stopping the silent drop of async callbacks. It does not route synchronous callbacks off the event loop (e.g. via run_in_executor); a slow sync callback still runs inline, same as before. That is a separate, larger change.

Changes

• crews/utils.py: add async def aprepare_kickoff(...) — the async counterpart of prepare_kickoff. It awaits before-callbacks that return an awaitable (sync callbacks run unchanged). The shared body is extracted into _prepare_kickoff_impl(...) so the two entry points stay DRY and behavior is identical apart from callback awaiting.
• crew.py: akickoff now calls await aprepare_kickoff(...) instead of prepare_kickoff(...), and awaits each after-callback result when it is awaitable.

Behavior preserved

• kickoff() (sync) — unchanged, still uses prepare_kickoff.
• kickoff_async() — unchanged; it wraps the sync kickoff in asyncio.to_thread, so all callbacks already run off the event loop.
• Sync callbacks in akickoff — unchanged (an isawaitable check is a no-op for them).

Test plan

• test_akickoff_awaits_async_after_callbacks — async after-callback is awaited; result stays a CrewOutput (was a coroutine before).
• test_akickoff_applies_async_after_callback_result — async after-callback's returned CrewOutput replaces the result.
• test_akickoff_awaits_async_before_callbacks — async before-callback is awaited and runs.
• test_akickoff_applies_async_before_callback_inputs — async before-callback's returned inputs flow into the crew.
• uv run pytest lib/crewai/tests/crew/test_async_crew.py lib/crewai/tests/test_crew.py lib/crewai/tests/test_checkpoint.py → all pass (sync kickoff + kickoff_async callbacks unaffected).
• uv run ruff check lib/ → clean.
• uv run mypy lib/crewai/src/crewai/crew.py lib/crewai/src/crewai/crews/utils.py → clean (strict).

Fixes #6481.

Per .github/CONTRIBUTING.md, this PR is authored by an AI agent (Claude Code) and should carry the llm-generated label (external contributors can't self-apply labels).

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 61c78345-8128-464e-80f6-85b90f6ffcee

📥 Commits

Reviewing files that changed from the base of the PR and between 8e67c8d and 0e83479.

📒 Files selected for processing (2)

• lib/crewai/src/crewai/crews/utils.py
• lib/crewai/tests/crew/test_async_crew.py

🚧 Files skipped from review as they are similar to previous changes (2)

• lib/crewai/tests/crew/test_async_crew.py
• lib/crewai/src/crewai/crews/utils.py

---

📝 Walkthrough

Walkthrough

This PR adds async-aware handling for before_kickoff_callbacks and after_kickoff_callbacks in Crew.akickoff(). It introduces aprepare_kickoff, centralizes shared preparation logic, awaits async callbacks, and adds regression tests for callback execution, result replacement, and input propagation.

Changes

Async kickoff callback support

Layer / File(s)	Summary
Async kickoff preparation lib/crewai/src/crewai/crews/utils.py	prepare_kickoff delegates to shared preparation logic, while aprepare_kickoff awaits async before-callbacks and passes normalized inputs without rerunning them.
Crew.akickoff() callback wiring lib/crewai/src/crewai/crew.py	akickoff() awaits aprepare_kickoff(...) and awaits awaitable after_kickoff_callbacks results before returning.
Regression coverage lib/crewai/tests/crew/test_async_crew.py	Tests verify async callback execution, after-callback result replacement, merged before-callback inputs reaching task execution, single execution, and None input handling.

Suggested reviewers: vinibrsl

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 70.59% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly states the main change: awaiting async before/after kickoff callbacks in akickoff.
Description check	✅ Passed	The description matches the implemented changes and explains the async callback support and preserved behavior.
Linked Issues check	✅ Passed	The PR satisfies #6481 by awaiting async before/after kickoff callbacks, adding async kickoff prep, and preserving existing sync behavior.
Out of Scope Changes check	✅ Passed	The changes stay within the async callback support scope, with tests and a small sentinel/mypy fix supporting the same objective.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[ErenAta16]

This, #6494, and #6482 are three open PRs fixing the same bug (async before/after_kickoff_callbacks silently discarded in akickoff) — the crew.py hunk is identical across all three (same inspect.isawaitable guard around the after-callback loop, same aprepare_kickoff swap-in). Compared the three crews/utils.py implementations, since that is where they actually diverge:

This one and #6482 both extract shared helpers so prepare_kickoff (sync) and aprepare_kickoff (async) share the common setup/teardown logic (event emission, input interpolation, agent setup, planning) rather than duplicating it — this one via a single _prepare_kickoff_impl taking an optional normalized_inputs override, #6482 via three smaller helpers (_begin_prepare_kickoff/_normalize_inputs/_finish_prepare_kickoff). #6494 instead copy-pastes the full ~80-line body of prepare_kickoff into a new aprepare_kickoff, which works but means the two functions can drift out of sync on a future change to the shared logic.

One honest scope note that applies equally to all three: the descriptions (this PR included) mention IO-bound sync callbacks blocking the event loop as part of the motivation, but none of the three actually route sync callbacks through run_in_executor or similar — they only add the inspect.isawaitable check, so a slow sync callback still runs inline on the event loop same as before. Not a defect in this PR specifically, just flagging that the fix (correctly) narrows to "stop silently dropping async callables" rather than also solving the blocking-event-loop half of the stated problem.

[coderabbitai]

🧹 Nitpick comments (1)

lib/crewai/tests/crew/test_async_crew.py (1)

274-281: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Fix type hint mismatch in after_callback.

The type hint for after_callback declares the return type as CrewOutput, but it returns marker, which is defined as a TaskOutput. While this works at runtime because _post_kickoff handles it, it violates static typing and will cause warnings in type checkers like mypy.

Consider updating the return type hint to match the returned object.

♻️ Proposed fix

         marker = TaskOutput(
             description="replaced",
             raw="async callback result",
             agent="Test Agent",
         )

-        async def after_callback(_result: CrewOutput) -> CrewOutput:
+        async def after_callback(_result: Any) -> TaskOutput:
             return marker

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@lib/crewai/tests/crew/test_async_crew.py` around lines 274 - 281, Update the
after_callback function’s return annotation from CrewOutput to TaskOutput so it
matches the marker object returned by the callback and passes static type
checking.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@lib/crewai/tests/crew/test_async_crew.py`:
- Around line 274-281: Update the after_callback function’s return annotation
from CrewOutput to TaskOutput so it matches the marker object returned by the
callback and passes static type checking.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 58d5a7ed-ba2f-430d-af5a-ec31dd7f4ffa

📥 Commits

Reviewing files that changed from the base of the PR and between 99d5a42 and a76d3c7.

📒 Files selected for processing (3)

• lib/crewai/src/crewai/crew.py
• lib/crewai/src/crewai/crews/utils.py
• lib/crewai/tests/crew/test_async_crew.py

🚧 Files skipped from review as they are similar to previous changes (2)

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

[nolanchic]

Agree — this PR narrows to "stop silently dropping async callables", and routing sync callbacks through run_in_executor is deliberately out of scope, as you noted. The blocking-I/O line in the description was meant as motivation for making akickoff async-aware, not a claim that this change also fixes event-loop blocking.

I trimmed that framing from the description and added an explicit scope note that sync callbacks still run inline, so the description no longer overstates what the change does.

[ErenAta16]

Thanks for trimming the description, that reads accurately now.

Came back to look at the normalized_inputs sentinel more carefully and found a real edge case in it. None is doing double duty: it means both "caller did not pre-apply the callbacks" and, unavoidably, "caller pre-applied them and the result was None". _arun_before_kickoff_callbacks can genuinely return None, since it assigns normalized = result with whatever the callback handed back. So a before-callback that returns None makes aprepare_kickoff pass normalized_inputs=None, and _prepare_kickoff_impl then treats that as "not yet applied" and runs every before-callback a second time.

Simulated the exact control flow from this branch:

=== async path, callback mutates in place and returns None ===
  callback invocations: 2 (expected 1)
  result: None

=== sync path, same callback (control) ===
  callback invocations: 1 (expected 1)

=== async path, callback returns the dict ===
  callback invocations: 1 (expected 1)

So it's async-only, and only for callbacks that don't return the dict. That's not an exotic user error though, mutating the passed dict in place and falling off the end of the function is a natural thing to write, and it happens to work fine on kickoff today. The failure is also quiet: the callbacks just run twice, so anything non-idempotent in them (appending to a list, incrementing a counter, an API call) silently doubles.

Worth noting the async path is the one place where a double-run is most likely to be visibly wrong, since an async callback re-run through _run_before_kickoff_callbacks returns an un-awaited coroutine, which is the original bug this PR exists to fix.

A distinct sentinel avoids the ambiguity entirely:

_NOT_APPLIED = object()

def _prepare_kickoff_impl(crew, inputs, input_files, *, normalized_inputs=_NOT_APPLIED):
    if normalized_inputs is _NOT_APPLIED:
        normalized_inputs = _run_before_kickoff_callbacks(crew, _normalize_inputs(inputs))

Then None is just a value that round-trips, and the sync path keeps working by not passing the argument at all.

Alternatively, restructuring so the impl never runs callbacks (both prepare_kickoff and aprepare_kickoff apply their own, then call a shared tail) sidesteps the flag altogether, which is closer to what #6482 and #6494 do. That's a bigger diff, but it removes the "did someone already do this?" question rather than encoding the answer in a parameter.

Either way it'd be worth a test for the returns-None callback on the async path, since none of the three PRs currently covers it.

[nolanchic]

Good catch — confirmed. The None collision reproduces exactly as you described: an async before-callback that mutates the dict in place and returns None gets re-run, while the same callback on the sync path runs once.

Went with the sentinel:

_NOT_APPLIED = object()

def _prepare_kickoff_impl(crew, inputs, input_files, *, normalized_inputs=_NOT_APPLIED):
    if normalized_inputs is _NOT_APPLIED:
        normalized_inputs = _run_before_kickoff_callbacks(crew, _normalize_inputs(inputs))

prepare_kickoff no longer passes the argument, so the sync path is unchanged, and a pre-applied None now round-trips instead of triggering a re-run.

I considered the restructure (impl never runs callbacks; both callers apply their own and call a shared tail — what #6482 / #6494 do). It's cleaner in that it removes the flag entirely, but it's a meaningfully larger diff, and you'd just signed off on narrowing this PR's scope to "stop silently dropping async callables", so I kept the sentinel to stay minimal. Happy to switch to the restructure here if you'd rather have one canonical implementation rather than three variants — just say the word.

Added the two tests you flagged as missing across all three PRs:

• test_akickoff_before_callback_returning_none_runs_once — the async returns-None case. It spies _run_before_kickoff_callbacks and asserts it's never entered on the async path (call_count == 0), so the failure surfaces as the double-run itself rather than the downstream un-awaited-coroutine symptom. Verified it fails on the pre-sentinel code.
• test_akickoff_async_before_callback_returning_none_preserves_none — the empty-callback-list + inputs=None edge, where aprepare_kickoff legitimately produces None; confirms it round-trips without being treated as "not yet applied".

One clarification on the None semantics, since "round-trips" can be read two ways: the sentinel stops None from being misread as "not yet applied" (the bug you found), so a returns-None callback no longer causes a double-run. But a callback that returns None still doesn't flow its inputs to the crew on either path — the runner keeps whatever the callback returned, and returning the (possibly mutated) dict is required for inputs to reach task interpolation. That's the existing behavior on sync kickoff today, not something this PR changes; I called it out in the _prepare_kickoff_impl docstring rather than silently relying on it. If you'd rather have returns-None preserve the prior dict (via normalized = result if result is not None else normalized in the runner), that's a behavioral change I'd rather do in its own PR.

Also cleared a mypy --strict regression this branch had introduced (_prepare_kickoff_impl returned Any because normalized was last assigned from InterceptionContext.payload) — cast it back to dict | None on return.

Pushed in 0e83479.