1# Core framework
2pip install crewai
3
4# With 50+ built-in tools
5pip install 'crewai[tools]'

1# Create new crew project
2crewai create crew my_project
3cd my_project
4
5# Install dependencies
6crewai install
7
8# Run the crew
9crewai run

 1from crewai import Agent, Task, Crew, Process
 2
 3# 1. Define agents
 4researcher = Agent(
 5    role="Senior Research Analyst",
 6    goal="Discover cutting-edge developments in AI",
 7    backstory="You are an expert analyst with a keen eye for emerging trends.",
 8    verbose=True
 9)
10
11writer = Agent(
12    role="Technical Writer",
13    goal="Create clear, engaging content about technical topics",
14    backstory="You excel at explaining complex concepts to general audiences.",
15    verbose=True
16)
17
18# 2. Define tasks
19research_task = Task(
20    description="Research the latest developments in {topic}. Find 5 key trends.",
21    expected_output="A detailed report with 5 bullet points on key trends.",
22    agent=researcher
23)
24
25write_task = Task(
26    description="Write a blog post based on the research findings.",
27    expected_output="A 500-word blog post in markdown format.",
28    agent=writer,
29    context=[research_task]  # Uses research output
30)
31
32# 3. Create and run crew
33crew = Crew(
34    agents=[researcher, writer],
35    tasks=[research_task, write_task],
36    process=Process.sequential,  # Tasks run in order
37    verbose=True
38)
39
40# 4. Execute
41result = crew.kickoff(inputs={"topic": "AI Agents"})
42print(result.raw)

 1from crewai import Agent
 2
 3agent = Agent(
 4    role="Data Scientist",                    # Job title/role
 5    goal="Analyze data to find insights",     # What they aim to achieve
 6    backstory="PhD in statistics...",         # Background context
 7    llm="gpt-4o",                             # LLM to use
 8    tools=[],                                 # Tools available
 9    memory=True,                              # Enable memory
10    verbose=True,                             # Show reasoning
11    allow_delegation=True,                    # Can delegate to others
12    max_iter=15,                              # Max reasoning iterations
13    max_rpm=10                                # Rate limit
14)

 1from crewai import Task
 2
 3task = Task(
 4    description="Analyze the sales data for Q4 2024. {context}",
 5    expected_output="A summary report with key metrics and trends.",
 6    agent=analyst,                            # Assigned agent
 7    context=[previous_task],                  # Input from other tasks
 8    output_file="report.md",                  # Save to file
 9    async_execution=False,                    # Run synchronously
10    human_input=False                         # No human approval needed
11)

 1from crewai import Crew, Process
 2
 3crew = Crew(
 4    agents=[researcher, writer, editor],      # Team members
 5    tasks=[research, write, edit],            # Tasks to complete
 6    process=Process.sequential,               # Or Process.hierarchical
 7    verbose=True,
 8    memory=True,                              # Enable crew memory
 9    cache=True,                               # Cache tool results
10    max_rpm=10,                               # Rate limit
11    share_crew=False                          # Opt-in telemetry
12)
13
14# Execute with inputs
15result = crew.kickoff(inputs={"topic": "AI trends"})
16
17# Access results
18print(result.raw)                             # Final output
19print(result.tasks_output)                    # All task outputs
20print(result.token_usage)                     # Token consumption

1crew = Crew(
2    agents=[researcher, writer],
3    tasks=[research_task, write_task],
4    process=Process.sequential  # Task 1 → Task 2 → Task 3
5)

1crew = Crew(
2    agents=[researcher, writer, analyst],
3    tasks=[research_task, write_task, analyze_task],
4    process=Process.hierarchical,  # Manager delegates tasks
5    manager_llm="gpt-4o"           # LLM for manager
6)

1pip install 'crewai[tools]'

 1from crewai_tools import (
 2    SerperDevTool,           # Web search
 3    ScrapeWebsiteTool,       # Web scraping
 4    FileReadTool,            # Read files
 5    PDFSearchTool,           # Search PDFs
 6    WebsiteSearchTool,       # Search websites
 7    CodeDocsSearchTool,      # Search code docs
 8    YoutubeVideoSearchTool,  # Search YouTube
 9)
10
11# Assign tools to agent
12researcher = Agent(
13    role="Researcher",
14    goal="Find accurate information",
15    backstory="Expert at finding data online.",
16    tools=[SerperDevTool(), ScrapeWebsiteTool()]
17)

 1from crewai.tools import BaseTool
 2from pydantic import Field
 3
 4class CalculatorTool(BaseTool):
 5    name: str = "Calculator"
 6    description: str = "Performs mathematical calculations. Input: expression"
 7
 8    def _run(self, expression: str) -> str:
 9        try:
10            result = eval(expression)
11            return f"Result: {result}"
12        except Exception as e:
13            return f"Error: {str(e)}"
14
15# Use custom tool
16agent = Agent(
17    role="Analyst",
18    goal="Perform calculations",
19    tools=[CalculatorTool()]
20)

 1researcher:
 2  role: "{topic} Senior Data Researcher"
 3  goal: "Uncover cutting-edge developments in {topic}"
 4  backstory: >
 5    You're a seasoned researcher with a knack for uncovering
 6    the latest developments in {topic}. Known for your ability
 7    to find relevant information and present it clearly.
 8
 9reporting_analyst:
10  role: "Reporting Analyst"
11  goal: "Create detailed reports based on research data"
12  backstory: >
13    You're a meticulous analyst who transforms raw data into
14    actionable insights through well-structured reports.

 1research_task:
 2  description: >
 3    Conduct thorough research about {topic}.
 4    Find the most relevant information for {year}.
 5  expected_output: >
 6    A list with 10 bullet points of the most relevant
 7    information about {topic}.
 8  agent: researcher
 9
10reporting_task:
11  description: >
12    Review the research and create a comprehensive report.
13    Focus on key findings and recommendations.
14  expected_output: >
15    A detailed report in markdown format with executive
16    summary, findings, and recommendations.
17  agent: reporting_analyst
18  output_file: report.md

 1from crewai import Agent, Crew, Process, Task
 2from crewai.project import CrewBase, agent, crew, task
 3from crewai_tools import SerperDevTool
 4
 5@CrewBase
 6class MyProjectCrew:
 7    """My Project crew"""
 8
 9    @agent
10    def researcher(self) -> Agent:
11        return Agent(
12            config=self.agents_config['researcher'],
13            tools=[SerperDevTool()],
14            verbose=True
15        )
16
17    @agent
18    def reporting_analyst(self) -> Agent:
19        return Agent(
20            config=self.agents_config['reporting_analyst'],
21            verbose=True
22        )
23
24    @task
25    def research_task(self) -> Task:
26        return Task(config=self.tasks_config['research_task'])
27
28    @task
29    def reporting_task(self) -> Task:
30        return Task(
31            config=self.tasks_config['reporting_task'],
32            output_file='report.md'
33        )
34
35    @crew
36    def crew(self) -> Crew:
37        return Crew(
38            agents=self.agents,
39            tasks=self.tasks,
40            process=Process.sequential,
41            verbose=True
42        )

 1from my_project.crew import MyProjectCrew
 2
 3def run():
 4    inputs = {
 5        'topic': 'AI Agents',
 6        'year': 2025
 7    }
 8    MyProjectCrew().crew().kickoff(inputs=inputs)
 9
10if __name__ == "__main__":
11    run()

 1from crewai.flow.flow import Flow, listen, start, router
 2from pydantic import BaseModel
 3
 4class MyState(BaseModel):
 5    confidence: float = 0.0
 6
 7class MyFlow(Flow[MyState]):
 8    @start()
 9    def gather_data(self):
10        return {"data": "collected"}
11
12    @listen(gather_data)
13    def analyze(self, data):
14        self.state.confidence = 0.85
15        return analysis_crew.kickoff(inputs=data)
16
17    @router(analyze)
18    def decide(self):
19        return "high" if self.state.confidence > 0.8 else "low"
20
21    @listen("high")
22    def generate_report(self):
23        return report_crew.kickoff()
24
25# Run flow
26flow = MyFlow()
27result = flow.kickoff()

 1# Enable all memory types
 2crew = Crew(
 3    agents=[researcher],
 4    tasks=[research_task],
 5    memory=True,           # Enable memory
 6    embedder={             # Custom embeddings
 7        "provider": "openai",
 8        "config": {"model": "text-embedding-3-small"}
 9    }
10)

1from crewai import LLM
2
3llm = LLM(model="gpt-4o")                              # OpenAI (default)
4llm = LLM(model="claude-sonnet-4-5-20250929")                       # Anthropic
5llm = LLM(model="ollama/llama3.1", base_url="http://localhost:11434")  # Local
6llm = LLM(model="azure/gpt-4o", base_url="https://...")              # Azure
7
8agent = Agent(role="Analyst", goal="Analyze data", llm=llm)

1agent = Agent(
2    role="...",
3    max_iter=10,           # Limit iterations
4    max_rpm=5              # Rate limit
5)

1task2 = Task(
2    description="...",
3    context=[task1],       # Explicitly pass context
4    agent=writer
5)

1# Use environment variable for storage
2import os
3os.environ["CREWAI_STORAGE_DIR"] = "./my_storage"