CrewAI example: automated meeting notes and follow-ups

You want a crewai meeting notes automation example that actually runs in production, not a toy notebook. This guide walks you through building a three-agent pipeline: one agent transcribes and summarizes, a second extracts structured action items with owners and due dates, and a third drafts personalized follow-up emails. You’ll wire them together with CrewAI’s sequential process, add guardrails for hallucination-prone fields, and verify each stage with deterministic tests.

Step 1: Set up the project and dependencies

Create a virtual environment and install the minimal set. CrewAI sits on top of LangChain, so you need an LLM provider. This example uses OpenAI-compatible endpoints — swap in your preferred provider.

python -m venv .venv
source .venv/bin/activate
pip install crewai langchain-openai python-dotenv pydantic==2.8.2

Create a .env file with your API key and base URL. If you route through a gateway that handles fallback and model selection, point OPENAI_BASE_URL there.

# .env
OPENAI_API_KEY=sk-...
OPENAI_BASE_URL=https://api.openai.com/v1  # or your gateway endpoint
MODEL_NAME=gpt-4o-mini

Verify the environment loads:

# test_env.py
import os
from dotenv import load_dotenv

load_dotenv()
assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY not set"
assert os.getenv("MODEL_NAME"), "MODEL_NAME not set"
print("Environment OK")

Run python test_env.py — you should see “Environment OK”.

Step 2: Define the data models

Pydantic models give you validation at the boundaries between agents. This prevents the summarizer from hallucinating fields the action-item extractor expects.

# models.py
from datetime import date, datetime
from enum import Enum
from pydantic import BaseModel, Field
from typing import Optional


class Priority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    URGENT = "urgent"


class ActionItem(BaseModel):
    owner: str = Field(..., min_length=1, max_length=100)
    due_date: Optional[date] = None
    priority: Priority = Priority.MEDIUM
    context: Optional[str] = Field(None, max_length=300)


class MeetingSummary(BaseModel):
    title: str
    date: date
    participants: list[str]
    key_decisions: list[str] = Field(default_factory=list)
    action_items: list[ActionItem] = Field(default_factory=list)
    next_steps: list[str] = Field(default_factory=list)


class FollowUpEmail(BaseModel):
    to: str
    subject: str
    body: str
    action_items_reference: list[str] = Field(default_factory=list)

Run a quick validation test:

# test_models.py
from datetime import date
from models import ActionItem, MeetingSummary, Priority

item = ActionItem(
    description="Finalize Q3 budget forecast",
    owner="Sarah Chen",
    due_date=date(2024, 8, 15),
    priority=Priority.HIGH,
)
summary = MeetingSummary(
    title="Q3 Planning Sync",
    date=date(2024, 7, 28),
    participants=["Sarah Chen", "Marcus Webb", "Priya Patel"],
    key_decisions=["Move to rolling forecasts", "Hire two analysts"],
    action_items=[item],
)
print(summary.model_dump_json(indent=2))

Step 3: Build the transcription and summarization agent

This agent takes raw transcript text (from Whisper, AssemblyAI, or a manual paste) and produces a MeetingSummary. Keep the prompt tight — explicit output format, few-shot examples, and a hard constraint that every action item must have an owner.

# agents/summarizer.py
from crewai import Agent, Task
from langchain_openai import ChatOpenAI
from models import MeetingSummary
import os

llm = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0.1,
    base_url=os.getenv("OPENAI_BASE_URL"),
)

summarizer = Agent(
    role="Meeting Summarizer",
    goal="Convert raw meeting transcripts into structured summaries with decisions and action items",
    backstory=(
        "You are an executive assistant who has summarized thousands of board meetings, "
        "standups, and planning sessions. You extract only what was explicitly stated — "
        "no inference, no hallucination."
    ),
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

summarize_task = Task(
    description=(
        "Summarize the following meeting transcript.\n\n"
        "TRANSCRIPT:\n{transcript}\n\n"
        "Return a JSON object matching the MeetingSummary schema. "
        "Every action item MUST have an owner explicitly mentioned in the transcript. "
        "If no due date is stated, omit the field. Priority defaults to medium."
    ),
    expected_output="Valid JSON matching the MeetingSummary schema",
    agent=summarizer,
    output_json=MeetingSummary,
)

Test it with a sample transcript:

# test_summarizer.py
from agents.summarizer import summarize_task, summarizer
from crewai import Crew

transcript = """
Sarah: Thanks everyone for joining the Q3 planning sync. First item: we're moving to rolling forecasts instead of quarterly. Marcus, you're leading that migration.
Marcus: Got it. Target is end of August for the first rolling cycle.
Priya: I'll need two additional analysts to handle the increased cadence. Sarah, can we get budget approval?
Sarah: Approved. Priya, hire two analysts by September 1st. Also, we need the Q3 budget forecast finalized by August 15th — that's a hard deadline from finance.
Marcus: I'll own the forecast. Priya, you're on hiring.
Priya: Understood.
Sarah: Great. Next steps: Marcus sends the rolling forecast spec by Friday. Priya posts the job reqs by Wednesday.
"""

crew = Crew(
    agents=[summarizer],
    tasks=[summarize_task],
    verbose=True,
)

result = crew.kickoff(inputs={"transcript": transcript})
print(result.raw)

Verify the output: every action item has an owner from the transcript, due dates are ISO format, priorities are valid enum values. If the model omits due_date for items without explicit dates, that’s correct behavior.

Step 4: Build the action-item extraction agent

The summarizer already emits action items, but a dedicated extractor adds normalization: it validates owners against the participant list, infers priority from language cues (“blocker”, “ASAP”, “hard deadline”), and ensures no duplicate items slip through.

# agents/extractor.py
from crewai import Agent, Task
from langchain_openai import ChatOpenAI
from models import MeetingSummary, ActionItem, Priority
from typing import List
import os
import json

llm = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0.0,
    base_url=os.getenv("OPENAI_BASE_URL"),
)

extractor = Agent(
    role="Action Item Extractor",
    goal="Normalize and validate action items from a meeting summary",
    backstory=(
        "You audit action items for completeness. You check that every owner attended the meeting, "
        "deduplicate items that refer to the same work, and assign priority based on explicit urgency signals."
    ),
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

extract_task = Task(
    description=(
        "You receive a MeetingSummary JSON. Perform these transformations:\n"
        "1. Remove any action item whose owner is not in the participants list.\n"
        "2. Deduplicate items with semantically identical descriptions (keep the one with the earliest due date).\n"
        "3. Upgrade priority to HIGH if the transcript contains 'blocker', 'urgent', 'ASAP', 'hard deadline', or 'critical'.\n"
        "4. Upgrade to URGENT if the due date is within 3 business days of the meeting date.\n"
        "5. Return the updated MeetingSummary with cleaned action_items.\n\n"
        "INPUT SUMMARY:\n{summary_json}"
    ),
    expected_output="Valid JSON matching the MeetingSummary schema with cleaned action_items",
    agent=extractor,
    output_json=MeetingSummary,
)

Chain it after the summarizer:

# pipeline.py
from crewai import Crew, Process
from agents.summarizer import summarizer, summarize_task
from agents.extractor import extractor, extract_task
from models import MeetingSummary
import json

# Wire the output of summarize_task into extract_task
extract_task.context = [summarize_task]

crew = Crew(
    agents=[summarizer, extractor],
    tasks=[summarize_task, extract_task],
    process=Process.sequential,
    verbose=True,
)

def run_pipeline(transcript: str) -> MeetingSummary:
    result = crew.kickoff(inputs={"transcript": transcript})
    # The final task's output is the cleaned summary
    return MeetingSummary.model_validate_json(result.raw)

Test the full chain:

# test_pipeline.py
from pipeline import run_pipeline

transcript = """
Sarah: Thanks everyone for joining the Q3 planning sync. First item: we're moving to rolling forecasts instead of quarterly. Marcus, you're leading that migration.
Marcus: Got it. Target is end of August for the first rolling cycle.
Priya: I'll need two additional analysts to handle the increased cadence. Sarah, can we get budget approval?
Sarah: Approved. Priya, hire two analysts by September 1st. Also, we need the Q3 budget forecast finalized by August 15th — that's a hard deadline from finance.
Marcus: I'll own the forecast. Priya, you're on hiring.
Priya: Understood.
Sarah: Great. Next steps: Marcus sends the rolling forecast spec by Friday. Priya posts the job reqs by Wednesday.
"""

summary = run_pipeline(transcript)
print(summary.model_dump_json(indent=2))

# Verification checks
assert all(item.owner in summary.participants for item in summary.action_items), "Owner not in participants"
descriptions = [item.description.lower() for item in summary.action_items]
assert len(descriptions) == len(set(descriptions)), "Duplicate action items remain"
print("All verification checks passed")

Run python test_pipeline.py. You should see three action items (forecast, hiring, spec), all owners in participants, “hard deadline” item marked HIGH or URGENT, no duplicates.

Step 5: Build the follow-up email agent

This agent takes the cleaned MeetingSummary and writes one email per participant. Each email references only that person’s action items, uses a professional tone, and includes a clear subject line.

# agents/emailer.py
from crewai import Agent, Task
from langchain_openai import ChatOpenAI
from models import MeetingSummary, FollowUpEmail
from typing import List
import os

llm = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0.3,
    base_url=os.getenv("OPENAI_BASE_URL"),
)

emailer = Agent(
    role="Follow-up Email Drafter",
    goal="Write personalized follow-up emails for each meeting participant",
    backstory=(
        "You write concise, action-oriented follow-ups. Each email addresses one recipient, "
        "lists only their action items with due dates, and closes with the next meeting date if known. "
        "No fluff, no corporate speak."
    ),
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

email_task = Task(
    description=(
        "You receive a MeetingSummary. For each participant, generate a FollowUpEmail with:\n"
        "- to: participant name\n"
        "- subject: 'Follow-up: {meeting_title} - Your Action Items'\n"
        "- body: greeting, 2-sentence meeting recap, numbered list of their action items "
        "(description, due date, priority), sign-off\n"
        "- action_items_reference: list of their action item descriptions\n\n"
        "Return a JSON array of FollowUpEmail objects.\n\n"
        "INPUT SUMMARY:\n{summary_json}"
    ),
    expected_output="JSON array of FollowUpEmail objects, one per participant",
    agent=emailer,
    # CrewAI doesn't support list output_json directly; we'll parse manually
)

# Custom parser since output_json expects a single model
def parse_emails(raw: str) -> List[FollowUpEmail]:
    import json
    data = json.loads(raw)
    return [FollowUpEmail.model_validate(item) for item in data]

Add the email task to the pipeline:

# pipeline.py (updated)
from crewai import Crew, Process
from agents.summarizer import summarizer, summarize_task
from agents.extractor import extractor, extract_task
from agents.emailer import emailer, email_task, parse_emails
from models import MeetingSummary, FollowUpEmail
import json

extract_task.context = [summarize_task]
email_task.context = [extract_task]

crew = Crew(
    agents=[summarizer, extractor, emailer],
    tasks=[summarize_task, extract_task, email_task],
    process=Process.sequential,
    verbose=True,
)

def run_full_pipeline(transcript: str) -> tuple[MeetingSummary, list[FollowUpEmail]]:
    result = crew.kickoff(inputs={"transcript": transcript})
    # Last task output is the email array
    emails = parse_emails(result.raw)
    # The extractor task output is the cleaned summary
    # Access it via the task output
    summary_json = extract_task.output.raw
    summary = MeetingSummary.model_validate_json(summary_json)
    return summary, emails

Test the complete crewai meeting notes automation example end to end:

# test_full.py
from pipeline import run_full_pipeline

transcript = """
Sarah: Thanks everyone for joining the Q3 planning sync. First item: we're moving to rolling forecasts instead of quarterly. Marcus, you're leading that migration.
Marcus: Got it. Target is end of August for the first rolling cycle.
Priya: I'll need two additional analysts to handle the increased cadence. Sarah, can we get budget approval?
Sarah: Approved. Priya, hire two analysts by September 1st. Also, we need the Q3 budget forecast finalized by August 15th — that's a hard deadline from finance.
Marcus: I'll own the forecast. Priya, you're on hiring.
Priya: Understood.
Sarah: Great. Next steps: Marcus sends the rolling forecast spec by Friday. Priya posts the job reqs by Wednesday.
"""

summary, emails = run_full_pipeline(transcript)

print("=== SUMMARY ===")
print(summary.model_dump_json(indent=2))

print("\n=== EMAILS ===")
for email in emails:
    print(f"\n--- To: {email.to} ---")
    print(f"Subject: {email.subject}")
    print(email.body)

# Verification
assert len(emails) == 3, f"Expected 3 emails, got {len(emails)}"
for email in emails:
    assert email.to in summary.participants, f"Email recipient {email.to} not in participants"
    # Each email should reference only that person's items
    for ref in email.action_items_reference:
        owner_items = [item.description for item in summary.action_items if item.owner == email.to]
        assert ref in owner_items, f"Email for {email.to} references item not assigned to them"
print("\nAll verification checks passed")

Run python test_full.py. You should see three personalized emails. Marcus gets the forecast and spec items. Priya gets hiring and job reqs. Sarah gets zero action items (she assigned them) but still receives a recap email — adjust the prompt if you want to skip non-owners.

Step 6: Add idempotency and observability

Production pipelines need retries, logging, and idempotency keys so re-running on the same transcript doesn’t generate duplicate emails.

# pipeline.py (additions)
import hashlib
import logging
from pathlib import Path
from typing import Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

CACHE_DIR = Path(".crewai_cache")
CACHE_DIR.mkdir(exist_ok=True)

def transcript_hash(transcript: str) -> str:
    return hashlib.sha256(transcript.encode()).hexdigest()[:16]

def run_full_pipeline(transcript: str, force: bool = False) -> tuple[MeetingSummary, list[FollowUpEmail]]:
    key = transcript_hash(transcript)
    cache_file = CACHE_DIR / f"{key}.json"
    
    if cache_file.exists() and not force:
        logger.info(f"Cache hit for {key}")
        import json
        data = json.loads(cache_file.read_text())
        return (
            MeetingSummary.model_validate(data["summary"]),
            [FollowUpEmail.model_validate(e) for e in data["emails"]],
        )
    
    logger.info(f"Running pipeline for {key}")
    result = crew.kickoff(inputs={"transcript": transcript})
    emails = parse_emails(result.raw)
    summary_json = extract_task.output.raw
    summary = MeetingSummary.model_validate_json(summary_json)
    
    cache_file.write_text(json.dumps({
        "summary": summary.model_dump(mode="json"),
        "emails": [e.model_dump(mode="json") for e in emails],
    }, indent=2))
    
    return summary, emails

Verify idempotency:

# test_idempotency.py
from pipeline import run_full_pipeline

transcript = "Sarah: Quick sync. Marcus, send the spec by Friday. Thanks."

summary1, emails1 = run_full_pipeline(transcript)
summary2, emails2 = run_full_pipeline(transcript)

assert summary1.model_dump() == summary2.model_dump()
assert [e.model_dump() for e in emails1] == [e.model_dump() for e in emails2]
print("Idempotency verified")

Step 7: Wire into your ingestion path

The pipeline is a pure function — transcript in, (summary, emails) out. Connect it wherever transcripts land: a webhook from your transcription service, a scheduled job polling a folder, or a CLI for ad-hoc runs.

# cli.py
import sys
from pathlib import Path
from pipeline import run_full_pipeline

def main():
    if len(sys.argv) < 2:
        print("Usage: python cli.py <transcript_file> [--force]")
        sys.exit(1)
    
    transcript_path = Path(sys.argv[1])
    force = "--force" in sys.argv
    
    transcript = transcript_path.read_text()
    summary, emails = run_full_pipeline(transcript, force=force)
    
    # Output summary to stdout, emails to files
    print(summary.model_dump_json(indent=2))
    
    out_dir = Path("output") / transcript_path.stem
    out_dir.mkdir(parents=True, exist_ok=True)
    for email in emails:
        (out_dir / f"{email.to.replace(' ', '_')}.eml").write_text(
            f"Subject: {email.subject}\n\n{email.body}"
        )
    print(f"\nWrote {len(emails)} emails to {out_dir}")

if __name__ == "__main__":
    main()

Test the CLI:

echo "Sarah: Marcus, send spec by Friday. Priya, post reqs by Wednesday." > test_transcript.txt
python cli.py test_transcript.txt
ls output/test_transcript/

You should see Marcus.eml and Priya.eml with correct content.

Step 8: Guardrails for production

Three failure modes deserve explicit handling:

1. Missing participants in action items — The extractor already filters these, but log a warning when it happens.

# In extract_task description, add:
# "Log a warning for each removed item: 'Removed action item assigned to non-participant: {owner}'"

2. Hallucinated due dates — The summarizer omits dates not in the transcript. If your downstream system requires dates, add a validation step that flags items with due_date: null and priority: HIGH/URGENT.

# validation.py
from models import MeetingSummary
from datetime import date, timedelta

def validate_summary(summary: MeetingSummary) -> list[str]:
    warnings = []
    for item in summary.action_items:
        if item.priority in (Priority.HIGH, Priority.URGENT) and item.due_date is None:
            warnings.append(f"High-priority item missing due date: {item.description}")
        if item.due_date and item.due_date < summary.date:
            warnings.append(f"Due date before meeting date: {item.description}")
    return warnings

3. Email delivery failures — The pipeline produces .eml files. Hand off to your mail service (SendGrid, Postmark, SES) with its own retry logic. Don’t couple the CrewAI flow to email sending.

Verification checklist

Before declaring this ready, run through:

• python test_env.py passes
• python test_models.py outputs valid JSON
• python test_summarizer.py produces summary with correct owners
• python test_pipeline.py passes all assertions (no duplicate items, owners validated, priority escalation works)
• python test_full.py generates three emails, each referencing only that recipient’s items
• python test_idempotency.py passes — second run returns cached results
• python cli.py test_transcript.txt writes .eml files to output/

What to extend next

• Thread detection: If transcripts span recurring meetings, add a meeting-series ID and carry forward open action items.
• Calendar integration: Parse .ics invites to pre-populate participants and meeting date.
• Slack/Teams delivery: Post summaries to a channel, DM action items to owners.
• Vector search: Embed summaries for “what did we decide about X last quarter?” queries.

The crewai meeting notes automation example above is deliberately minimal — three agents, sequential process, Pydantic boundaries, file-based cache. Each piece is replaceable. Swap the LLM, swap the transcription source, swap the email transport. The contract between stages is the schema, not the implementation.