CrewAI: Agents, Tasks, Flows and Crews Reference Guide

# pip — core + extras you'll likely need
pip install crewai
pip install 'crewai[tools]'                 # built-in tool suite
pip install crewai-tools                    # standalone tools package

# env — pick the providers you'll use
export OPENAI_API_KEY=sk-...
export ANTHROPIC_API_KEY=sk-...
export SERPER_API_KEY=...                   # optional, web search tool

# scaffold a new project (recommended layout)
crewai create crew my_project
cd my_project && crewai install
crewai run

from crewai.flow.flow import Flow, listen, start, router
from pydantic import BaseModel

class State(BaseModel):
    topic: str = ""
    draft: str = ""
    score: int = 0

class ContentFlow(Flow[State]):

    @start()
    def pick_topic(self):
        self.state.topic = "vector databases in 2026"

    @listen(pick_topic)
    def draft(self):
        self.state.draft = ContentCrew().crew().kickoff(
            inputs={"topic": self.state.topic}
        ).raw

    @router(draft)
    def review(self):
        self.state.score = score_draft(self.state.draft)
        return "ship" if self.state.score >= 8 else "revise"

    @listen("revise")
    def revise(self):
        # loops back into draft with feedback
        ...

    @listen("ship")
    def publish(self):
        publish_to_cms(self.state.draft)

ContentFlow().kickoff()

from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class SearchArgs(BaseModel):
    query: str = Field(..., description="What to look up")
    top_k: int = Field(5, description="Max results")

class InternalDocs(BaseTool):
    name: str = "internal_docs_search"
    description: str = "Search the company wiki. Use for product or policy questions."
    args_schema: type[BaseModel] = SearchArgs

    def _run(self, query: str, top_k: int = 5) -> str:
        # Replace with your real retriever.
        hits = my_vector_store.search(query, k=top_k)
        return "\n\n".join(f"- {h.title}: {h.snippet}" for h in hits)

# Attach to an agent
support_agent = Agent(
    role="Support Specialist",
    goal="Answer customer questions using internal docs",
    backstory="Polite, exact, never invents URLs.",
    tools=[InternalDocs()],
)

# config/agents.yaml — declarative agent definitions
researcher:
  role: >
    Senior {topic} Researcher
  goal: >
    Surface the 5 most consequential developments in {topic} this quarter
  backstory: >
    A meticulous analyst who values primary sources over hot takes.
  llm: openai/gpt-4o-mini
  verbose: true

writer:
  role: >
    Tech Editor
  goal: >
    Turn the research brief into a 600-word post for a developer audience
  backstory: >
    Former staff writer at a respected eng newsletter. Cuts filler.
  llm: anthropic/claude-sonnet-4-6

# config/tasks.yaml — declarative task definitions
research_task:
  description: >
    Investigate recent advances in {topic}. Pull 5 primary sources from the
    last 90 days and summarise each in 2-3 sentences.
  expected_output: >
    A markdown list of 5 items, each with title, URL, and a 2-sentence summary.
  agent: researcher

write_task:
  description: >
    Using the research brief, write a 600-word post for working developers.
    Lead with the most actionable finding.
  expected_output: >
    A markdown post with H2 sections and code blocks where relevant.
  agent: writer
  context: [research_task]
  output_file: out/post.md

from crewai import Agent, Crew, Task, Process
from crewai.project import CrewBase, agent, crew, task

@CrewBase
class ContentCrew:
    """Research + write crew driven by YAML configs."""
    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"

    @agent
    def researcher(self) -> Agent:
        return Agent(config=self.agents_config["researcher"], verbose=True)

    @agent
    def writer(self) -> Agent:
        return Agent(config=self.agents_config["writer"], verbose=True)

    @task
    def research_task(self) -> Task:
        return Task(config=self.tasks_config["research_task"])

    @task
    def write_task(self) -> Task:
        return Task(config=self.tasks_config["write_task"])

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True,
        )

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

search = SerperDevTool()

researcher = Agent(
    role="Senior Researcher",
    goal="Surface the most consequential developments in {topic}",
    backstory="A meticulous analyst who prefers primary sources.",
    tools=[search],
    llm="openai/gpt-4o-mini",
    verbose=True,
)

writer = Agent(
    role="Tech Editor",
    goal="Turn the brief into a tight 600-word post",
    backstory="Former newsletter staff writer. Cuts filler.",
    llm="anthropic/claude-sonnet-4-6",
)

brief = Task(
    description="Investigate {topic}. Five primary sources, 2-3 sentences each.",
    expected_output="Markdown list with title, URL, summary.",
    agent=researcher,
)

post = Task(
    description="Write a 600-word post from the brief. Lead with the action.",
    expected_output="Markdown with H2 sections.",
    agent=writer,
    context=[brief],
    output_file="out/post.md",
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[brief, post],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "vector databases in 2026"})
print(result.raw)