OverviewHistoryStatsSecurity
npx skills add ...
Documentation
SKILL.md
npx skills add crewaiinc/skills --skill getting-started
CrewAI architecture decisions and project scaffolding. Use when starting a new crewAI project, choosing between LLM.call() vs Agent.kickoff() vs Crew.kickoff() vs Flow, scaffolding with 'crewai create flow', setting up YAML config (agents.yaml, tasks.yaml), wiring @CrewBase crew.py, writing Flow main.py with @start/@listen, building experimental conversational Flows with handle_turn()/chat(), or using {variable} interpolation.
npx skills add crewaiinc/skills --skill getting-started
How to choose the right abstraction, scaffold a project, and wire everything together.
NEVER manually create crewAI project files. Always scaffold with the CLI:
This is not optional. Even if you only need one crew, even if you know the file structure by heart — run the CLI first, then modify the generated files. Do NOT write main.py, crew.py, agents.yaml, tasks.yaml, or pyproject.toml by hand from scratch.
Why: The CLI sets up correct imports, directory structure, pyproject.toml config, and boilerplate that is easy to get subtly wrong when done manually. The reference material below teaches you how the pieces work so you can modify scaffolded code, not so you can replace the scaffolding step.
Workflow:
crewai create flow <name> (use underscores, not hyphens)crewai install then crewai runcrewAI has five common abstraction choices. Pick the simplest one that fits your need:
| Level | When to Use | Overhead | Example |
|---|---|---|---|
LLM.call() | Single prompt, no tools, structured extraction | Lowest | Parse an email into fields |
Agent.kickoff() | One agent with tools and reasoning, no multi-agent coordination | Low | Research a topic with web search |
Crew.kickoff() | Multiple agents collaborating on related tasks | Medium | Research + write + review pipeline |
Flow wrapping crews/agents/LLM calls | Production app with state, routing, conditionals, error handling | Full | Multi-step workflow with branching logic |
Conversational Flow | Multi-turn chat where each user line re-runs a Flow with the same session id | Full + experimental | Support assistant with routed chat, research, and escalation turns |
Rule of thumb: For any production application, always start with a Flow. You can embed LLM.call(), Agent.kickoff(), or Crew.kickoff() inside Flow steps. This gives you state management, error handling, and room to grow.
For chat applications, start with a conversational Flow rather than trying to make Crew.kickoff() or Flow.kickoff() act like a chat loop. The conversational surface is experimental, but it is the intended API for multi-turn sessions: call flow.handle_turn(message, session_id=...) for every user line, or flow.chat() for a local terminal REPL. Official guide: https://docs.crewai.com/en/guides/flows/conversational-flows.
Use for simple, single-turn tasks where you don't need tools or agent reasoning.
When NOT to use: If you need tools, multi-step reasoning, or retries — use an Agent instead.
Use when you need one agent with tools and reasoning, but don't need multi-agent coordination.
Note:
Agent.kickoff()wraps results — access structured output viaresult.pydantic. This differs fromLLM.call(), which returns the Pydantic object directly.
When NOT to use: If you need multiple agents passing context to each other — use a Crew.
As stated above: NEVER skip crewai create flow. This section documents what the CLI generates so you know what to modify — not so you can recreate it by hand.
Warning: Always use underscores in project names, not hyphens.
crewai create flow my-projectcreates a directory that is not a valid Python identifier, causingModuleNotFoundErroron import. Usemy_projectinstead.
This generates:
Do not use
crewai create crewunless you are certain you will never need routing, state, or multiple crews. Prefercrewai create flowas the default.
The scaffold uses YAML files for agent and task definitions. This separates configuration from code and supports {variable} interpolation.
Key rules:
{variable} placeholders are replaced at runtime via crew.kickoff(inputs={...})expected_output is always a string (never a Pydantic class name)agent value must match an agent key in agents.yamlProcess.sequential, each task auto-receives all prior task outputs as contextcontext=[other_task] to explicitly pass outputThe @CrewBase decorator auto-loads YAML config files and collects @agent and @task methods.
Important: Method names must match YAML keys. def researcher(self) maps to the researcher: key in agents.yaml.
Flows are the recommended way to build production crewAI applications. They provide state management, conditional routing, human-in-the-loop, and persistence — wrapping crews, agents, and LLM calls into a coherent workflow.
Key points:
flow.kickoff(inputs={"topic": "AI Agents"}) populates self.state.topic (keys must match Pydantic field names). The YAML {variable} substitution happens later, when you call crew.kickoff(inputs={"topic": self.state.topic}) inside a Flow step. The chain is: flow inputs → state → crew inputs → YAML substitution.@listen method runs after its dependency completesStructured (recommended for production):
Unstructured (quick prototyping):
Use structured state for type safety, IDE autocompletion, and validation. Use unstructured only for throwaway prototypes.
Many production Flows skip Crews entirely and orchestrate individual agents via Agent.kickoff(). This gives you fine-grained control — each Flow step calls a specific agent, passes state, and stores the result. The Flow handles orchestration; agents handle reasoning.
Why this pattern works well:
Agent.kickoff(), LLM.call(), and Crew.kickoff() freelyWhen to use Agent.kickoff() vs Crew.kickoff() in a Flow:
Use Agent.kickoff() when | Use Crew.kickoff() when |
|---|---|
| Each step is a distinct agent with different tools | Multiple agents need to collaborate on ONE task |
| You want the Flow to control sequencing | Agents need to pass context to each other within a step |
| Steps are independent and don't need inter-agent delegation | You need hierarchical process with a manager |
| You want maximum control over what data flows between steps | The sub-workflow is self-contained and reusable |
Combine response_format with state for typed data flow between agents:
A Flow can combine all crewAI abstractions in a single workflow:
@routerUse @router for conditional branching — return a string label, and @listen("label") binds to branches:
or_() and and_()@persistFor long-running workflows that need to survive restarts:
handle_turn() (Experimental)Use a conversational Flow when the product is a chat session: support assistants, routed research helpers, onboarding wizards, or any UI where the same user sends multiple turns.
Core model:
handle_turn(message, session_id=...) appends the user line to state.messages, resets per-turn execution tracking, and calls kickoff(inputs={"id": session_id}) internallyFlow.kickoff() does not accept user_message= or session_id= keyword argsroute_turn() plus @listen("ROUTE") handlersappend_assistant_message(reply) in handlers so the next turn sees assistant historytry/finally and call finalize_session_traces(); flow.chat() does this for local REPLsUse RouterConfig when you want LLM-driven routing. The router catalog is auto-built from @listen("ROUTE") handlers and their docstrings, so do not duplicate the route list in the router prompt.
See Conversational Flows for the full lifecycle, routing, persistence, and trace guidance.
@human_feedbackinputsThe {variable} pattern is how you make crews reusable.
In YAML, {topic} and {current_year} get replaced:
Common mistakes:
{variable} in the prompt{{ }} instead of single-brace { } → crewAI uses single bracesOr run directly:
| Symptom | Likely Cause | Fix |
|---|---|---|
{topic} appears literally in agent output | Missing inputs= in kickoff() | Pass crew.kickoff(inputs={"topic": "..."}) |
KeyError on self.agents_config['name'] | Method name doesn't match YAML key | Ensure @agent def researcher matches researcher: in YAML |
ModuleNotFoundError on import | Wrong path or hyphens in project name | Use underscores; check from .crews.crew_name.crew_name import CrewClass |
| Crew runs but Flow state is empty | Not writing results back to self.state | Assign crew output to self.state.field in the @listen method |
Process.SEQUENTIAL raises AttributeError | Uppercase enum | Use lowercase: Process.sequential |
| Agent ignores tools | Tools assigned to agent but task needs them | Move tools to task level or verify agent has the right tools |
| Agent fabricates search results | No tools assigned — agent can't actually search | Add tools=[SerperDevTool()] or equivalent; an agent with no tools will hallucinate data |
@listen never fires | Listener string doesn't match router return value, or passed a string instead of method reference | @router must return the exact string @listen("label") expects; for method chaining use @listen(method_ref) not @listen("method_name") |
| Flow step runs twice unexpectedly | Multiple @start() methods or or_ listener | Use and_() if you need all upstream steps to complete first |
AuthenticationError or API key not found | Missing env var | Set OPENAI_API_KEY (and SERPER_API_KEY for search tools) in .env |
| Agent retries endlessly on structured output | Pydantic model too complex for the LLM | Simplify the model, reduce nesting, or use a more capable llm |
Agent loops to max_iter without finishing | Task description too vague or conflicting with expected_output | Make expected_output specific and achievable; lower max_iter to fail faster |
| Flow state not updating across steps | Using unstructured state without proper key access | Switch to structured Pydantic state or ensure dict keys are consistent |
@router return value ignored | Method not decorated with @router | Use @router(condition) not @listen(condition) for branching methods |
Flow.kickoff(user_message=..., session_id=...) fails | Conversational kwargs are not accepted by kickoff() | Use flow.handle_turn(message, session_id=...) for chat messages |
| Chat history missing assistant replies | Handler returned text but did not record it on older/explicit paths | Call self.append_assistant_message(reply) inside route handlers |
| Trace never exports for chat session | Deferred conversational trace was not finalized | Call flow.finalize_session_traces() in finally, or use flow.chat() |
Follow-up chat modeled with @human_feedback | Human feedback approves a step output, not the next user message | Use conversational handle_turn() for follow-up chat lines |
For deeper dives into specific topics, see:
@router, or_(), and_(), @persist, streaming, and @human_feedback patternshandle_turn(), chat(), ConversationConfig, router behavior, persistence, and tracingFor related skills:
from crewai import LLM
from pydantic import BaseModel
class EmailFields(BaseModel):
sender: str
subject: str
urgency: str
llm = LLM(model="openai/gpt-4o")
# Without response_format — returns a string
raw = llm.call(messages=[{"role": "user", "content": "Summarize this text..."}])
print(raw) # str
# With response_format — returns the Pydantic object directly
result = llm.call(
messages=[{"role": "user", "content": f"Extract fields from this email: {email_text}"}],
response_format=EmailFields
)
print(result.sender) # str — access Pydantic fields directly
print(result.urgency) # strfrom crewai import Agent
from crewai_tools import SerperDevTool
from pydantic import BaseModel
class ResearchFindings(BaseModel):
main_points: list[str]
key_technologies: list[str]
researcher = Agent(
role="AI Researcher",
goal="Research the latest AI developments",
backstory="Expert AI researcher with deep technical knowledge.",
llm="openai/gpt-4o", # Optional: defaults to OPENAI_MODEL_NAME env var or "gpt-4"
tools=[SerperDevTool()],
)
# Unstructured output
result = researcher.kickoff("What are the latest LLM developments?")
print(result.raw) # str
print(result.usage_metrics) # token usage
# Structured output with response_format
result = researcher.kickoff(
"Summarize latest AI developments",
response_format=ResearchFindings,
)
print(result.pydantic.main_points)crewai create flow my_projectmy_project/
├── src/my_project/
│ ├── crews/
│ │ └── my_crew/
│ │ ├── config/
│ │ │ ├── agents.yaml # Agent definitions (role, goal, backstory)
│ │ │ └── tasks.yaml # Task definitions (description, expected_output)
│ │ └── my_crew.py # Crew class with @CrewBase
│ ├── tools/
│ │ └── custom_tool.py
│ ├── main.py # Flow class with @start/@listen
│ └── ...
├── .env # API keys (OPENAI_API_KEY, etc.)
└── pyproject.tomlresearcher:
role: >
{topic} Senior Data Researcher
goal: >
Uncover cutting-edge developments in {topic}
backstory: >
You're a seasoned researcher with a knack for uncovering
the latest developments in {topic}.
# Optional overrides:
# llm: openai/gpt-4o
# max_iter: 20
# max_rpm: 10
reporting_analyst:
role: >
{topic} Reporting Analyst
goal: >
Create detailed reports based on {topic} research findings
backstory: >
You're a meticulous analyst known for turning complex data
into clear, actionable reports.research_task:
description: >
Conduct thorough research about {topic}.
Identify key trends, breakthrough technologies,
and potential industry impacts.
expected_output: >
A detailed report with analysis of the top 5
developments in {topic}, with sources and implications.
agent: researcher
reporting_task:
description: >
Review the research and create a comprehensive report about {topic}.
expected_output: >
A polished report formatted in markdown with sections
for each key finding.
agent: reporting_analyst
output_file: output/report.mdfrom crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai_tools import SerperDevTool
@CrewBase
class ResearchCrew:
"""Research and reporting crew."""
agents_config = "config/agents.yaml"
tasks_config = "config/tasks.yaml"
@agent
def researcher(self) -> Agent:
return Agent(
config=self.agents_config["researcher"],
tools=[SerperDevTool()],
)
@agent
def reporting_analyst(self) -> Agent:
return Agent(
config=self.agents_config["reporting_analyst"],
)
@task
def research_task(self) -> Task:
return Task(config=self.tasks_config["research_task"])
@task
def reporting_task(self) -> Task:
return Task(
config=self.tasks_config["reporting_task"],
context=[self.research_task()], # Explicit dependency (optional in sequential)
output_file="output/report.md",
)
@crew
def crew(self) -> Crew:
return Crew(
agents=self.agents, # auto-collected by @agent
tasks=self.tasks, # auto-collected by @task
process=Process.sequential,
verbose=True,
)from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel
from .crews.research_crew.research_crew import ResearchCrew
class ResearchState(BaseModel):
topic: str = ""
report: str = ""
class ResearchFlow(Flow[ResearchState]):
@start()
def begin(self):
print(f"Starting research on: {self.state.topic}")
@listen(begin)
def run_research(self):
result = ResearchCrew().crew().kickoff(
inputs={"topic": self.state.topic}
)
self.state.report = result.raw
def kickoff():
flow = ResearchFlow()
flow.kickoff(inputs={"topic": "AI Agents"})
if __name__ == "__main__":
kickoff()from pydantic import BaseModel
class MyState(BaseModel):
topic: str = ""
research: str = ""
draft: str = ""
approved: bool = False
class MyFlow(Flow[MyState]):
...class MyFlow(Flow): # No type parameter — state is a dict
@start()
def begin(self):
self.state["topic"] = "AI" # dict-style accessfrom crewai import Agent, LLM
from crewai.flow.flow import Flow, listen, start
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
from pydantic import BaseModel
class ResearchState(BaseModel):
query: str = ""
raw_research: str = ""
analysis: str = ""
report: str = ""
class DeepResearchFlow(Flow[ResearchState]):
@start()
def gather_research(self):
"""Agent with tools does the actual searching."""
researcher = Agent(
role="Senior Research Analyst",
goal="Find comprehensive, factual information about the given topic",
backstory="You're an expert researcher who always cites sources and flags uncertainty.",
tools=[SerperDevTool(), ScrapeWebsiteTool()],
llm="openai/gpt-4o",
)
result = researcher.kickoff(
f"Research this topic thoroughly: {self.state.query}"
)
self.state.raw_research = result.raw
@listen(gather_research)
def analyze_findings(self):
"""A different agent analyzes the raw research — no tools needed."""
analyst = Agent(
role="Data Analyst",
goal="Extract key insights, patterns, and actionable recommendations",
backstory="You turn raw data into clear, structured analysis.",
llm="openai/gpt-4o",
)
result = analyst.kickoff(
f"Analyze these research findings and extract key insights:\n\n{self.state.raw_research}"
)
self.state.analysis = result.raw
@listen(analyze_findings)
def write_report(self):
"""A writer agent produces the final deliverable."""
writer = Agent(
role="Technical Writer",
goal="Produce clear, actionable reports for non-technical readers",
backstory="You specialize in making complex information accessible.",
llm="openai/gpt-4o",
)
result = writer.kickoff(
f"Write a comprehensive report based on this analysis:\n\n{self.state.analysis}"
)
self.state.report = result.rawclass Insights(BaseModel):
key_points: list[str]
recommendations: list[str]
confidence: float
class AnalysisFlow(Flow[AnalysisState]):
@start()
def research(self):
researcher = Agent(role="Researcher", goal="...", backstory="...", tools=[SerperDevTool()])
result = researcher.kickoff(
f"Research {self.state.topic}",
response_format=Insights,
)
# result.pydantic gives you the typed Insights object
self.state.key_points = result.pydantic.key_points
self.state.recommendations = result.pydantic.recommendationsclass ProductFlow(Flow[ProductState]):
@start()
def classify_request(self):
# LLM.call() for simple classification
llm = LLM(model="openai/gpt-4o")
self.state.category = llm.call(
messages=[{"role": "user", "content": f"Classify: {self.state.request}"}],
response_format=Category
).category
@router(classify_request)
def route_by_category(self):
if self.state.category == "simple":
return "quick_answer"
return "deep_research"
@listen("quick_answer")
def handle_simple(self):
# Agent.kickoff() for single-agent work
agent = Agent(role="Helper", goal="Answer quickly", backstory="...")
result = agent.kickoff(self.state.request)
self.state.answer = result.raw
@listen("deep_research")
def handle_complex(self):
# Crew.kickoff() for multi-agent collaboration
result = ResearchCrew().crew().kickoff(
inputs={"topic": self.state.request}
)
self.state.answer = result.rawfrom crewai.flow.flow import Flow, listen, router, start, or_
class QualityFlow(Flow[QAState]):
@start()
def generate_content(self):
result = WriterCrew().crew().kickoff(inputs={"topic": self.state.topic})
self.state.draft = result.raw
@router(generate_content)
def check_quality(self):
llm = LLM(model="openai/gpt-4o")
score = llm.call(
messages=[{"role": "user", "content": f"Rate 1-10: {self.state.draft}"}],
response_format=QualityScore
)
if score.rating >= 7:
return "approved"
return "needs_revision"
@listen("approved")
def publish(self):
self.state.published = True
@listen("needs_revision")
def revise(self):
result = EditorCrew().crew().kickoff(
inputs={"draft": self.state.draft}
)
self.state.draft = result.rawfrom crewai.flow.flow import Flow, listen, start, or_, and_
class ParallelFlow(Flow[MyState]):
@start()
def fetch_data_a(self):
...
@start()
def fetch_data_b(self):
...
# Runs when BOTH fetches complete
@listen(and_(fetch_data_a, fetch_data_b))
def merge_results(self):
...
# Runs when EITHER source provides data
@listen(or_(fetch_data_a, fetch_data_b))
def process_first_available(self):
...from crewai.flow.flow import Flow, start, listen, persist
from crewai.flow.persistence import SQLiteFlowPersistence
@persist(SQLiteFlowPersistence()) # Class-level: persists all methods
class LongRunningFlow(Flow[MyState]):
@start()
def step_one(self):
self.state.data = "processed"
@listen(step_one)
def step_two(self):
# If the process crashes here, restarting with the same
# state ID will resume from after step_one
...from uuid import uuid4
from crewai import Agent, Flow
from crewai.flow import listen
from crewai.experimental.conversational import (
ConversationConfig,
ConversationState,
)
@ConversationConfig(defer_trace_finalization=True)
class SupportFlow(Flow[ConversationState]):
conversational = True
def research_agent(self) -> Agent:
return Agent(
role="Support Research Specialist",
goal="Answer the user's current research question with accurate sources.",
backstory="You are precise, evidence-driven, and explicit about uncertainty.",
tools=[...],
)
def route_turn(self, context):
message = (self.state.current_user_message or "").lower()
if "docs" in message or "crewai" in message:
return "CREWAI_DOCS"
if "research" in message or "search" in message:
return "RESEARCH"
return "converse"
@listen("CREWAI_DOCS")
def handle_docs(self):
"""Look up CrewAI documentation for framework/API questions."""
reply = "I would query the CrewAI docs here."
self.append_assistant_message(reply)
return reply
@listen("RESEARCH")
def handle_research(self):
"""Fresh research, current lookups, and tool-backed investigation."""
result = self.research_agent().kickoff(self.state.current_user_message)
reply = result.raw
self.append_assistant_message(reply)
return reply
flow = SupportFlow()
session_id = str(uuid4())
try:
flow.handle_turn("What can you do?", session_id=session_id)
flow.handle_turn("Check the CrewAI docs for flows.", session_id=session_id)
finally:
flow.finalize_session_traces()from crewai.flow.flow import Flow, start, listen, router
from crewai.flow.human_feedback import human_feedback
class ApprovalFlow(Flow[ReviewState]):
@start()
def generate_draft(self):
result = WriterCrew().crew().kickoff(inputs={"topic": self.state.topic})
self.state.draft = result.raw
@human_feedback(
message="Review the draft and provide feedback",
emit=["approved", "needs_revision"],
llm="openai/gpt-4o",
default_outcome="approved"
)
@listen(generate_draft)
def review_step(self):
return self.state.draft
@listen("approved")
def publish(self):
...
@listen("needs_revision")
def revise(self):
feedback = self.last_human_feedback
# Use feedback.feedback_text for revision
...flow = MyFlow()
flow.plot() # Display in notebook
flow.plot("my_flow") # Save as my_flow.png# Variables flow through: kickoff → YAML templates → agent/task prompts
crew.kickoff(inputs={
"topic": "AI Agents",
"current_year": "2025",
"target_audience": "developers",
})research_task:
description: >
Research {topic} trends for {current_year},
targeting {target_audience}.# Install dependencies
crewai install
# Run the flow
crewai runcd my_project
uv run src/my_project/main.py