CrewAI Tutorial (Python): building conditional routing for intermediate developers

This tutorial shows you how to build a CrewAI flow that routes work conditionally based on the output of an agent. You need this when one request can take multiple paths, like sending a claim to fraud review only if the evidence looks suspicious.

What You'll Need

• •Python 3.10 or newer
• •crewai
• •crewai-tools if you want external tools later
• •An OpenAI API key set as OPENAI_API_KEY
• •A terminal and a virtual environment
• •Basic familiarity with CrewAI agents, tasks, and crews

Step-by-Step

1. •Start with a clean project and install the package. Keep this minimal so the routing logic is easy to inspect and debug.

python -m venv .venv
source .venv/bin/activate
pip install crewai python-dotenv

2. •Set your API key in an environment file or shell session. CrewAI reads the model credentials from your environment, so keep secrets out of code.

from dotenv import load_dotenv
load_dotenv()

import os

if not os.getenv("OPENAI_API_KEY"):
    raise ValueError("OPENAI_API_KEY is not set")

3. •Define two agents and one task that classifies incoming text into a route label. The key here is to force a strict output format so your routing logic stays deterministic.

from crewai import Agent, Task, Crew, Process

triage_agent = Agent(
    role="Triage Analyst",
    goal="Classify requests into billing, fraud, or general support",
    backstory="You route customer cases based on intent and risk signals.",
    verbose=True,
)

triage_task = Task(
    description=(
        "Read the customer message and return exactly one label: "
        "'billing', 'fraud', or 'support'."
    ),
    expected_output="A single lowercase label.",
    agent=triage_agent,
)

4. •Run the triage crew and branch in Python based on the result. This is the conditional routing layer: CrewAI produces the classification, and your application decides what happens next.

def route_case(message: str) -> str:
    crew = Crew(
        agents=[triage_agent],
        tasks=[triage_task],
        process=Process.sequential,
        verbose=True,
    )

    result = crew.kickoff(inputs={"message": message})
    label = str(result).strip().lower()
    return label


message = "I noticed two duplicate charges on my card this week."
route = route_case(message)

if route == "billing":
    print("Send to billing workflow")
elif route == "fraud":
    print("Send to fraud review workflow")
else:
    print("Send to general support workflow")

5. •Add a second crew for each downstream path. In production, keep each branch focused on one job so you can test and scale them independently.

billing_agent = Agent(
    role="Billing Specialist",
    goal="Resolve invoice and payment issues",
    backstory="You handle charge disputes, invoices, and payment corrections.",
)

fraud_agent = Agent(
    role="Fraud Investigator",
    goal="Assess whether an account activity pattern looks suspicious",
    backstory="You investigate duplicate charges, unauthorized access, and anomalies.",
)

billing_task = Task(
    description="Summarize the billing issue and propose the next action.",
    expected_output="A short resolution plan.",
    agent=billing_agent,
)

fraud_task = Task(
    description="Summarize why this may be fraud and recommend escalation steps.",
    expected_output="A short investigation plan.",
    agent=fraud_agent,
)

6. •Wire the branches into a single executable script. This gives you a full conditional routing pattern without needing extra orchestration frameworks.

def run_billing_workflow():
    crew = Crew(agents=[billing_agent], tasks=[billing_task], process=Process.sequential)
    return crew.kickoff(inputs={"issue": "duplicate charge"})

def run_fraud_workflow():
    crew = Crew(agents=[fraud_agent], tasks=[fraud_task], process=Process.sequential)
    return crew.kickoff(inputs={"issue": "duplicate charge"})

if __name__ == "__main__":
    message = "I noticed two duplicate charges on my card this week."
    route = route_case(message)

    if route == "billing":
        print(run_billing_workflow())
    elif route == "fraud":
        print(run_fraud_workflow())
    else:
        print("Route to support queue")

Testing It

Run the script with three different inputs: one billing case, one fraud case, and one generic support request. You want to confirm that the first agent consistently returns only one of the allowed labels, because any extra text will break your branching logic.

If you see inconsistent labels, tighten the triage task prompt further or add post-processing that maps messy outputs back to known routes. In production systems, I also log both the raw model output and the final route so I can audit bad decisions later.

A good smoke test is to print route before executing downstream workflows. If routing works but downstream agents fail, isolate whether the issue is in classification, branch selection, or task execution.

Next Steps

• •Add structured outputs with Pydantic so your routing returns validated data instead of plain text.
• •Replace if/elif branching with a router function that supports more than three paths.
• •Combine this pattern with CrewAI memory or tools when routing depends on customer history or external systems

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.