crewai-expert

name	crewai-expert
description	CrewAI development expertise — Flows, Crews, Agents, Tasks, Tools, Memory, and production patterns. Use whenever writing, editing, debugging, reviewing, or discussing CrewAI code (crewai imports, Flow/Crew/Agent/Task classes, crew.jsonc, agents.yaml, @start/@listen/@router decorators, crewai CLI commands).
paths	["**/crew.jsonc","**/agents/*.jsonc","**/agents.yaml","**/tasks.yaml","**/crew.py"]
allowed-tools	Bash(${CLAUDE_SKILL_DIR}/scripts/*), Bash(crewai *), Bash(uv *)

CrewAI Development

Apply these rules to ALL CrewAI work. For deep topics, load exactly the reference file listed in the pointer map at the bottom — do not guess APIs from memory; the framework changes fast.

Architecture decision framework

Official CrewAI guidance: start every production application with a Flow. Use a Crew inside a Flow step only when that step genuinely needs a team of agents acting autonomously.

Complexity	Precision needed	Build
Low	High	Flow with direct LLM calls (llm.call(...)) — no agents
Low	Low	Single Agent (agent.kickoff(...))
High	Low	Crew (agents collaborate, emergent behavior)
High	High	Flow orchestrating Crews — the default production shape

Deterministic steps rule: anything that does not require an LLM — parsing, validation, file/database I/O, branching logic, math, formatting — must be a plain Python method inside the Flow. Never spend an agent (or any LLM call) on deterministic work. A @router whose decision is computable from state must be plain Python returning a label, not an LLM call.

Escalate intelligence only as needed within a step:

1. Plain Python (free, instant, deterministic)
2. Direct LLM.call() (one model call, no agent loop)
3. Single Agent with tools (one agent, tool use)
4. Crew (multiple agents, delegation, autonomy)

Golden rules

1. 80/20 rule: task design matters more than agent design. One task = one outcome. Never write "god tasks" that research AND analyze AND write. Split them.
2. Every Task gets a specific expected_output describing what "done" looks like (format, length, sections). Vague expected_output is the #1 cause of bad results.
3. Structured outputs for anything consumed downstream: use output_pydantic=MyModel (or output_json). Never parse free text from a previous task when a Pydantic model can carry it.
4. Guardrails on tasks whose output feeds later steps: function guardrail for verifiable format rules (returns (bool, Any)), string guardrail (LLM-based) for subjective criteria. guardrails=[...] (list) takes precedence over guardrail (single).
5. Typed Flow state: production flows use class MyFlow(Flow[MyState]) with a Pydantic MyState — never ad-hoc self.state["key"] dicts.
6. Claude LLM config: max_tokens is REQUIRED for all Anthropic models. Always use provider-prefixed IDs (anthropic/claude-sonnet-4-6). Install the extra: uv add "crewai[anthropic]".
7. Specialists over generalists: agents get a specific role/goal/backstory ("Senior SQL Performance Analyst", not "Helpful Assistant"). Use process="sequential" unless coordination genuinely needs a hierarchical manager (which then requires manager_llm).
8. Config lives in JSONC (the current format): crew.jsonc + agents/*.jsonc, loaded with load_crew(...). YAML + @CrewBase is the --classic legacy scaffold — fine to maintain, don't create new projects with it unless asked.

Quick-pattern cheat sheet

Minimal production shape — Flow orchestrating a JSONC-configured crew:

# src/my_project/main.py
from pathlib import Path
from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, start, router


class PipelineState(BaseModel):
    topic: str = ""
    research: str = ""
    is_urgent: bool = False


class Pipeline(Flow[PipelineState]):

    @start()
    def load_input(self):
        # Deterministic step: plain Python, no LLM
        self.state.topic = self.state.topic.strip().lower()

    @listen(load_input)
    def research(self):
        # Agentic step: delegate to a crew
        from my_project.crews.research_crew.research_crew import build_crew
        result = build_crew().kickoff(inputs={"topic": self.state.topic})
        self.state.research = result.raw
        return result

    @router(research)
    def triage(self):
        # Deterministic routing: computable from state → plain Python
        return "urgent" if self.state.is_urgent else "standard"

    @listen("urgent")
    def escalate(self):
        ...

    @listen("standard")
    def publish(self):
        ...


def kickoff():
    Pipeline().kickoff(inputs={"topic": "AI agents"})

// src/my_project/crews/research_crew/crew.jsonc
{
  "agents": [{ "file": "agents/researcher.jsonc" }],
  "tasks": [
    {
      "description": "Research {topic}: find the 5 most significant recent developments.",
      "expected_output": "A markdown list of 5 developments, each with a 2-3 sentence summary and source.",
      "agent": { "ref": "researcher" }
    }
  ],
  "process": "sequential"
}

// src/my_project/crews/research_crew/agents/researcher.jsonc
{
  "role": "{topic} Senior Researcher",
  "goal": "Uncover accurate, current developments in {topic}",
  "backstory": "A meticulous analyst known for verifiable sourcing.",
  "llm": "anthropic/claude-sonnet-4-6",
  "verbose": true
}

# src/my_project/crews/research_crew/research_crew.py
from pathlib import Path
from crewai import load_crew

def build_crew():
    return load_crew(Path(__file__).with_name("crew.jsonc"))

Key flow imports (verified): from crewai.flow.flow import Flow, listen, start, or_, and_, router, from crewai.flow.persistence import persist, from crewai.flow.human_feedback import human_feedback.

Bundled scripts — use these instead of hand-typing CLI commands

• bash ${CLAUDE_SKILL_DIR}/scripts/check-env.sh — run BEFORE any CrewAI work in a project you haven't touched this session: verifies Python 3.10–3.13, uv, crewai CLI, and API key presence.
• bash ${CLAUDE_SKILL_DIR}/scripts/run.sh — crewai run with logged output (prints the log path).
• bash ${CLAUDE_SKILL_DIR}/scripts/test.sh [iterations] [model] — crewai test wrapper.
• bash ${CLAUDE_SKILL_DIR}/scripts/replay.sh [task_id] — lists latest task outputs; replays from a task id when given.
• bash ${CLAUDE_SKILL_DIR}/scripts/plot-flow.sh — generates the flow HTML diagram and prints its path. Use when debugging routing.

Pointer map — load exactly one reference per deep topic

When the work involves...	Read
Flow decorators, state, routers, or_/and_, persistence/resume, streaming, @human_feedback, plot()	references/flows.md
Agent/Task/Crew parameters, JSONC config format, @CrewBase classic pattern, processes, conditional tasks, guardrails, structured outputs, callbacks	references/crews.md
Custom tools (BaseTool, @tool), args_schema, caching, async tools, crewai-tools catalog, MCP servers, tool hooks	references/tools.md
Memory (unified Memory class, embedders, reset), Knowledge sources, flow remember()/recall()	references/memory-knowledge.md
LLM class, Anthropic/Claude setup, extended thinking, provider prefixes, .env conventions, LiteLLM migration	references/llm-config.md
crewai test/train/replay, event listeners, execution hooks, observability, HITL webhooks, rate limits, deployment, full CLI table	references/production.md

Common pitfalls (reject these in any code you write or review)

• God tasks; vague expected_output; free-text parsing between tasks instead of output_pydantic.
• LLM-based @router for decisions computable from state.
• allow_delegation=True without a reason (causes delegation loops and question spam).
• hierarchical process without manager_llm, or used where sequential suffices.
• Anthropic model without max_tokens; model IDs missing the anthropic/ prefix.
• Deprecated LiteLLM prefixes (ollama/, groq/, mistral/...) — native providers replaced LiteLLM; see llm-config reference.
• Unstructured self.state["..."] dict state in production flows.
• Secrets committed: .env must be gitignored; keys only via environment.
• Large knowledge_sources re-embedded on every kickoff (see memory-knowledge reference for the fix).