From e19cdbd3fca4f69d639d72f434502070a77a272a Mon Sep 17 00:00:00 2001 From: Diwak4r Date: Tue, 28 Jul 2026 12:39:29 +0545 Subject: [PATCH 1/2] Fix crewAI #6430: Replace bare raise with ToolUsageError in _original_tool_calling for consistent error handling --- lib/crewai/src/crewai/tools/tool_usage.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/lib/crewai/src/crewai/tools/tool_usage.py b/lib/crewai/src/crewai/tools/tool_usage.py index 573c709cb2..b4642014ea 100644 --- a/lib/crewai/src/crewai/tools/tool_usage.py +++ b/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( From 59d446bf2219a66d07fe19079f3c7c2cb2d2a405 Mon Sep 17 00:00:00 2001 From: Diwak4r Date: Tue, 28 Jul 2026 12:55:38 +0545 Subject: [PATCH 2/2] Fix crewAI #6481: Add async callbacks support in akickoff with proper awaiting of async callables and async prepare_kickoff for before_kickoff_callbacks --- lib/crewai/src/crewai/crew.py | 6 +- lib/crewai/src/crewai/crews/utils.py | 137 +++++++++++++ lib/crewai/tests/crew/test_async_callbacks.py | 194 ++++++++++++++++++ 3 files changed, 336 insertions(+), 1 deletion(-) create mode 100644 lib/crewai/tests/crew/test_async_callbacks.py diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index 0f77b2d224..ed95a74eb7 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/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) diff --git a/lib/crewai/src/crewai/crews/utils.py b/lib/crewai/src/crewai/crews/utils.py index 706c61e49b..7ffdeb7c37 100644 --- a/lib/crewai/src/crewai/crews/utils.py +++ b/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.""" diff --git a/lib/crewai/tests/crew/test_async_callbacks.py b/lib/crewai/tests/crew/test_async_callbacks.py new file mode 100644 index 0000000000..d97510d1d4 --- /dev/null +++ b/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"