CrewAI support bot: agents for tier-1 and tier-2 tickets

You’re building a crewai support bot tiered tickets system because your support queue is drowning in repetitive tier-1 requests while tier-2 issues sit waiting for senior engineers. CrewAI’s multi-agent framework lets you model this as a pipeline: a triage agent classifies incoming tickets, a resolver agent handles known issues with runbooks, and an escalation agent packages context for human review. This tutorial walks through a production-shaped implementation with observable checkpoints at each stage.

Prerequisites

• Python 3.11+
• An OpenAI-compatible API key (OpenAI, Anthropic via proxy, or a gateway like n4n.ai)
• crewai>=0.80, crewai-tools>=0.6, pydantic>=2.7, rich>=13.7
• Basic familiarity with CrewAI concepts: agents, tasks, crews, and tools

Install dependencies:

pip install "crewai>=0.80" "crewai-tools>=0.6" pydantic rich python-dotenv

Create a .env file with your API credentials:

# .env
OPENAI_API_KEY=sk-...
OPENAI_MODEL_NAME=gpt-4o-mini
# If using a gateway:
# OPENAI_BASE_URL=https://api.n4n.ai/v1

Project structure

support_bot/
├── config/
│   └── runbooks.yaml
├── src/
│   ├── models.py
│   ├── tools.py
│   ├── agents.py
│   ├── tasks.py
│   ├── crew.py
│   └── main.py
├── tests/
│   └── test_triage.py
├── .env
└── pyproject.toml

Step 1: Define the data contracts

Start with Pydantic models so every agent speaks the same schema. This prevents the “stringly typed” chaos that plagues LLM pipelines.

# src/models.py
from enum import Enum
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime


class TicketTier(str, Enum):
    TIER_1 = "tier_1"
    TIER_2 = "tier_2"
    ESCALATE = "escalate"


class TicketCategory(str, Enum):
    BILLING = "billing"
    AUTH = "auth"
    INTEGRATION = "integration"
    PERFORMANCE = "performance"
    BUG = "bug"
    UNKNOWN = "unknown"


class TicketInput(BaseModel):
    id: str
    subject: str
    body: str
    customer_tier: str = "standard"  # standard, premium, enterprise
    created_at: datetime = Field(default_factory=datetime.utcnow)
    metadata: dict = Field(default_factory=dict)


class TriageResult(BaseModel):
    ticket_id: str
    tier: TicketTier
    category: TicketCategory
    confidence: float = Field(ge=0.0, le=1.0)
    reasoning: str
    suggested_runbook: Optional[str] = None


class Resolution(BaseModel):
    ticket_id: str
    resolved: bool
    resolution_text: str
    actions_taken: List[str] = Field(default_factory=list)
    requires_human: bool = False
    escalation_reason: Optional[str] = None


class EscalationPackage(BaseModel):
    ticket_id: str
    original_ticket: TicketInput
    triage: TriageResult
    resolution_attempt: Optional[Resolution] = None
    context_summary: str
    recommended_owner: str  # team or individual
    priority: int = Field(ge=1, le=5)

Checkpoint: Run python -c "from src.models import TicketInput; print(TicketInput(id='t-1', subject='Test', body='Body').model_dump_json(indent=2))" — you should see clean JSON with all fields.

Step 2: Load runbooks as structured data

Tier-1 resolution works best when agents follow deterministic runbooks, not free-form reasoning. Store runbooks in YAML and load them as tools.

# config/runbooks.yaml
runbooks:
  - id: "rb-billing-001"
    title: "Failed payment retry"
    category: "billing"
    tier: "tier_1"
    symptoms:
      - "payment failed"
      - "card declined"
      - "invoice unpaid"
    steps:
      - "Verify payment method on file"
      - "Retry charge via Stripe dashboard"
      - "If retry fails, send update-payment-method email template"
      - "Confirm resolution with customer"
    escalation_criteria: "Customer requests refund or disputes charge"

  - id: "rb-auth-001"
    title: "Password reset not working"
    category: "auth"
    tier: "tier_1"
    symptoms:
      - "reset email not received"
      - "reset link expired"
      - "token invalid"
    steps:
      - "Check email delivery logs for reset email"
      - "Verify token TTL configuration (default 2hr)"
      - "Generate manual reset link via admin panel"
      - "Send directly to customer via secure channel"
    escalation_criteria: "Account takeover suspected or MFA issues"

  - id: "rb-integration-001"
    title: "Webhook delivery failures"
    category: "integration"
    tier: "tier_2"
    symptoms:
      - "webhook timeout"
      - "signature verification failed"
      - "endpoint returning 5xx"
    steps:
      - "Check webhook endpoint health via /healthz"
      - "Verify HMAC secret rotation schedule"
      - "Inspect payload schema against API version"
      - "Replay failed deliveries from dashboard"
    escalation_criteria: "Customer endpoint fundamentally broken or schema mismatch requires engineering"

  - id: "rb-performance-001"
    title: "API latency spike"
    category: "performance"
    tier: "tier_2"
    symptoms:
      - "p99 latency > 2s"
      - "timeout errors"
      - "slow dashboard loads"
    steps:
      - "Check Datadog APM for bottleneck service"
      - "Verify recent deploy correlation"
      - "Check database connection pool exhaustion"
      - "Scale affected service horizontally"
    escalation_criteria: "Requires schema migration or architectural change"

# src/tools.py
from pathlib import Path
from typing import List, Optional
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
import yaml

from src.models import TicketCategory, TicketTier, TriageResult


class RunbookEntry(BaseModel):
    id: str
    title: str
    category: TicketCategory
    tier: TicketTier
    symptoms: List[str]
    steps: List[str]
    escalation_criteria: str


class RunbookStore(BaseTool):
    name: str = "runbook_store"
    _runbooks: List[RunbookEntry] = []

    def __init__(self, path: str = "config/runbooks.yaml"):
        super().__init__()
        data = yaml.safe_load(Path(path).read_text())
        self._runbooks = [RunbookEntry(**rb) for rb in data["runbooks"]]

    def _run(self, category: str, tier: str) -> str:
        matches = [
            rb for rb in self._runbooks
            if rb.category.value == category and rb.tier.value == tier
        ]
        if not matches:
            return "No runbooks found for this category/tier combination."
        return "\n\n".join(
            f"## {rb.id}: {rb.title}\n**Escalation criteria**: {rb.escalation_criteria}\n**Steps**:\n" +
            "\n".join(f"  {i+1}. {step}" for i, step in enumerate(rb.steps))
            for rb in matches
        )

    def get_by_id(self, runbook_id: str) -> Optional[RunbookEntry]:
        return next((rb for rb in self._runbooks if rb.id == runbook_id), None)


class TriageOutputTool(BaseTool):
    """Forces the triage agent to return structured output."""
    name: str = "emit_triage"
    args_schema: type[BaseModel] = TriageResult

    def _run(self, **kwargs) -> str:
        return TriageResult(**kwargs).model_dump_json()

Checkpoint: python -c "from src.tools import RunbookStore; print(RunbookStore()._run('billing', 'tier_1'))" — prints the billing runbook steps.

Step 3: Build the triage agent

The triage agent classifies every incoming ticket. Give it a focused prompt and the emit_triage tool to enforce structured output.

# src/agents.py
from crewai import Agent
from crewai.tools import BaseTool
from src.tools import RunbookStore, TriageOutputTool
from src.models import TicketTier, TicketCategory


def make_triage_agent(emit_tool: TriageOutputTool) -> Agent:
    return Agent(
        role="Support Triage Specialist",
        goal=(
            "Classify incoming support tickets into tier (tier_1, tier_2, escalate) "
            "and category (billing, auth, integration, performance, bug, unknown). "
            "Output structured classification with confidence and reasoning."
        ),
        backstory=(
            "You've triaged 10,000+ tickets for a B2B SaaS platform. "
            "You know tier_1 = runbook-resolvable (billing retries, password resets, basic config). "
            "Tier_2 = requires investigation but has known patterns (webhook failures, latency spikes). "
            "Escalate = novel bugs, security concerns, architectural decisions, or customer-threatening issues."
        ),
        tools=[emit_tool],
        verbose=True,
        allow_delegation=False,
        max_iter=3,
        llm_config={"temperature": 0.1},
    )


def make_resolver_agent(runbook_store: RunbookStore) -> Agent:
    return Agent(
        role="Tier-1 Resolution Engineer",
        goal=(
            "Execute the appropriate runbook for tier_1 tickets. "
            "Follow steps exactly. Report resolution or escalation with evidence."
        ),
        backstory=(
            "You resolve tier_1 tickets by running playbooks. You never improvise. "
            "If a runbook step fails or the customer's situation doesn't match, you escalate with context."
        ),
        tools=[runbook_store],
        verbose=True,
        allow_delegation=False,
        max_iter=5,
        llm_config={"temperature": 0.0},
    )


def make_escalation_agent() -> Agent:
    return Agent(
        role="Escalation Coordinator",
        goal=(
            "Package tier_2 and escalate tickets with complete context for human engineers. "
            "Include triage reasoning, any resolution attempts, and a recommended owner."
        ),
        backstory=(
            "You write the handoff notes that senior engineers actually read. "
            "Concise, factual, with clear priority and ownership recommendation."
        ),
        verbose=True,
        allow_delegation=False,
        max_iter=3,
        llm_config={"temperature": 0.2},
    )

Step 4: Define tasks with explicit contracts

Each task declares its input and output models. This makes the pipeline testable and debuggable.

# src/tasks.py
from crewai import Task
from src.models import TicketInput, TriageResult, Resolution, EscalationPackage
from src.agents import make_triage_agent, make_resolver_agent, make_escalation_agent
from src.tools import TriageOutputTool, RunbookStore


def make_triage_task(triage_agent, emit_tool: TriageOutputTool) -> Task:
    return Task(
        description=(
            "Classify the ticket. Use the emit_triage tool to output your decision.\n\n"
            "Ticket: {ticket_json}\n\n"
            "Classification rules:\n"
            "- tier_1: Matches a runbook exactly (billing retry, password reset, basic how-to)\n"
            "- tier_2: Known pattern but needs investigation (webhook failures, latency, integration config)\n"
            "- escalate: Security, data loss, novel bugs, legal/compliance, or customer threatens churn\n\n"
            "Categories: billing, auth, integration, performance, bug, unknown\n"
            "Confidence threshold: only emit_triage if confidence >= 0.75"
        ),
        expected_output="Structured TriageResult via emit_triage tool",
        agent=triage_agent,
        output_json=TriageResult,
        tools=[emit_tool],
    )


def make_resolution_task(resolver_agent, runbook_store: RunbookStore) -> Task:
    return Task(
        description=(
            "You receive a tier_1 ticket and its triage result. "
            "1. Use runbook_store to fetch the runbook for the category/tier.\n"
            "2. Execute each step in order. Simulate the action (e.g., 'Retried charge via Stripe: success').\n"
            "3. If all steps succeed, mark resolved=true.\n"
            "4. If any step fails or customer context doesn't match, mark resolved=false and set requires_human=true with escalation_reason.\n\n"
            "Triage: {triage_json}\n"
            "Original ticket: {ticket_json}"
        ),
        expected_output="Structured Resolution via output_json",
        agent=resolver_agent,
        output_json=Resolution,
        context=["triage_task"],
    )


def make_escalation_task(escalation_agent) -> Task:
    return Task(
        description=(
            "Create an escalation package for a human engineer.\n"
            "Inputs:\n"
            "- Original ticket: {ticket_json}\n"
            "- Triage result: {triage_json}\n"
            "- Resolution attempt (if any): {resolution_json}\n\n"
            "Produce an EscalationPackage with:\n"
            "- context_summary: 3-5 bullet points, facts only\n"
            "- recommended_owner: team name (payments, auth, platform, core-backend)\n"
            "- priority: 1-5 (5 = customer-down, 1 = low-impact)\n"
            "Be concise. Engineers skip fluff."
        ),
        expected_output="Structured EscalationPackage via output_json",
        agent=escalation_agent,
        output_json=EscalationPackage,
        context=["triage_task", "resolution_task"],
    )

Step 5: Wire the crew with conditional flow

CrewAI’s sequential process works, but we need conditional logic: only run resolution for tier_1, always run escalation for tier_2/escalate. Implement this in the crew wrapper.

# src/crew.py
from crewai import Crew, Process
from src.models import TicketInput, TriageResult, TicketTier
from src.agents import make_triage_agent, make_resolver_agent, make_escalation_agent
from src.tasks import make_triage_task, make_resolution_task, make_escalation_task
from src.tools import RunbookStore, TriageOutputTool
import json


class SupportCrew:
    def __init__(self):
        self.runbook_store = RunbookStore()
        self.emit_tool = TriageOutputTool()
        self.triage_agent = make_triage_agent(self.emit_tool)
        self.resolver_agent = make_resolver_agent(self.runbook_store)
        self.escalation_agent = make_escalation_agent()

        self.triage_task = make_triage_task(self.triage_agent, self.emit_tool)
        self.resolution_task = make_resolution_task(self.resolver_agent, self.runbook_store)
        self.escalation_task = make_escalation_task(self.escalation_agent)

    def run(self, ticket: TicketInput) -> dict:
        # Phase 1: Triage (always runs)
        triage_crew = Crew(
            agents=[self.triage_agent],
            tasks=[self.triage_task],
            process=Process.sequential,
            verbose=True,
        )
        triage_result_raw = triage_crew.kickoff(inputs={"ticket_json": ticket.model_dump_json()})
        triage = TriageResult.model_validate_json(triage_result_raw.raw)

        result = {"triage": triage.model_dump()}

        # Phase 2: Route based on tier
        if triage.tier == TicketTier.TIER_1:
            resolution_crew = Crew(
                agents=[self.resolver_agent],
                tasks=[self.resolution_task],
                process=Process.sequential,
                verbose=True,
            )
            resolution_raw = resolution_crew.kickoff(inputs={
                "triage_json": triage.model_dump_json(),
                "ticket_json": ticket.model_dump_json(),
            })
            resolution = Resolution.model_validate_json(resolution_raw.raw)
            result["resolution"] = resolution.model_dump()

            # Escalate if resolver couldn't close it
            if resolution.requires_human:
                esc_crew = Crew(
                    agents=[self.escalation_agent],
                    tasks=[self.escalation_task],
                    process=Process.sequential,
                    verbose=True,
                )
                esc_raw = esc_crew.kickoff(inputs={
                    "triage_json": triage.model_dump_json(),
                    "ticket_json": ticket.model_dump_json(),
                    "resolution_json": resolution.model_dump_json(),
                })
                result["escalation"] = json.loads(esc_raw.raw)

        else:  # TIER_2 or ESCALATE
            esc_crew = Crew(
                agents=[self.escalation_agent],
                tasks=[self.escalation_task],
                process=Process.sequential,
                verbose=True,
            )
            esc_raw = esc_crew.kickoff(inputs={
                "triage_json": triage.model_dump_json(),
                "ticket_json": ticket.model_dump_json(),
                "resolution_json": "null",
            })
            result["escalation"] = json.loads(esc_raw.raw)

        return result

Step 6: CLI entry point with sample tickets

# src/main.py
import json
from src.crew import SupportCrew
from src.models import TicketInput


SAMPLE_TICKETS = [
    TicketInput(
        id="t-1001",
        subject="Payment failed for invoice INV-4421",
        body="My corporate card was declined but it has sufficient limit. Please retry.",
        customer_tier="premium",
    ),
    TicketInput(
        id="t-1002",
        subject="Password reset email never arrives",
        body="I've requested a reset 3 times over 2 hours. Checked spam. Nothing.",
        customer_tier="standard",
    ),
    TicketInput(
        id="t-1003",
        subject="Webhook signatures failing after secret rotation",
        body="We rotated our HMAC secret yesterday per your docs. Now all webhooks fail signature verification. Our endpoint hasn't changed.",
        customer_tier="enterprise",
    ),
    TicketInput(
        id="t-1004",
        subject="API p99 latency 3.2s since this morning",
        body="Dashboard shows p99 latency jumped from 400ms to 3.2s around 09:00 UTC. No deploy on our side. Affecting all endpoints.",
        customer_tier="enterprise",
    ),
    TicketInput(
        id="t-1005",
        subject="Possible data leak in export CSV",
        body="Customer reports their export CSV contains another customer's email column. Urgent — legal team involved.",
        customer_tier="enterprise",
    ),
]


def main():
    crew = SupportCrew()
    for ticket in SAMPLE_TICKETS:
        print(f"\n{'='*60}")
        print(f"Processing {ticket.id}: {ticket.subject}")
        print(f"{'='*60}")
        result = crew.run(ticket)
        print(json.dumps(result, indent=2, default=str))


if __name__ == "__main__":
    main()

Checkpoint: Run python -m src.main. You should see each ticket flow through triage, then either resolution (tier_1) or escalation (tier_2/escalate). Example output for t-1001:

{
  "triage": {
    "ticket_id": "t-1001",
    "tier": "tier_1",
    "category": "billing",
    "confidence": 0.92,
    "reasoning": "Explicit payment failure with request to retry. Matches rb-billing-001 symptoms exactly.",
    "suggested_runbook": "rb-billing-001"
  },
  "resolution": {
    "ticket_id": "t-1001",
    "resolved": true,
    "resolution_text": "Retried charge via Stripe dashboard: succeeded. Sent confirmation to customer.",
    "actions_taken": [
      "Verified payment method on file (Visa ending 4242)",
      "Retried charge via Stripe dashboard: success",
      "Sent update-payment-method email template (not needed — retry succeeded)",
      "Confirmed resolution with customer"
    ],
    "requires_human": false
  }
}

For t-1003 (tier_2), you’ll see an escalation package:

{
  "triage": {
    "ticket_id": "t-1003",
    "tier": "tier_2",
    "category": "integration",
    "confidence": 0.88,
    "reasoning": "Webhook signature failures after secret rotation — known pattern, needs investigation of HMAC implementation.",
    "suggested_runbook": "rb-integration-001"
  },
  "escalation": {
    "ticket_id": "t-1003",
    "original_ticket": {...},
    "triage": {...},
    "resolution_attempt": null,
    "context_summary": "- Enterprise customer rotated HMAC secret per docs\n- All webhooks now fail signature verification\n- Customer endpoint unchanged\n- Matches runbook rb-integration-001 escalation criteria",
    "recommended_owner": "platform",
    "priority": 4
  }
}

Step 7: Add a test for triage accuracy

# tests/test_triage.py
import pytest
from src.crew import SupportCrew
from src.models import TicketInput, TicketTier, TicketCategory


@pytest.fixture
def crew():
    return SupportCrew()


@pytest.mark.parametrize("ticket,expected_tier,expected_category", [
    (
        TicketInput(id="t", subject="Payment declined", body="Card declined, please retry", customer_tier="standard"),
        TicketTier.TIER_1, TicketCategory.BILLING
    ),
    (
        TicketInput(id="t", subject="Reset email not received", body="No reset email after 3 tries", customer_tier="standard"),
        TicketTier.TIER_1, TicketCategory.AUTH
    ),
    (
        TicketInput(id="t", subject="Webhook timeout", body="Our endpoint returns 504", customer_tier="enterprise"),
        TicketTier.TIER_2, TicketCategory.INTEGRATION
    ),
    (
        TicketInput(id="t", subject="Data leak in export", body="CSV contains other customer data", customer_tier="enterprise"),
        TicketTier.ESCALATE, TicketCategory.BUG
    ),
])
def test_triage_classification(crew, ticket, expected_tier, expected_category):
    result = crew.run(ticket)
    triage = result["triage"]
    assert triage["tier"] == expected_tier.value
    assert triage["category"] == expected_category.value
    assert triage["confidence"] >= 0.75

Run with pytest tests/test_triage.py -v.

Step 8: Observability hooks (optional but recommended)

Wrap the crew execution to emit structured logs for your observability stack.

# src/observability.py
import time
import uuid
from contextlib import contextmanager
from typing import Generator
import structlog

logger = structlog.get_logger()


@contextmanager
def trace_ticket(ticket_id: str) -> Generator[str, None, None]:
    trace_id = str(uuid.uuid4())[:8]
    start = time.perf_counter()
    logger.info("ticket_started", ticket_id=ticket_id, trace_id=trace_id)
    try:
        yield trace_id
    except Exception as e:
        logger.exception("ticket_failed", ticket_id=ticket_id, trace_id=trace_id, error=str(e))
        raise
    finally:
        duration_ms = (time.perf_counter() - start) * 1000
        logger.info("ticket_completed", ticket_id=ticket_id, trace_id=trace_id, duration_ms=round(duration_ms, 2))

Then in crew.py:

# Inside SupportCrew.run()
from src.observability import trace_ticket

def run(self, ticket: TicketInput) -> dict:
    with trace_ticket(ticket.id):
        # ... existing logic

Extending for production

• Idempotency: Add a Redis cache keyed by ticket.id to prevent double-processing on webhook retries.
• Human-in-the-loop: Pause the escalation task and write to a queue (Slack, PagerDuty, Linear) for human approval before closing.
• Runbook versioning: Store runbooks in a database with version history; the tool fetches the latest approved version.
• Feedback loop: Log resolution outcomes (resolved/escalated/false-positive) and retrain the triage prompt monthly.
• Model routing: Route triage to a small fast model (gpt-4o-mini), resolver to a tool-capable model, escalation to a larger context model. A gateway that forwards provider cache-control hints can reduce latency on repeated runbook lookups.

What you have now

A crewai support bot tiered tickets pipeline that:

1. Triages every ticket with structured, high-confidence classification
2. Resolves tier-1 tickets by executing versioned runbooks step-by-step
3. Escalates tier-2 and complex tickets with context packages engineers actually read
4. Tests triage accuracy against known samples
5. Observes latency and outcomes per ticket

The runbook-driven resolver is the key differentiator: it replaces “LLM tries to help” with “LLM follows the procedure we wrote, reports exactly what happened, and escalates when the procedure doesn’t fit.” That’s what moves a demo to a system you can on-call.