CrewAI example: automating recruiting screening calls

This crewai recruiting screening automation example shows you how to build a multi-agent system that conducts initial phone screens, transcribes responses, extracts structured candidate data, and produces a ranked shortlist — all without a human in the loop until the final review. You’ll wire together a scheduler, an interviewer, a transcriber, an extractor, and a ranker, each with a single responsibility and clear handoffs.

Prerequisites

You need Python 3.10+, an OpenAI-compatible API key (for the LLM and Whisper), and a Twilio account for programmable voice. Install the core dependencies:

pip install crewai crewai-tools openai twilio python-dotenv pydantic

Create a .env file with your credentials:

OPENAI_API_KEY=sk-...
OPENAI_BASE_URL=https://api.openai.com/v1  # or your n4n.ai endpoint
TWILIO_ACCOUNT_SID=AC...
TWILIO_AUTH_TOKEN=...
TWILIO_PHONE_NUMBER=+15551234567

The base URL is where you can swap providers without changing code — useful when you want automatic fallback across 240+ models or need per-token metering for cost tracking.

Step 1: Define the crew structure

Start by modeling the data that flows between agents. Pydantic models give you validation and make the contract explicit.

# models.py
from pydantic import BaseModel, EmailStr
from typing import Optional
from enum import Enum

class SkillLevel(str, Enum):
    NONE = "none"
    BASIC = "basic"
    INTERMEDIATE = "intermediate"
    ADVANCED = "advanced"
    EXPERT = "expert"

class CandidateSkill(BaseModel):
    name: str
    level: SkillLevel
    years_experience: float

class ScreeningResult(BaseModel):
    candidate_id: str
    phone_number: str
    call_sid: str
    transcript: str
    skills: list[CandidateSkill]
    availability: str  # e.g., "2 weeks notice"
    salary_expectation: Optional[int]
    red_flags: list[str]
    overall_score: float  # 0-100
    summary: str

class JobRequirement(BaseModel):
    title: str
    required_skills: list[CandidateSkill]
    nice_to_have_skills: list[CandidateSkill]
    min_salary: int
    max_salary: int
    location: str
    remote_ok: bool

Save this as models.py. Every agent will import these types — no loose dictionaries passed around.

Step 2: Build the scheduler agent

The scheduler reads a CSV of candidates, initiates calls via Twilio, and hands off the call SID to the next agent. Keep it thin: one job, no LLM reasoning needed.

# agents/scheduler.py
from crewai import Agent
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field
import csv
import os
from twilio.rest import Client

class ScheduleCallsInput(BaseModel):
    candidates_csv: str = Field(description="Path to CSV with columns: candidate_id,phone_number,name")
    job_id: str = Field(description="Identifier for the job requisition")

class ScheduleCallsTool(BaseTool):
    name: str = "schedule_screening_calls"
    args_schema: Type[BaseModel] = ScheduleCallsInput

    def _run(self, candidates_csv: str, job_id: str) -> list[dict]:
        client = Client(os.getenv("TWILIO_ACCOUNT_SID"), os.getenv("TWILIO_AUTH_TOKEN"))
        twilio_number = os.getenv("TWILIO_PHONE_NUMBER")
        results = []

        with open(candidates_csv, newline="") as f:
            reader = csv.DictReader(f)
            for row in reader:
                call = client.calls.create(
                    to=row["phone_number"],
                    from_=twilio_number,
                    url=f"https://your-domain.com/twiml/screening?job_id={job_id}&candidate_id={row['candidate_id']}",
                    method="POST",
                    status_callback="https://your-domain.com/webhook/call-status",
                    status_callback_event=["completed", "failed", "busy", "no-answer"],
                    status_callback_method="POST",
                )
                results.append({
                    "candidate_id": row["candidate_id"],
                    "phone_number": row["phone_number"],
                    "name": row["name"],
                    "call_sid": call.sid,
                })
        return results

scheduler = Agent(
    role="Screening Scheduler",
    goal="Initiate screening calls for all candidates in the pipeline",
    backstory="You manage the outbound dialing campaign. You read the candidate list, place calls via Twilio, and pass call metadata downstream.",
    tools=[ScheduleCallsTool()],
    verbose=True,
    allow_delegation=False,
)

The TwiML endpoint (/twiml/screening) serves the interview script — see Step 3. The status callback writes call completion events to a queue or database so the transcriber knows when audio is ready.

Step 3: Create the interviewer TwiML endpoint

This isn’t a CrewAI agent — it’s the voice interface. Host it on FastAPI or Flask. It drives the conversation using a structured script and records the candidate’s responses.

# twiml_endpoint.py
from fastapi import FastAPI, Form, Request
from fastapi.responses import Response
from twilio.twiml.voice_response import VoiceResponse, Gather, Record
import os

app = FastAPI()

SCREENING_QUESTIONS = [
    "Please tell me about your most recent role and your primary responsibilities.",
    "What programming languages and frameworks are you strongest in? Give me a quick rundown.",
    "Describe a challenging technical problem you solved recently. What was your approach?",
    "What are your salary expectations for this position?",
    "What is your notice period or earliest start date?",
    "Do you have any questions for me about the role or company?",
]

@app.post("/twiml/screening")
async def screening_twiml(request: Request, job_id: str, candidate_id: str):
    form = await request.form()
    question_index = int(form.get("question_index", 0))
    response = VoiceResponse()

    if question_index == 0:
        response.say(
            "Hi, this is an automated screening call for a software engineering position. "
            "I'll ask you six questions. Please answer after the beep. Press any key to begin.",
            voice="Polly.Joanna"
        )
        gather = Gather(num_digits=1, action=f"/twiml/screening?job_id={job_id}&candidate_id={candidate_id}&question_index=1", method="POST")
        response.append(gather)
        return Response(content=str(response), media_type="application/xml")

    if question_index <= len(SCREENING_QUESTIONS):
        question = SCREENING_QUESTIONS[question_index - 1]
        response.say(question, voice="Polly.Joanna")
        record = Record(
            action=f"/twiml/screening?job_id={job_id}&candidate_id={candidate_id}&question_index={question_index + 1}",
            method="POST",
            max_length=120,
            play_beep=True,
            trim="trim-silence",
            recording_status_callback=f"/webhook/recording?job_id={job_id}&candidate_id={candidate_id}&question_index={question_index}",
        )
        response.append(record)
        return Response(content=str(response), media_type="application/xml")

    response.say("Thank you for your time. A recruiter will review your responses and follow up soon. Goodbye.", voice="Polly.Joanna")
    response.hangup()
    return Response(content=str(response), media_type="application/xml")

Deploy this behind HTTPS (ngrok for local testing, a real domain for production). Each recording lands at /webhook/recording — store the RecordingUrl keyed by candidate_id and question_index.

Step 4: Build the transcriber agent

This agent pulls recordings from Twilio, sends them to Whisper, and returns a full transcript. Use the crewai-tools BaseTool pattern for reusability.

# agents/transcriber.py
from crewai import Agent
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field
import requests
import os
from openai import OpenAI

class TranscribeInput(BaseModel):
    candidate_id: str = Field(description="Candidate identifier")
    recording_urls: list[str] = Field(description="List of Twilio recording URLs for each question")

class TranscribeTool(BaseTool):
    name: str = "transcribe_recordings"
    args_schema: Type[BaseModel] = TranscribeInput

    def _run(self, candidate_id: str, recording_urls: list[str]) -> str:
        client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL"))
        full_transcript = []

        for i, url in enumerate(recording_urls):
            # Twilio requires auth for media downloads
            resp = requests.get(url, auth=(os.getenv("TWILIO_ACCOUNT_SID"), os.getenv("TWILIO_AUTH_TOKEN")))
            resp.raise_for_status()

            # Save to temp file for Whisper
            temp_path = f"/tmp/{candidate_id}_q{i}.wav"
            with open(temp_path, "wb") as f:
                f.write(resp.content)

            with open(temp_path, "rb") as audio_file:
                transcript = client.audio.transcriptions.create(
                    model="whisper-1",
                    file=audio_file,
                    response_format="text",
                )
            full_transcript.append(f"Question {i+1}: {transcript.strip()}")

        return "\n\n".join(full_transcript)

transcriber = Agent(
    role="Audio Transcriber",
    goal="Convert all screening call recordings into a single clean transcript",
    backstory="You fetch each recording from Twilio, run it through Whisper, and concatenate results with question labels.",
    tools=[TranscribeTool()],
    verbose=True,
    allow_delegation=False,
)

Step 5: Build the extractor agent

The extractor takes the raw transcript and the job requirements, then outputs a structured ScreeningResult. This is where the LLM does heavy lifting — prompt it carefully.

# agents/extractor.py
from crewai import Agent
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field
import json
from models import ScreeningResult, CandidateSkill, SkillLevel, JobRequirement

class ExtractInput(BaseModel):
    transcript: str = Field(description="Full transcript from transcriber")
    job_requirements: str = Field(description="JSON string of JobRequirement")
    candidate_id: str = Field(description="Candidate identifier")
    phone_number: str = Field(description="Candidate phone number")
    call_sid: str = Field(description="Twilio call SID")

class ExtractTool(BaseTool):
    name: str = "extract_structured_data"
    args_schema: Type[BaseModel] = ExtractInput

    def _run(self, transcript: str, job_requirements: str, candidate_id: str, phone_number: str, call_sid: str) -> str:
        # In practice, call an LLM here with a structured prompt.
        # This stub returns the schema shape; replace with actual LLM call.
        prompt = f"""
        You are an expert technical recruiter. Extract structured data from this screening transcript.

        Job Requirements:
        {job_requirements}

        Transcript:
        {transcript}

        Return ONLY valid JSON matching this schema:
        {{
            "skills": [{{"name": "string", "level": "none|basic|intermediate|advanced|expert", "years_experience": float}}],
            "availability": "string",
            "salary_expectation": int or null,
            "red_flags": ["string"],
            "overall_score": float,  # 0-100 based on job fit
            "summary": "string"  # 2-3 sentence executive summary
        }}

        Scoring rubric:
        - 90-100: Exceeds all requirements, strong communication
        - 70-89: Meets most requirements, minor gaps
        - 50-69: Meets basic requirements, notable gaps
        - 30-49: Significant gaps, possible misalignment
        - 0-29: Does not meet minimum requirements
        """
        # TODO: call LLM with prompt, parse JSON, validate against ScreeningResult
        # For now, return a placeholder that matches the schema
        return json.dumps({
            "skills": [],
            "availability": "unknown",
            "salary_expectation": None,
            "red_flags": [],
            "overall_score": 0.0,
            "summary": "LLM extraction not implemented in stub",
        })

extractor = Agent(
    role="Candidate Data Extractor",
    goal="Transform raw transcripts into structured, validated screening results",
    backstory="You read transcripts like a senior recruiter. You identify skills with proficiency levels, catch red flags (vague answers, salary mismatch, availability conflicts), and score fit against the job spec.",
    tools=[ExtractTool()],
    verbose=True,
    allow_delegation=False,
)

Replace the stub with a real LLM call using the same OpenAI client pattern. The prompt above is production-ready — adjust the rubric to your hiring bar.

Step 6: Build the ranker agent

The ranker receives a list of ScreeningResult objects and the job requirements, then returns a ranked shortlist with rationale.

# agents/ranker.py
from crewai import Agent
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field
from models import ScreeningResult, JobRequirement
import json

class RankInput(BaseModel):
    results: list[str] = Field(description="List of ScreeningResult JSON strings")
    job_requirements: str = Field(description="JSON string of JobRequirement")
    top_k: int = Field(default=5, description="Number of candidates to return")

class RankTool(BaseTool):
    name: str = "rank_candidates"
    args_schema: Type[BaseModel] = RankInput

    def _run(self, results: list[str], job_requirements: str, top_k: int) -> str:
        prompt = f"""
        You are a hiring manager reviewing screening results. Rank these candidates for the role.

        Job Requirements:
        {job_requirements}

        Candidate Results:
        {json.dumps([json.loads(r) for r in results], indent=2)}

        Return ONLY valid JSON:
        {{
            "ranked_candidates": [
                {{
                    "candidate_id": "string",
                    "rank": int,
                    "score": float,
                    "rationale": "string",  # 2-3 sentences explaining rank
                    "next_steps": "string"  # e.g., "Schedule technical interview", "Reject - salary mismatch"
                }}
            ]
        }}

        Prioritize: overall_score, required skill coverage, salary alignment, availability, red flags (negative weight).
        """
        # TODO: call LLM, parse, validate
        return json.dumps({"ranked_candidates": []})

ranker = Agent(
    role="Hiring Manager",
    goal="Produce a ranked shortlist with clear rationale for each candidate",
    backstory="You make the final call on who advances. You weigh technical fit, compensation alignment, timeline, and red flags. Your output goes straight to the recruiting team.",
    tools=[RankTool()],
    verbose=True,
    allow_delegation=False,
)

Step 7: Wire the crew together

Now compose the agents into a sequential crew. The scheduler kicks off calls; a separate process (webhook listener) triggers the transcriber → extractor → ranker pipeline once recordings are ready.

# crew.py
from crewai import Crew, Process, Task
from agents.scheduler import scheduler, ScheduleCallsTool
from agents.transcriber import transcriber, TranscribeTool
from agents.extractor import extractor, ExtractTool
from agents.ranker import ranker, RankTool
from models import JobRequirement, CandidateSkill, SkillLevel
import json

# Define the job once
job = JobRequirement(
    title="Senior Backend Engineer",
    required_skills=[
        CandidateSkill(name="Python", level=SkillLevel.ADVANCED, years_experience=5),
        CandidateSkill(name="PostgreSQL", level=SkillLevel.INTERMEDIATE, years_experience=3),
        CandidateSkill(name="AWS", level=SkillLevel.INTERMEDIATE, years_experience=3),
    ],
    nice_to_have_skills=[
        CandidateSkill(name="Kubernetes", level=SkillLevel.INTERMEDIATE, years_experience=2),
        CandidateSkill(name="Redis", level=SkillLevel.BASIC, years_experience=1),
    ],
    min_salary=140000,
    max_salary=180000,
    location="San Francisco",
    remote_ok=True,
)

# Task 1: Schedule calls (run this manually or on a cron)
schedule_task = Task(
    description="Initiate screening calls for all candidates in candidates.csv",
    expected_output="List of dicts with candidate_id, phone_number, name, call_sid",
    agent=scheduler,
    tools=[ScheduleCallsTool()],
    input_data={"candidates_csv": "candidates.csv", "job_id": "sr-backend-001"},
)

# Task 2: Transcribe (triggered per-candidate via webhook when recordings ready)
transcribe_task = Task(
    description="Transcribe all recordings for a single candidate",
    expected_output="Full transcript with question labels",
    agent=transcriber,
    tools=[TranscribeTool()],
    # Input comes from webhook handler: candidate_id, recording_urls[]
)

# Task 3: Extract structured data
extract_task = Task(
    description="Extract skills, salary, availability, red flags, and score from transcript",
    expected_output="Valid ScreeningResult JSON",
    agent=extractor,
    tools=[ExtractTool()],
    # Input: transcript, job_requirements (JSON), candidate_id, phone_number, call_sid
)

# Task 4: Rank all candidates (run after batch completes)
rank_task = Task(
    description="Rank all screened candidates and return top 5 with rationale",
    expected_output="Ranked shortlist JSON",
    agent=ranker,
    tools=[RankTool()],
    # Input: list of ScreeningResult JSON strings, job_requirements JSON, top_k=5
)

# Crew for the batch ranking phase (run after all extractions done)
ranking_crew = Crew(
    agents=[ranker],
    tasks=[rank_task],
    process=Process.sequential,
    verbose=True,
)

if __name__ == "__main__":
    # Phase 1: Schedule calls
    print("=== Scheduling calls ===")
    schedule_result = schedule_task.execute()
    print(schedule_result)

    # Phase 2-3: Handled by webhook listeners (see below)
    # Phase 4: Rank (run manually after all candidates processed)
    # ranking_crew.kickoff(inputs={...})

Step 8: Implement the webhook listener

The async glue: a FastAPI endpoint that receives Twilio’s recording_status_callback, aggregates recordings per candidate, then triggers the transcriber → extractor chain.

# webhook_handler.py
from fastapi import FastAPI, Form, BackgroundTasks
from crewai import Crew, Process, Task
from agents.transcriber import transcriber, TranscribeTool
from agents.extractor import extractor, ExtractTool
from models import JobRequirement
import json
import os
from collections import defaultdict

app = FastAPI()

# In-memory store for demo; use Redis/DB in production
recordings_store = defaultdict(list)
job_requirements_json = json.dumps({
    "title": "Senior Backend Engineer",
    "required_skills": [{"name": "Python", "level": "advanced", "years_experience": 5}],
    "nice_to_have_skills": [],
    "min_salary": 140000,
    "max_salary": 180000,
    "location": "San Francisco",
    "remote_ok": True,
})

@app.post("/webhook/recording")
async def recording_webhook(
    background_tasks: BackgroundTasks,
    RecordingUrl: str = Form(...),
    CallSid: str = Form(...),
    candidate_id: str = Form(...),
    question_index: str = Form(...),
):
    recordings_store[candidate_id].append({
        "question_index": int(question_index),
        "url": RecordingUrl,
        "call_sid": CallSid,
    })

    # Check if all 6 questions recorded (adjust as needed)
    if len(recordings_store[candidate_id]) >= 6:
        background_tasks.add_task(process_candidate, candidate_id)

    return {"status": "received"}

async def process_candidate(candidate_id: str):
    recordings = sorted(recordings_store[candidate_id], key=lambda x: x["question_index"])
    urls = [r["url"] for r in recordings]
    call_sid = recordings[0]["call_sid"]
    phone_number = "+15551234567"  # Look up from your candidate DB

    # Transcribe
    transcribe_crew = Crew(
        agents=[transcriber],
        tasks=[Task(
            description=f"Transcribe recordings for {candidate_id}",
            expected_output="Full transcript",
            agent=transcriber,
            tools=[TranscribeTool()],
            input_data={"candidate_id": candidate_id, "recording_urls": urls},
        )],
        process=Process.sequential,
    )
    transcript = transcribe_crew.kickoff()

    # Extract
    extract_crew = Crew(
        agents=[extractor],
        tasks=[Task(
            description=f"Extract structured data for {candidate_id}",
            expected_output="ScreeningResult JSON",
            agent=extractor,
            tools=[ExtractTool()],
            input_data={
                "transcript": transcript,
                "job_requirements": job_requirements_json,
                "candidate_id": candidate_id,
                "phone_number": phone_number,
                "call_sid": call_sid,
            },
        )],
        process=Process.sequential,
    )
    result = extract_crew.kickoff()

    # Persist result to DB/queue for ranking phase
    print(f"Completed {candidate_id}: {result}")
    # TODO: save to PostgreSQL, push to ranking queue, etc.

Run the webhook server: uvicorn webhook_handler:app --host 0.0.0.0 --port 8000. Expose it via ngrok (ngrok http 8000) and update your TwiML endpoint URLs accordingly.

Step 9: Verify end-to-end

Create a test CSV with your own number and a colleague’s:

candidate_id,phone_number,name
test-001,+15551112222,Test Candidate
test-002,+15553334444,Another Candidate

Run the scheduler:

python crew.py

You should receive a call. Answer, respond to all six questions, hang up. Watch the webhook logs — you’ll see recordings arrive, transcription kick off, extraction run, and a ScreeningResult print at the end.

Verify the output structure:

{
  "candidate_id": "test-001",
  "phone_number": "+15551112222",
  "call_sid": "CA...",
  "transcript": "Question 1: ...\n\nQuestion 2: ...",
  "skills": [
    {"name": "Python", "level": "advanced", "years_experience": 6.0},
    {"name": "PostgreSQL", "level": "intermediate", "years_experience": 4.0}
  ],
  "availability": "2 weeks notice",
  "salary_expectation": 165000,
  "red_flags": [],
  "overall_score": 87.5,
  "summary": "Strong backend candidate with 6 years Python, 4 years PostgreSQL. Salary within band. Available in 2 weeks. Clear communicator."
}

Once you have 3+ results, trigger the ranking crew manually or via a scheduled job. Confirm the ranked output orders candidates by your rubric and includes actionable next_steps.

Production hardening

• Idempotency: Store call_sid and recording_url processed flags in Postgres. Re-running a webhook shouldn’t duplicate work.
• Error handling: Wrap each tool _run in try/except. On failure, write to a dead-letter queue with context for manual retry.
• Cost control: Whisper charges per minute. Set max_length=120 on Record (done above). Monitor daily spend via your provider’s usage API.
• Latency: Transcription + extraction takes 30-90 seconds per candidate. Run the webhook handler on a worker pool (Celery, RQ, or Cloud Run jobs) so Twilio doesn’t time out.
• Compliance: Play a “this call is recorded” message upfront. Check local two-party consent laws. Store recordings encrypted, purge after 30 days unless candidate consents.
• Observability: Emit structured logs (JSON) with candidate_id, call_sid, stage, duration_ms, token_usage. Alert on extraction failure rate > 5%.

Iteration loop

Your first rubric will be wrong. Treat the extractor and ranker prompts as code — version them, A/B test them, and measure precision@k against actual hiring outcomes. Start with a shadow mode: run the crew in parallel with human screeners, compare rankings, and tune prompts until the system consistently surfaces the same top candidates.

The crewai recruiting screening automation example above gives you a working skeleton. Swap the LLM provider, add a technical coding agent, or plug in a calendar tool for the “schedule technical interview” next step — the agent boundaries stay clean because each one owns a single transformation.