CrewAI Tutorial (Python): building conditional routing for beginners

This tutorial shows you how to build a CrewAI flow that routes work conditionally in Python, so one path handles simple requests and another handles complex ones. You need this when a single agent is not enough and you want the system to choose the right next step based on the input or an intermediate result.

What You'll Need

• •Python 3.10+
• •crewai
• •crewai-tools
• •An OpenAI API key
• •Basic familiarity with:
  • •Agent
  • •Task
  • •Crew
  • •Flow
• •A terminal and a virtual environment

Install the packages:

pip install crewai crewai-tools

Set your API key:

export OPENAI_API_KEY="your-key-here"

Step-by-Step

1. •Create a small routing flow with two branches.
   The idea is simple: classify the request first, then route to either a fast answer path or a deeper analysis path.

from crewai.flow.flow import Flow, start, listen
from pydantic import BaseModel

class RoutingState(BaseModel):
    request: str = ""
    route: str = ""
    result: str = ""

class RoutingFlow(Flow[RoutingState]):
    @start()
    def classify_request(self):
        text = self.state.request.lower()
        if any(word in text for word in ["summary", "simple", "quick"]):
            self.state.route = "simple"
        else:
            self.state.route = "complex"
        return self.state.route

2. •Add conditional listeners for each branch.
   Each listener only runs when its upstream step returns the matching value. This is the part beginners usually miss: the return value from the routing step controls which branch executes.

from crewai import Agent, Task, Crew, Process

class RoutingFlow(Flow[RoutingState]):
    @start()
    def classify_request(self):
        text = self.state.request.lower()
        if any(word in text for word in ["summary", "simple", "quick"]):
            self.state.route = "simple"
        else:
            self.state.route = "complex"
        return self.state.route

    @listen("simple")
    def handle_simple(self):
        agent = Agent(
            role="Concise Assistant",
            goal="Answer briefly and directly",
            backstory="You give short, accurate responses.",
            verbose=False,
        )
        task = Task(
            description=f"Give a short answer to: {self.state.request}",
            expected_output="A concise answer",
            agent=agent,
        )
        crew = Crew(agents=[agent], tasks=[task], process=Process.sequential)
        self.state.result = crew.kickoff().raw

    @listen("complex")
    def handle_complex(self):
        agent = Agent(
            role="Analyst",
            goal="Provide a structured analysis",
            backstory="You break down problems carefully.",
            verbose=False,
        )
        task = Task(
            description=f"Analyze this request in detail: {self.state.request}",
            expected_output="A structured analysis",
            agent=agent,
        )
        crew = Crew(agents=[agent], tasks=[task], process=Process.sequential)
        self.state.result = crew.kickoff().raw

3. •Add an output step so you can inspect the final state.
   This keeps debugging easy and makes it obvious which route was taken.

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

class RoutingFlow(Flow[RoutingState]):
    @start()
    def classify_request(self):
        text = self.state.request.lower()
        if any(word in text for word in ["summary", "simple", "quick"]):
            self.state.route = "simple"
        else:
            self.state.route = "complex"
        return self.state.route

    @listen("simple")
    def handle_simple(self):
        agent = Agent(role="Concise Assistant", goal="Answer briefly and directly", backstory="You give short responses.")
        task = Task(description=f"Give a short answer to: {self.state.request}", expected_output="A concise answer", agent=agent)
        crew = Crew(agents=[agent], tasks=[task])
        self.state.result = crew.kickoff().raw

    @listen("complex")
    def handle_complex(self):
        agent = Agent(role="Analyst", goal="Provide a structured analysis", backstory="You break down problems carefully.")
        task = Task(description=f"Analyze this request in detail: {self.state.request}", expected_output="A structured analysis", agent=agent)
        crew = Crew(agents=[agent], tasks=[task])
        self.state.result = crew.kickoff().raw

    @end()
    def show_result(self):
        return {
            "route": self.state.route,
            "result": self.state.result,
        }

4. •Run the flow with different inputs.
   Use one test case that should hit the simple branch and another that should hit the complex branch.

if __name__ == "__main__":
    flow1 = RoutingFlow()
    flow1.state.request = "Give me a quick summary of CrewAI flows."
    print(flow1.kickoff())

    flow2 = RoutingFlow()
    flow2.state.request = "Explain how conditional routing works in production systems."
    print(flow2.kickoff())

5. •Make the routing more explicit with helper methods if you want cleaner production code.
   In real projects, I keep branch logic separate from agent execution so it is easier to test and change later.

def is_simple_request(text: str) -> bool:
    keywords = ["summary", "simple", "quick"]
    lowered = text.lower()
    return any(word in lowered for word in keywords)

class RoutingFlow(Flow[RoutingState]):
    @start()
    def classify_request(self):
        self.state.route = "simple" if is_simple_request(self.state.request) else "complex"
        return self.state.route

Testing It

Run the script and confirm that each input lands on the right branch. The first example should set route to simple, while the second should set it to complex. If both branches work, you will see different outputs from different agents based on the same flow structure.

If something fails, check three things first:

• •Your OPENAI_API_KEY is set correctly
• •You installed a recent version of crewai
• •Your model access is valid for whatever default LLM CrewAI is using

For debugging, print self.state.route inside each listener and verify that your classifier returns exactly "simple" or "complex".

Next Steps

• •Add a third branch for escalation, such as "compliance_review" or "human_handoff"
• •Replace keyword routing with an LLM-based classifier task
• •Persist flow state to a database so routing decisions are auditable

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.