Building a CrewAI crew for lead qualification

Lead qualification is a repetitive workflow that maps well to multi-agent orchestration. This crewai lead qualification crew example builds a three-agent crew that gathers signals, scores fit, and drafts a routing recommendation. You get runnable code and the exact verification step to confirm it works in your environment.

Step 1: Install dependencies and configure the LLM

CrewAI delegates text generation to a LangChain-compatible chat model. Install the framework, the OpenAI wrapper, and a dotenv loader.

pip install crewai langchain-openai python-dotenv

Create a .env file. If you run this at scale, point the base URL at an OpenAI-compatible gateway to get fallback across providers. For example, n4n.ai exposes one OpenAI-compatible endpoint covering 240+ models with automatic fallback when a provider is rate-limited or degraded.

OPENAI_API_KEY=sk-your-key
LLM_BASE_URL=https://api.n4n.ai/v1

Load it in Python. Keep temperature=0 for scoring and routing—non-determinism here creates noisy CRM data.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("LLM_BASE_URL"),  # None falls back to OpenAI
    max_retries=3,
)

Pin the CrewAI version in your requirements.txt. The agent and task APIs have shifted between minor releases; crewai==0.28.0 is a safe baseline for the code below.

Step 2: Define the agents

A qualification crew needs three distinct roles: a researcher who collects signals, a scoring agent who applies your ICP, and a writer who turns the score into a routing action. Keep each agent’s backstory tight; verbose personas waste tokens and dilute instruction adherence.

from crewai import Agent

researcher = Agent(
    role="Signal Researcher",
    goal="Find recent company signals for a lead (funding, hires, tech stack, job posts)",
    backstory="Senior OSINT analyst for B2B sales teams",
    llm=llm,
    verbose=False,
    allow_delegation=False,
)

scorer = Agent(
    role="Fit Scorer",
    goal="Score lead fit 1-10 against ICP: mid-market SaaS, engineering-led, recent funding",
    backstory="RevOps lead who has reviewed 2k+ inbound leads",
    llm=llm,
    verbose=False,
    allow_delegation=False,
)

writer = Agent(
    role="Routing Writer",
    goal="Draft a Slack-ready routing recommendation based on the fit score",
    backstory="Sales engineer who writes concise internal updates",
    llm=llm,
    verbose=False,
    allow_delegation=False,
)

Set allow_delegation=False. In a sequential crew, delegation spawns unplanned agent calls and makes token cost unpredictable. You want a deterministic graph for a production funnel.

Step 3: Define the tasks with dependencies

Tasks in CrewAI consume prior output via context. The scorer must wait for the researcher; the writer must wait for the scorer. Use {variables} for runtime inputs, but note that CrewAI fills them from the kickoff inputs or from upstream task outputs referenced by name.

from crewai import Task

research_task = Task(
    description="Research {lead_name} ({lead_domain}). Return 5 bullet signals with sources.",
    expected_output="Bulleted list of signals, each with a URL source.",
    agent=researcher,
)

score_task = Task(
    description="Using these signals: {signals}. Score fit 1-10 and justify in 2 sentences.",
    expected_output="JSON: {'score': int, 'rationale': str}",
    agent=scorer,
    context=[research_task],
)

routing_task = Task(
    description="Given score {score} and rationale {rationale}, write a 3-line Slack message.",
    expected_output="Plain text Slack message with @channel mention if score >= 8.",
    agent=writer,
    context=[score_task],
)

The expected_output field is not a parser—it’s a prompt hint. Enforce structure in your verification layer, not in CrewAI config. If you need strict schema, add a Pydantic model and validate after run.

Step 4: Assemble and run the crew

This crewai lead qualification crew example uses Process.sequential. Kickoff takes an inputs dict that fills the top-level task templates.

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, scorer, writer],
    tasks=[research_task, score_task, routing_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={
    "lead_name": "Acme Robotics",
    "lead_domain": "acmerobotics.com",
})

print("FINAL ROUTING MESSAGE:")
print(result)

Run with python main.py. The verbose flag logs each agent step, including the raw LLM calls. For a single lead, expect three sequential generations and a total latency roughly equal to their sum plus overhead.

If you later switch to Process.hierarchical, you must assign a manager_llm to the crew. For lead qualification, hierarchical adds no value—the dependency chain is linear.

Step 5: Verify the crew works

Don’t trust the happy path. Write an assertion script that checks structure, not LLM prose quality. Capture the raw score task output and validate it.

import json
import re

def verify_score(raw: str, min_score=1, max_score=10):
    match = re.search(r"\{.*\}", raw, re.DOTALL)
    if not match:
        raise ValueError("No JSON found in score task output")
    data = json.loads(match.group(0))
    assert isinstance(data["score"], int)
    assert min_score <= data["score"] <= max_score
    assert len(data["rationale"]) > 10
    print("Verification passed: score", data["score"])

# After kickoff:
# verify_score(crew.task_outputs['score_task'].raw)

Success criteria: the crew prints a Slack message, the score JSON parses, and the score falls in range. If verbose=True, you should see three agent executions without rate-limit errors. Add a pytest case that mocks the LLM with a FakeListChatModel to run this in CI without spending tokens.

Step 6: Harden for production volume

Running this crewai lead qualification crew example on a real inbound queue changes the failure modes. Providers throttle, models drift, and token cost becomes visible. Two concrete fixes:

First, the max_retries on ChatOpenAI handles transient 429s, but not sustained degradation. Second, centralize model routing. A gateway that honors client routing directives and forwards provider cache-control hints keeps your ChatOpenAI call unchanged while shifting traffic away from degraded models.

If you already point base_url at n4n.ai, you get per-token usage metering on every crew run, which lets you attribute cost to each lead without custom instrumentation. That’s the difference between a demo and a pipeline.

Tuning agent autonomy

CrewAI agents can call tools. For lead research, give the researcher a search tool instead of relying on parametric memory. Add tools=[serp_tool] to the agent constructor. The scorer should stay tool-less to avoid hallucinated ICP edits.

Avoiding context blowup

Sequential context passing duplicates text. For a three-agent crew it’s fine, but at ten agents, truncate the research_task output to the top 3 signals before passing. Use a callback on task completion to mutate context.

def trim_context(output):
    lines = output.splitlines()
    return "\n".join(lines[:4])

research_task.callback = trim_context

That keeps the scorer’s prompt small and your latency predictable.

Step 7: Extend the pattern

Swap the writer agent for a Router that calls your CRM API via a custom tool. The crew then closes the loop: research → score → create Opportunity. The same structure scales to account-based marketing or support triage. The key win is separation of concerns: each agent has one job, and the tasks enforce ordering.

Build the crewai lead qualification crew example once, then reuse the scaffold for any funnel stage that needs judgment plus structure. Add a fourth agent for “objection analysis” if your inbound leads come from reply threads. The framework cost is negligible; the orchestration clarity is what pays off when you debug a misrouted lead at 2 a.m.

Observability note

CrewAI does not ship a tracing backend. Emit verbose logs to JSON and ship them to your log store. Record the crew.task_outputs dict per run with the lead ID as a tag. Without that, you will not know which agent hallucinated when a score looks wrong two weeks later.