fix: support async callables in akickoff before/after callbacks#6494
Open
ashusnapx wants to merge 1 commit into
Open
fix: support async callables in akickoff before/after callbacks#6494ashusnapx wants to merge 1 commit into
ashusnapx wants to merge 1 commit into
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
🚧 Files skipped from review as they are similar to previous changes (3)
📝 WalkthroughWalkthroughAdds async kickoff preparation in ChangesAsync before/after kickoff callback support
Sequence Diagram(s)sequenceDiagram
participant Caller
participant Crew as Crew.akickoff
participant Utils as aprepare_kickoff
participant BeforeCB as before_kickoff_callbacks
participant AfterCB as after_kickoff_callbacks
Caller->>Crew: await akickoff(inputs)
Crew->>Utils: await aprepare_kickoff(crew, inputs)
Utils->>BeforeCB: call before_callback(normalized)
BeforeCB-->>Utils: result or awaitable
Utils->>Utils: await result if isawaitable
Utils-->>Crew: normalized inputs
Crew->>AfterCB: call after_callback(result)
AfterCB-->>Crew: result or awaitable
Crew->>Crew: await result if isawaitable
Crew-->>Caller: CrewOutput
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
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.
ashusnapx
force-pushed
the
feat/async-callback-fix
branch
from
eb92e41 to
04c4ced
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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.
Inline comments:
In `@lib/crewai/src/crewai/crews/utils.py`:
- Around line 362-474: The kickoff preparation logic is duplicated between
prepare_kickoff and aprepare_kickoff, creating drift risk as shared behavior
changes over time. Extract the common control flow into a shared internal helper
used by both functions, and keep only the before_kickoff_callbacks execution
different by injecting a callback runner that either returns directly or awaits
awaitable results. Preserve the existing behavior around crewai_event_bus
emission, store_files, _set_tasks_callbacks, setup_agents, and planning in the
shared helper.
- Around line 423-431: The kickoff planning path is still blocking the event
loop because `aprepare_kickoff` ultimately calls `crew._handle_crew_planning()`
synchronously via `CrewPlanner._handle_crew_planning()` and `execute_sync()`.
Update the async kickoff flow in `utils.py` so planning is offloaded to a thread
or handled through a non-blocking async path before returning from `akickoff()`,
using the existing `aprepare_kickoff`, `crew._handle_crew_planning`, and
`CrewPlanner` hooks to locate the change.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 460ccfbf-0675-415f-a37f-8ca756db70d3
📒 Files selected for processing (3)
lib/crewai/src/crewai/crew.pylib/crewai/src/crewai/crews/utils.pylib/crewai/tests/crew/test_async_crew.py
Comment on lines
+362
to
+474
| 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 use with akickoff(). | ||
|
|
||
| Mirrors :func:`prepare_kickoff` but awaits async before-kickoff | ||
| callbacks so that coroutines are not silently discarded. | ||
|
|
||
| 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. | ||
| """ | ||
| 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() | ||
|
|
||
| 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) | ||
|
|
||
| for before_callback in crew.before_kickoff_callbacks: | ||
| if normalized is None: | ||
| normalized = {} | ||
| normalized = before_callback(normalized) | ||
| if inspect.isawaitable(normalized): | ||
| normalized = await normalized | ||
|
|
||
| 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 | ||
|
|
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy lift
Large duplicated function body between prepare_kickoff and aprepare_kickoff.
aprepare_kickoff Introduces the new async function aprepare_kickoff, duplicating the core kickoff preparation/control flow from prepare_kickoff while extending before_kickoff_callbacks handling to await callback results when they are awaitable. Aside from the two lines added at 406-407, the ~110-line body is a verbatim copy of prepare_kickoff (lines 250-359). Any future fix/change to the shared logic (event emission, file storage, task/agent setup, planning) now has to be manually mirrored in both functions, which is a real drift risk.
Consider extracting a shared internal helper (parameterized by an "await-if-awaitable" strategy or an injected callback runner) that both prepare_kickoff and aprepare_kickoff call, keeping only the callback-invocation loop distinct.
🤖 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/src/crewai/crews/utils.py` around lines 362 - 474, The kickoff
preparation logic is duplicated between prepare_kickoff and aprepare_kickoff,
creating drift risk as shared behavior changes over time. Extract the common
control flow into a shared internal helper used by both functions, and keep only
the before_kickoff_callbacks execution different by injecting a callback runner
that either returns directly or awaits awaitable results. Preserve the existing
behavior around crewai_event_bus emission, store_files, _set_tasks_callbacks,
setup_agents, and planning in the shared helper.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Fixes silent discarding of async callables in
Crew.akickoff().Closes #6481
Problem
Crew.akickoff()is a native async execution path, butbefore_kickoff_callbacksandafter_kickoff_callbackswere invoked synchronously with noinspect.isawaitablecheck. Async callables were silently discarded — their coroutine objects were never awaited.Fix
after_kickoff_callbacksinakickoff()inspect.isawaitablecheck +awaitbefore_kickoff_callbacksinakickoff()aprepare_kickoff()async variant that awaits async callbacksFollows the same pattern used by
task_callbackandstep_callbackthroughout the codebase:Files Changed
crew.pyimport inspect, fixed after_callback loop, importsaprepare_kickoffcrews/utils.pyaprepare_kickoff()async variant withinspect.isawaitablefor before_callbackstests/crew/test_async_crew.pyTests