The crew must end with at most one asynchronous task" When Using Multiple Async Tasks in crewAI

[Arslanex]

I'm experiencing an issue with the crewAI library when trying to include multiple asynchronous tasks within a single Crew. According to the official documentation and examples, it seems possible to have multiple asynchronous tasks in one crew. However, I'm encountering an error that suggests otherwise.

pydantic_core._pydantic_core.ValidationError: 1 validation error for Crew
 The crew must end with at most one asynchronous task. 
[type=async_task_count, input_value={'agents': [Agent(role='Venue Coordinator', ...)], 'verbose': True}, input_type=dict]

• I'm using the crewAI library to automate event planning tasks, following the example provided in the official documentation and the lesson "Automate Event Planning."
• I've defined multiple agents and tasks, with two tasks set to run asynchronously (async_execution=True).
• Despite following the documentation, I'm receiving an error when attempting to run the code.

import os
import asyncio
from crewai import Agent, Crew, Task
from crewai_tools import ScrapeWebsiteTool, SerperDevTool
from pydantic import BaseModel
from langchain_groq import ChatGroq

# Set up API keys
os.environ["GROQ_API_KEY"] = "YOUR_GROQ_API_KEY"
groq_api_key = os.environ.get("GROQ_API_KEY")
if groq_api_key is None:
    raise ValueError("GROQ_API_KEY not found. Please check your .env file.")

# Define the LLM model
llm = ChatGroq(
    model="groq/mixtral-8x7b-32768",
    temperature=0.0,
    max_retries=2,
    api_key=groq_api_key
)

# Initialize tools
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()

# Agent 1: Venue Coordinator
venue_coordinator = Agent(
    role="Venue Coordinator",
    goal="Identify and book an appropriate venue based on event requirements",
    tools=[search_tool, scrape_tool],
    verbose=True,
    backstory=(
        "With a keen sense of space and understanding of event logistics, "
        "you excel at finding and securing the perfect venue that fits the event's theme, "
        "size, and budget constraints."
    ),
    llm=llm
)

# Agent 2: Logistics Manager
logistics_manager = Agent(
    role='Logistics Manager',
    goal="Manage all logistics for the event including catering and equipment",
    tools=[search_tool, scrape_tool],
    verbose=True,
    backstory=(
        "Organized and detail-oriented, you ensure that every logistical aspect of the event "
        "from catering to equipment setup is flawlessly executed to create a seamless experience."
    ),
    llm=llm
)

# Agent 3: Marketing and Communications Agent
marketing_communications_agent = Agent(
    role="Marketing and Communications Agent",
    goal="Effectively market the event and communicate with participants",
    tools=[search_tool, scrape_tool],
    verbose=True,
    backstory=(
        "Creative and communicative, you craft compelling messages and engage with potential attendees "
        "to maximize event exposure and participation."
    ),
    llm=llm
)

# Define a Pydantic model for venue details
class VenueDetails(BaseModel):
    name: str
    address: str
    capacity: int
    booking_status: str

# Define tasks
venue_task = Task(
    description="Find a venue in {event_city} that meets criteria for {event_topic}.",
    expected_output="All the details of a specifically chosen venue you found to accommodate the event.",
    human_input=True,
    output_json=VenueDetails,
    output_file="venue_details.json",
    agent=venue_coordinator
)

logistics_task = Task(
    description="Coordinate catering and equipment for an event with {expected_participants} participants on {tentative_date}.",
    expected_output="Confirmation of all logistics arrangements including catering and equipment setup.",
    human_input=True,
    async_execution=True,
    agent=logistics_manager
)

marketing_task = Task(
    description="Promote the {event_topic} aiming to engage at least {expected_participants} potential attendees.",
    expected_output="Report on marketing activities and attendee engagement formatted as markdown.",
    async_execution=True,
    output_file="marketing_report.md",
    agent=marketing_communications_agent
)

# Define the crew with agents and tasks
event_management_crew = Crew(
    agents=[venue_coordinator, logistics_manager, marketing_communications_agent],
    tasks=[venue_task, logistics_task, marketing_task],
    verbose=True
)

# Event details
event_details = {
    'event_topic': "Tech Innovation Conference",
    'event_description': "A gathering of tech innovators and industry leaders to explore future technologies.",
    'event_city': "Istanbul",
    'tentative_date': "2025-05-15",
    'expected_participants': 2000,
    'budget': 100000,
    'venue_type': "Conference Hall"
}

# Kickoff the crew
result = event_management_crew.kickoff(inputs=event_details)

Problem Details:

• When running this code, I receive the error message about only allowing at most one asynchronous task.
• This is confusing because the official documentation and lesson examples include multiple asynchronous tasks within a single crew.
• The only difference I can identify is that the examples use gpt-4-turbo, while I'm using groq/mixtral-8x7b-32768 via the GROQ API.

Steps I've Taken:

• Library Versions: Updated crewai and crewai_tools to the latest versions.
  Documentation Review: Re-examined the "Automate Event Planning" lesson and official documentation. Confirmed that multiple asynchronous tasks are used in a single crew in the examples.

Questions:

• Is there a limitation in the crewai library that restricts a crew to only one asynchronous task, despite the documentation suggesting otherwise?
• Could this error be related to the library version, or is there an additional configuration needed to allow multiple asynchronous tasks in a crew?
• How can I modify my code to match the examples and successfully include multiple asynchronous tasks within a single crew?

Any help or guidance would be greatly appreciated!
Thank you in advance.

[KeepALifeUS]

This constraint exists because crewAI needs to collect results before the crew completes. Here's how to work around it:

Understanding the Constraint

The final task(s) must be synchronous so the crew can:

1. Wait for all results
2. Return a complete output
3. Not leave dangling async operations

Pattern: Parallel Async + Sync Aggregator

from crewai import Task, Crew, Agent

# These can run in parallel
logistics_task = Task(
    description="Handle logistics...",
    async_execution=True,  # Runs async
    agent=logistics_agent
)

marketing_task = Task(
    description="Handle marketing...",
    async_execution=True,  # Runs async
    agent=marketing_agent
)

# Final task MUST be sync to collect results
aggregation_task = Task(
    description="Combine logistics and marketing results into final report",
    async_execution=False,  # Sync - collects async results
    agent=coordinator_agent,
    context=[logistics_task, marketing_task]  # Waits for both
)

crew = Crew(
    agents=[logistics_agent, marketing_agent, coordinator_agent],
    tasks=[logistics_task, marketing_task, aggregation_task]  # Ends with sync
)

Why This Works

• Async tasks run in parallel (performance win)
• Sync aggregator waits for all to complete
• Crew returns complete results

Alternative: External State

If you truly need multiple async endpoints:

state = {"results": {}}

# Each async task writes to state
def on_logistics_complete(output):
    state["results"]["logistics"] = output

def on_marketing_complete(output):
    state["results"]["marketing"] = output

This decouples result collection from the crew execution.