CrewAI Flows: Event-Driven Agent Orchestration Explained

What Are CrewAI Flows?

CrewAI Flows is a powerful, event-driven orchestration framework built on top of the CrewAI library. While the core CrewAI library enables developers to compose static teams of AI agents that collaborate on predefined tasks, Flows introduces a dynamic layer: agents and tasks can now be triggered by events, messages, or state changes, creating reactive, long-running pipelines. Think of it as moving from a fixed script to a responsive operating system for your AI agents.

In a traditional CrewAI setup, you define a Task list and a Crew executes them sequentially or hierarchically. With Flows, you model your system as a directed graph of event handlers and listeners. Each node in the flow is a function (or a Crew execution) that activates when a specific event fires. This allows for infinite flexibility: branching based on user input, retrying failed steps, waiting for external webhooks, or orchestrating multiple crews over time.

Flows is part of the CrewAI enterprise‑grade toolkit and is included in the crewai package (version ≥0.51.0). It uses decorators like @flow.start() and @flow.listen() to wire together your logic. You can think of a Flow as a stateful, event‑driven state machine that coordinates agents, tools, and manual checkpoints.

Why Event-Driven Orchestration Matters

Modern AI applications rarely follow a simple, one‑shot request‑response pattern. Consider a customer support system: a ticket arrives, must be classified, escalated if urgent, responded to, and then archived—but only after the customer confirms satisfaction. A rigid sequential pipeline would block, fail, or require constant polling. Event‑driven orchestration solves this by:

• Reactivity: Agents react immediately to new information (new messages, API callbacks, time‑outs).
• Resilience: Decoupled components mean a failure in one agent doesn’t halt the entire system; you can retry or reroute events.
• Scalability: Each event handler can be scaled independently, and long‑running flows can span hours or days without holding resources.
• Flexibility: You can combine multiple Crews, human‑in‑the‑loop steps, and conditional logic naturally.

Without event‑driven design, you end up building fragile, monolithic chains that are hard to maintain. CrewAI Flows gives you the building blocks to create robust, production‑grade AI orchestration with minimal boilerplate.

Core Concepts of CrewAI Flows

Before diving into code, let's outline the key abstractions:

• Flow: A class that encapsulates the entire event‑driven process. It holds state, manages event dispatch, and persists data between handlers.
• State: A typed dictionary or Pydantic model that represents the shared data flowing through the pipeline. Every handler can read and update it.
• Decorators: @flow.start() marks the entry point (triggered when the flow is kicked off). @flow.listen() binds a handler to a specific event type or a previous step’s completion.
• Events: Implicit events are generated when a function decorated with @flow.listen() finishes; you can also define custom event types (e.g., "escalation_needed") for more granular control.

Handlers can be plain Python functions, async functions, or even executions of a full Crew. This means you can mix lightweight logic (classification, validation) with heavy agent‑driven reasoning seamlessly.

Getting Started: A Practical Example

Let's build a customer support flow that reacts to an incoming query, classifies it, and then either auto‑responds or escalates. We'll use CrewAI agents for the heavy lifting and Flows to wire them together.

1. Installation and Setup

Install CrewAI with the flows support (requires Python 3.10+):

pip install crewai>=0.51.0
# Also ensure you have an LLM provider configured, e.g. OpenAI
export OPENAI_API_KEY="your-key-here"

2. Define the State Model

We'll use a simple typed dictionary as the flow’s shared state. It holds the customer query, classification, and final response.

from typing import TypedDict, Optional

class SupportState(TypedDict):
    query: str
    classification: Optional[str]  # e.g., "billing", "technical", "general"
    priority: Optional[str]        # "normal", "urgent"
    response: Optional[str]
    escalated: bool

3. Create the Agent and Task Definitions

Define two agents: a classifier and a responder. They will be used inside the flow.

from crewai import Agent, Task, Crew

classifier = Agent(
    role="Ticket Classifier",
    goal="Accurately classify customer support tickets.",
    backstory="You are an expert at reading queries and identifying their category and urgency.",
    llm="gpt-4o"
)

responder = Agent(
    role="Customer Support Responder",
    goal="Write a helpful, concise reply to the customer.",
    backstory="You are a friendly support agent with deep product knowledge.",
    llm="gpt-4o"
)

def classify_task(query: str) -> Task:
    return Task(
        description=f"Classify this query: '{query}'. Output JSON with 'category' and 'priority'.",
        agent=classifier,
        expected_output="JSON string with category and priority."
    )

def respond_task(query: str, classification: str, priority: str) -> Task:
    return Task(
        description=(
            f"Write a response to this customer query: '{query}'.\n"
            f"Classification: {classification}, Priority: {priority}.\n"
            "If priority is urgent, apologize and promise a follow-up within 1 hour. Otherwise, provide a solution."
        ),
        agent=responder,
        expected_output="A polished email response."
    )

4. Build the Flow

Now we create the SupportFlow class, using decorators to define the event chain.

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

class SupportFlow(Flow[SupportState]):

    @start()
    def receive_query(self):
        # This is the entry point: we simulate receiving a query.
        # In production, this could be triggered by an API or message queue.
        self.state["query"] = "My internet keeps dropping every 10 minutes. I need a fix urgently!"
        print(f"📥 Received query: {self.state['query']}")
        # The flow automatically fires an event after this method finishes.

    @listen(receive_query)
    def classify_query(self):
        # This handler runs after 'receive_query' completes.
        query = self.state["query"]
        task = classify_task(query)
        # Execute a mini-crew with just the classifier agent
        crew = Crew(agents=[classifier], tasks=[task], verbose=False)
        result = crew.kickoff()
        # Parse the classification result
        try:
            data = json.loads(result)
        except Exception:
            # Fallback if output not JSON
            data = {"category": "general", "priority": "normal"}
        self.state["classification"] = data.get("category", "general")
        self.state["priority"] = data.get("priority", "normal")
        self.state["escalated"] = self.state["priority"] == "urgent"
        print(f"🔍 Classified as: {self.state['classification']} / {self.state['priority']}")

    @listen(classify_query)
    def handle_response(self):
        # Triggered after classification finishes.
        if self.state["escalated"]:
            print("🚨 Urgent ticket! Routing to human escalation queue.")
            # Here you could send a message to a human agent or another flow.
            self.state["response"] = "Escalation in progress. A specialist will contact you shortly."
        else:
            crew = Crew(agents=[responder], tasks=[
                respond_task(
                    self.state["query"],
                    self.state["classification"],
                    self.state["priority"]
                )
            ], verbose=False)
            result = crew.kickoff()
            self.state["response"] = result
            print(f"✅ Auto-response ready: {self.state['response'][:100]}...")

    @listen(handle_response)
    def finalize(self):
        # Final step: log the outcome and clear transient state.
        print("📋 Ticket processing complete.")
        print(f"Final response: {self.state['response']}")
        # Flow ends here; state can be persisted to a database.

5. Execute the Flow

Instantiate the flow and call kickoff(). The framework will invoke the handlers in order based on the event wiring.

if __name__ == "__main__":
    flow = SupportFlow()
    flow.kickoff()

# Expected console output:
# 📥 Received query: My internet keeps dropping...
# 🔍 Classified as: technical / urgent
# 🚨 Urgent ticket! Routing to human escalation queue.
# 📋 Ticket processing complete.
# Final response: Escalation in progress...

Notice how we didn’t explicitly call each function. The @start() and @listen() decorators define the event‑driven topology. When receive_query finishes, an internal event fires, and classify_query (which listens to it) automatically runs—and so on. This decoupling allows you to later insert new handlers, conditionally branch, or even add parallel listeners.

Advanced Usage: Custom Events and Branching

Flows go beyond simple linear chains. You can define custom events and trigger multiple listeners from a single point, or use conditions to branch.

Custom Events

Instead of listening to a specific function, you can emit and listen to string‑based events. This is perfect when you want to decouple the flow and allow multiple entry points.

from crewai.flow.flow import Flow, start, listen, emit_event

class MultiChannelFlow(Flow[SupportState]):

    @start()
    def start_from_api(self):
        # Entry point for API-triggered flows
        self.state["query"] = "Refund request for order #12345"
        emit_event("ticket_created")  # Fire a named event

    @listen("ticket_created")
    def classify_from_event(self):
        # This handler listens to the named event, not a function.
        print("Classifying ticket from event...")
        # classification logic...

    # You could also have an alternative entry point
    @start("manual_entry")
    def start_manual(self, query: str):
        self.state["query"] = query
        emit_event("ticket_created")  # Reuse the same event

Conditional Branching

You can use regular Python logic inside a handler to decide what to do next, including conditionally calling a sub‑flow or emitting different events.

@listen(classify_query)
def route_based_on_priority(self):
    if self.state["priority"] == "urgent":
        emit_event("escalation_needed")
    else:
        # Continue to normal response handler automatically
        # (by just finishing, the next listener will trigger)
        pass

@listen("escalation_needed")
def handle_escalation(self):
    # Dedicated handler for escalation
    self.state["response"] = "Escalated to Tier 2 support."

Best Practices for CrewAI Flows

To build production‑ready event‑driven agents, keep these guidelines in mind:

• Idempotent Handlers: Design handlers so that replaying an event doesn’t cause duplicate side effects. Use state flags (like escalated) to guard against double processing.
• Explicit State Updates: Always update the flow’s state dictionary explicitly. Don’t rely on implicit global variables; the flow object is your single source of truth.
• Error Handling: Wrap risky operations (LLM calls, external APIs) in try/except. You can emit error events like "error_occurred" and attach a dedicated error‑handling listener for retries or logging.
• Logging and Observability: Use Python’s logging module or CrewAI’s built‑in callbacks to trace event flow. The verbose=True flag on crews helps during development.
• Keep Handlers Focused: Each handler should do one small job (classify, respond, notify). This makes testing and debugging easier and encourages reuse.
• Test in Isolation: Test each handler individually by mocking the state and events. The crewai.flow module provides a flow.test() utility (experimental) to simulate events.
• Persistence: Flows can be long‑running. Save the state to a database or file between steps if the process spans hours. The flow.restore() method can reload state.
• Parallelism: Multiple listeners on the same event run concurrently. Be careful with shared state and use locks if necessary. Prefer immutable state updates (replace the whole dict) to avoid race conditions.

Conclusion

CrewAI Flows transforms agent orchestration from a static script into a dynamic, reactive system. By adopting event‑driven design, you can build AI applications that gracefully handle real‑world complexity—waiting for user input, reacting to changing conditions, and coordinating multiple specialized agents over time. The declarative decorator syntax keeps your code clean and maintainable, while the underlying event bus ensures robust, decoupled execution. Start small with a linear flow, then gradually introduce custom events, branching, and parallel listeners as your system evolves. The combination of CrewAI’s powerful agent primitives and Flows’ orchestration layer gives you a complete toolkit for the next generation of AI‑powered applications.