CrewAI Flows: Event-Driven Agent Orchestration Explained

What Are CrewAI Flows?

CrewAI Flows represent a paradigm shift from rigid, linear pipelines toward event-driven agent orchestration. Instead of prescribing every step in advance, you design a system where agents react to events — user inputs, API callbacks, internal state changes, or messages from other agents. This allows complex, adaptive workflows that can branch, loop, wait for human approval, or trigger new chains of activity on the fly.

At its core, a Flow is a directed, reactive graph of operations. You define listeners that wait for specific events, and when those events fire, the flow executes associated tasks using your CrewAI agents. Flows can start automatically, respond to webhooks, or be triggered programmatically. This makes them ideal for chatbots, autonomous support systems, data pipelines that need conditional rerouting, and any scenario where the next step depends on dynamic runtime information.

Why Event-Driven Orchestration Matters

Traditional sequential crews (e.g., crew.kickoff()) work beautifully for straightforward, fixed workflows. But real-world applications often require:

• Human-in-the-loop approvals: pause execution, wait for a human decision, then resume.
• Conditional branching: based on agent output or external data, route to different tasks.
• External triggers: start a workflow when a webhook arrives or a file is uploaded.
• Long-running conversations: maintain state across multiple interactions.
• Parallel execution and timeouts: react to multiple events concurrently.

CrewAI Flows address all of these. By decoupling the when from the what, you build systems that feel alive — they listen, respond, and adapt. This leads to more robust, maintainable, and scalable AI agent applications.

Core Concepts: Flows, Events, and Listeners

Before diving into code, let’s clarify the building blocks of the Flow framework:

• Flow – A class that encapsulates the entire event-driven orchestration. It holds state, manages agent crews, and wires together listeners and routers.
• Event – A named signal that represents something happening (e.g., "user_message", "task_completed", "approval_granted"). Events can carry payload data.
• Listener – A method decorated with @listen that executes whenever a specific event is emitted. Listeners can trigger agents, modify state, or emit new events.
• Router – A special listener that inspects an event’s payload and conditionally triggers one of several downstream listeners. Routers enable branching logic.
• Start – An entry point method decorated with @start that fires when the flow begins (e.g., on application startup or when a webhook invokes the flow).
• State – The flow can maintain a structured state (often a Pydantic model) that persists across events, allowing listeners to share context without tight coupling.

How to Use CrewAI Flows: A Step-by-Step Guide

Let’s build a concrete example: a customer support flow that listens for incoming queries, classifies them, routes to the appropriate specialized agent (tech support or billing), and then emits a final resolution event.

1. Installation and Setup

Ensure you have the latest CrewAI with Flow support:

pip install crewai[flow] -U

Set your API keys as needed (e.g., OPENAI_API_KEY). Then create a new Python file for your flow.

2. Define the State and Agents

First, model the state that the flow will carry across events. Then instantiate the agents that will handle tasks.

from pydantic import BaseModel
from crewai import Agent

class SupportState(BaseModel):
    user_id: str = ""
    query: str = ""
    category: str = ""
    resolution: str = ""
    history: list = []

# Specialized agents
tech_agent = Agent(
    role="Tech Support Specialist",
    goal="Resolve technical issues with clear, step-by-step guidance.",
    backstory="You are an expert in troubleshooting hardware and software.",
    allow_delegation=False,
    verbose=True
)

billing_agent = Agent(
    role="Billing Support Specialist",
    goal="Handle payment, invoice, and subscription questions.",
    backstory="You are a patient financial support expert.",
    allow_delegation=False,
    verbose=True
)

classifier_agent = Agent(
    role="Query Classifier",
    goal="Classify the user query as 'tech' or 'billing' based on intent.",
    backstory="You analyze support tickets and assign categories.",
    allow_delegation=False,
    verbose=True
)

3. Create the Flow Class

Now define a Flow subclass. Use the @start decorator to capture the initial user query and emit an event, then use @listen to react.

from crewai.flow import Flow, start, listen, router
from crewai import Task, Crew

class CustomerSupportFlow(Flow[SupportState]):
    
    @start("user_query_arrived")
    def receive_query(self, user_id: str, query: str):
        """Entry point triggered externally with user input."""
        self.state.user_id = user_id
        self.state.query = query
        print(f"Flow started for user {user_id}: {query}")
        # Emit an event that will be caught by the classifier listener
        self.emit("user_query_arrived", {"query": query})

    @listen("user_query_arrived")
    def classify_and_route(self, event_data: dict):
        """Classify the query, update state, and route to the appropriate handler."""
        # Create a task for classification
        classify_task = Task(
            description=f"Classify this support query: '{event_data['query']}'",
            expected_output="A single word: 'tech' or 'billing'.",
            agent=classifier_agent
        )
        # Execute using a one-shot crew
        crew = Crew(agents=[classifier_agent], tasks=[classify_task], verbose=True)
        result = crew.kickoff()
        category = result.strip().lower()
        self.state.category = category
        # Emit a new event carrying the category and the original query
        self.emit("query_classified", {"category": category, "query": event_data['query']})

    @router("query_classified")
    def route_by_category(self, event_data: dict):
        """Router that decides which listener should handle the classified query."""
        category = event_data["category"]
        if category == "tech":
            # Triggers the tech_listener method below
            return "tech_handler"
        elif category == "billing":
            return "billing_handler"
        else:
            # Fallback or error handler – here we just return None to stop
            print("Unknown category, stopping flow.")
            return None

    @listen("tech_handler")
    def handle_tech(self, event_data: dict):
        """Resolve tech queries using the tech agent."""
        task = Task(
            description=f"User query: {event_data['query']}. Provide a step-by-step technical solution.",
            expected_output="A detailed technical resolution.",
            agent=tech_agent
        )
        crew = Crew(agents=[tech_agent], tasks=[task], verbose=True)
        resolution = crew.kickoff()
        self.state.resolution = resolution
        self.emit("resolution_ready", {"resolution": resolution, "category": "tech"})

    @listen("billing_handler")
    def handle_billing(self, event_data: dict):
        """Resolve billing queries using the billing agent."""
        task = Task(
            description=f"User query: {event_data['query']}. Provide a clear billing resolution.",
            expected_output="A billing resolution with next steps.",
            agent=billing_agent
        )
        crew = Crew(agents=[billing_agent], tasks=[task], verbose=True)
        resolution = crew.kickoff()
        self.state.resolution = resolution
        self.emit("resolution_ready", {"resolution": resolution, "category": "billing"})

    @listen("resolution_ready")
    def finalize(self, event_data: dict):
        """Log the resolution and notify the user (e.g., store in DB, send message)."""
        print(f"Resolution ready: {event_data['resolution']}")
        self.state.history.append({
            "query": self.state.query,
            "category": event_data["category"],
            "resolution": event_data["resolution"]
        })
        # Here you could also emit a final event to an external system
        print("Flow complete. State preserved.")

4. Running the Flow

You can kick off the flow programmatically by calling the start method. In a production setting, you’d likely trigger it from a web framework route.

if __name__ == "__main__":
    flow = CustomerSupportFlow()
    # Simulate an incoming user query
    flow.receive_query(user_id="user123", query="My laptop won't turn on. What should I do?")
    # Alternatively, you can use flow.kickoff() if no explicit entry args are needed

When receive_query runs, it emits "user_query_arrived". The classifier listens, classifies the query, and emits "query_classified". The router then directs to "tech_handler" or "billing_handler", and finally the resolution event triggers finalize. The entire chain is event-driven, asynchronous in spirit (though synchronous execution by default), and easily extensible.

Advanced Event Handling: Human Approval and Conditional Loops

Flows shine when you introduce waiting periods or external approval steps. Imagine you want to send the resolution back to the user for confirmation before closing the ticket. You can design a listener that emits an event like "awaiting_confirmation" and then pause until a human clicks “Approve”. This is achieved by having an external system (e.g., a UI) call a method on the flow instance to resume.

@listen("resolution_ready")
def request_approval(self, event_data: dict):
    # Instead of finalizing directly, emit an event for human review
    self.emit("pending_approval", {"resolution": event_data["resolution"]})
    # The flow now waits; no further automatic listeners fire until approval

@listen("approval_granted")
def approved_resolution(self, event_data: dict):
    # Called externally when a human approves
    self.state.history.append({...})
    print("Human approved. Finalizing.")

Similarly, you can implement loops by re-emitting an event with updated data, causing listeners to re-execute until a condition is met. Use state fields to track iterations and prevent infinite loops.

Best Practices for Production Flows

• Keep State Immutable where possible – Treat the Pydantic state as the single source of truth. Modify it only within listeners to avoid race conditions if you later introduce async execution.
• Design Small, Focused Listeners – Each listener should do one thing: classify, resolve, notify. This makes debugging and testing straightforward.
• Use Explicit Event Names – Choose descriptive, verb-noun event names (e.g., "payment_received", "error_timeout"). Avoid ambiguous names like "step1_done".
• Handle Errors Gracefully – Attach error listeners (@listen("error")) or wrap listener logic in try-except to emit an "error_occurred" event with context. Never let an exception silently halt the flow.
• Test in Isolation – Unit test individual listeners by calling them directly with mock event payloads. CrewAI Flows are just Python classes; they are fully testable.
• Log Event Chains – Emit a "flow_trace" event at each step containing a correlation ID. This gives you an audit log of the entire flow execution for debugging.
• Plan for Idempotency – If an event can be delivered more than once (e.g., from a webhook retry), ensure listeners handle it safely using state checks.
• Secure External Triggers – When flows are started via webhooks, validate payloads and authenticate callers before emitting internal events.

Conclusion

CrewAI Flows elevate agent orchestration from static scripts to dynamic, reactive systems. By embracing event-driven design, you can build AI applications that gracefully handle the unpredictability of real-world interactions — waiting for humans, branching on intent, and adapting on the fly. The pattern of @start, @listen, and @router gives you a powerful yet intuitive framework to compose complex workflows while keeping each piece simple and testable. Whether you're creating a customer support bot, an automated content pipeline, or a multi-step approval engine, CrewAI Flows provide the flexibility and control you need to deliver robust, production-grade agent systems.