CrewAI Developer Guide

Overview

CrewAI is a lean, lightning-fast Python framework for building collaborative AI agent teams and structured workflows. It empowers developers to create autonomous AI agents with specific roles, tools, and goals that work together to tackle complex tasks. This skill covers Crews (autonomous collaboration), Flows (structured orchestration), agents, tasks, and enterprise deployment.

Core Concepts

Agents: Specialized Team Members

Agents are autonomous AI units with specific roles, goals, and capabilities.

from crewai import Agent

Create a research agent

researcher = Agent( role='Senior Research Analyst', goal='Uncover cutting-edge developments in AI and data science', backstory="""You are an expert at a leading tech think tank. Your expertise lies in identifying emerging trends and technologies in AI, data science, and machine learning.""", verbose=True, allow_delegation=False, tools=[search_tool, scrape_tool] )

Create a writer agent

writer = Agent( role='Tech Content Strategist', goal='Craft compelling content on tech advancements', backstory="""You are a renowned content strategist, known for your insightful and engaging articles on technology and innovation. You transform complex concepts into compelling narratives.""", verbose=True, allow_delegation=True, tools=[write_tool] )

Agent Key Properties

agent = Agent( role='Role Name', # The agent's job title goal='Specific objective', # What the agent aims to achieve backstory='Background story', # Context and expertise verbose=True, # Enable detailed logging allow_delegation=False, # Can delegate tasks to other agents tools=[tool1, tool2], # Available tools llm=custom_llm, # Custom LLM configuration max_iter=15, # Maximum iterations for task max_rpm=10, # Rate limit (requests per minute) memory=True, # Enable memory cache=True, # Enable response caching system_template="template", # Custom system prompt template prompt_template="template", # Custom prompt template response_template="template" # Custom response template )

Tasks: Individual Assignments

Tasks define specific work to be completed by agents.

from crewai import Task

Research task

research_task = Task( description="""Conduct a comprehensive analysis of the latest advancements in AI. Identify key trends, breakthrough technologies, and potential industry impacts. Compile your findings in a detailed report.""", expected_output='A comprehensive 3-paragraph report on AI advancements', agent=researcher, tools=[search_tool], output_file='research_report.md' )

Writing task

write_task = Task( description="""Using the research analyst's report, develop an engaging blog post highlighting the most significant AI advancements. Make it accessible and engaging for a general audience.""", expected_output='A 4-paragraph blog post about AI advancements', agent=writer, context=[research_task], # Depends on research_task output output_file='blog_post.md' )

Task Key Properties

task = Task( description='Detailed task description', expected_output='Clear output format', agent=agent_instance, tools=[tool1, tool2], # Task-specific tools context=[previous_task], # Dependencies async_execution=False, # Run asynchronously output_json=OutputClass, # Structured output (Pydantic) output_pydantic=OutputClass, # Pydantic validation output_file='result.txt', # Save output to file callback=callback_function, # Callback on completion human_input=False # Request human feedback )

Crews: Organizing Agent Teams

Crews orchestrate agents working together toward a common goal.

from crewai import Crew, Process

Create a crew

crew = Crew( agents=[researcher, writer], tasks=[research_task, write_task], process=Process.sequential, # or Process.hierarchical verbose=True, memory=True, cache=True, max_rpm=10, share_crew=False )

Kickoff the crew

result = crew.kickoff() print(result)

Kickoff with custom inputs

result = crew.kickoff(inputs={ 'topic': 'Artificial Intelligence', 'audience': 'developers' })

Process Types

Sequential process (tasks run one after another)

crew = Crew( agents=[agent1, agent2], tasks=[task1, task2], process=Process.sequential )

Hierarchical process (manager delegates to agents)

crew = Crew( agents=[agent1, agent2], tasks=[task1, task2], process=Process.hierarchical, manager_llm='gpt-4' # Required for hierarchical )

Flows: Structured Workflow Orchestration

Flows provide event-driven, deterministic control over execution paths.

from crewai.flow.flow import Flow, listen, start

class BlogPostFlow(Flow):

@start()
def fetch_topic(self):
    """Entry point - fetch the topic to write about"""
    print("Starting blog post generation")
    return "AI advancements in 2024"

@listen(fetch_topic)
def research_topic(self, topic):
    """Research the topic"""
    print(f"Researching: {topic}")
    # Integrate with Crew for autonomous research
    research_crew = Crew(
        agents=[researcher],
        tasks=[research_task]
    )
    result = research_crew.kickoff(inputs={'topic': topic})
    return result

@listen(research_topic)
def write_blog_post(self, research_data):
    """Write the blog post"""
    print("Writing blog post...")
    write_crew = Crew(
        agents=[writer],
        tasks=[write_task]
    )
    result = write_crew.kickoff(inputs={'research': research_data})
    return result

@listen(write_blog_post)
def finalize(self, blog_post):
    """Finalize and save"""
    print("Blog post completed!")
    return blog_post

Execute flow

flow = BlogPostFlow() result = flow.kickoff()

Flow State Management

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

class ArticleState(BaseModel): topic: str = "" research: str = "" draft: str = "" final: str = ""

class ArticleFlow(Flow[ArticleState]):

@start()
def set_topic(self):
    self.state.topic = "AI Ethics"
    return self.state.topic

@listen(set_topic)
def research(self, topic):
    # Research logic
    self.state.research = "Research findings..."
    return self.state.research

@listen(research)
def write_draft(self, research):
    self.state.draft = "Draft content..."
    return self.state.draft

Access state

flow = ArticleFlow() flow.kickoff() print(flow.state.topic) print(flow.state.research)

Router Pattern

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

class ContentFlow(Flow):

@start()
def categorize_content(self):
    return "technical"  # or "marketing", "blog"

@router(categorize_content)
def route_content(self, category):
    if category == "technical":
        return "write_technical"
    elif category == "marketing":
        return "write_marketing"
    else:
        return "write_blog"

@listen("write_technical")
def write_technical_doc(self):
    return "Technical documentation..."

@listen("write_marketing")
def write_marketing_copy(self):
    return "Marketing content..."

@listen("write_blog")
def write_blog_post(self):
    return "Blog post..."

Tools: Extending Agent Capabilities

Built-in Tools

from crewai_tools import ( SerperDevTool, # Google search ScrapeWebsiteTool, # Web scraping FileReadTool, # Read files DirectoryReadTool, # Read directories CodeDocsSearchTool, # Search code documentation CSVSearchTool, # Search CSV files JSONSearchTool, # Search JSON files MDXSearchTool, # Search MDX files PDFSearchTool, # Search PDF files TXTSearchTool, # Search text files WebsiteSearchTool, # Search websites SeleniumScrapingTool, # Browser automation YoutubeChannelSearchTool, # YouTube search YoutubeVideoSearchTool # YouTube video search )

Using tools

search_tool = SerperDevTool() scrape_tool = ScrapeWebsiteTool() file_tool = FileReadTool()

agent = Agent( role='Researcher', tools=[search_tool, scrape_tool, file_tool] )

Custom Tools

from crewai_tools import BaseTool

class MyCustomTool(BaseTool): name: str = "Custom Tool Name" description: str = "Clear description of what the tool does"

def _run(self, argument: str) -> str:
    # Implementation
    result = perform_operation(argument)
    return result

Using custom tool

custom_tool = MyCustomTool() agent = Agent( role='Specialist', tools=[custom_tool] )

Function as Tool

from crewai import Agent

def calculate_sum(a: int, b: int) -> int: """Calculate the sum of two numbers""" return a + b

agent = Agent( role='Calculator', tools=[calculate_sum] # Pass function directly )

Memory: Learning from Past Interactions

from crewai import Crew, Agent, Task

Enable crew memory

crew = Crew( agents=[agent1, agent2], tasks=[task1, task2], memory=True, # Enable all memory types verbose=True )

Configure specific memory types

crew = Crew( agents=[agent1, agent2], tasks=[task1, task2], memory=True, memory_config={ 'short_term': True, # Remember within single run 'long_term': True, # Remember across runs 'entity': True # Remember entities (people, places) } )

Knowledge: RAG Integration

from crewai import Agent, Crew, Task, knowledge

Create knowledge source

docs_knowledge = knowledge.StringKnowledgeSource( content="Company policies and procedures...", metadata={"source": "policy_docs"} )

Using knowledge in agent

agent = Agent( role='Policy Expert', goal='Answer questions about company policies', backstory='Expert in company policies', knowledge_sources=[docs_knowledge] )

Load knowledge from files

pdf_knowledge = knowledge.PDFKnowledgeSource( file_path='./documents/handbook.pdf' )

txt_knowledge = knowledge.TextKnowledgeSource( file_path='./documents/faq.txt' )

agent = Agent( role='Support Agent', knowledge_sources=[pdf_knowledge, txt_knowledge] )

Structured Outputs with Pydantic

from pydantic import BaseModel from crewai import Task, Agent

class BlogPost(BaseModel): title: str content: str tags: list[str] word_count: int

Task with structured output

write_task = Task( description='Write a blog post about AI', expected_output='Blog post with title, content, tags, and word count', agent=writer, output_pydantic=BlogPost )

Execute and get structured output

result = crew.kickoff() blog_post: BlogPost = write_task.output.pydantic print(blog_post.title) print(blog_post.tags)

Training: Improving Performance

from crewai import Crew

crew = Crew( agents=[agent1, agent2], tasks=[task1, task2] )

Training loop

crew.train( n_iterations=10, inputs={'topic': 'AI'}, filename='trained_crew.pkl' )

Load trained crew

trained_crew = Crew.load('trained_crew.pkl')

Human-in-the-Loop

from crewai import Task

Task requiring human input

review_task = Task( description='Review the draft and provide feedback', expected_output='Approved draft or feedback for revision', agent=editor, human_input=True # Will pause and ask for input )

Conditional human input

task = Task( description='Generate report', expected_output='Final report', agent=analyst, callback=lambda output: validate_output(output), human_input=True if needs_review else False )

Testing Crews

from crewai import Crew import pytest

def test_research_crew(): # Setup crew = Crew( agents=[researcher], tasks=[research_task] )

# Execute
result = crew.kickoff(inputs={'topic': 'AI'})

# Assert
assert result is not None
assert 'AI' in result
assert len(result) > 100

def test_crew_with_mock(): # Mock agent behavior for testing mock_agent = Agent( role='Mock Agent', goal='Return test data', backstory='Test agent' )

mock_task = Task(
    description='Test task',
    expected_output='Test output',
    agent=mock_agent
)

crew = Crew(agents=[mock_agent], tasks=[mock_task])
result = crew.kickoff()

assert result == 'Test output'

Custom LLMs

from langchain_openai import ChatOpenAI from crewai import Agent, Crew

Using custom LLM

custom_llm = ChatOpenAI( model='gpt-4-turbo-preview', temperature=0.7, max_tokens=2000 )

agent = Agent( role='Writer', llm=custom_llm )

Crew-level LLM

crew = Crew( agents=[agent1, agent2], tasks=[task1, task2], manager_llm=custom_llm # For hierarchical process )

Async Execution

from crewai import Crew

crew = Crew( agents=[agent1, agent2], tasks=[task1, task2] )

Async kickoff

async def run_crew(): result = await crew.kickoff_async(inputs={'topic': 'AI'}) return result

Kickoff for each (parallel execution)

inputs_list = [ {'topic': 'AI'}, {'topic': 'ML'}, {'topic': 'Data Science'} ]

results = crew.kickoff_for_each(inputs=inputs_list)

Callbacks and Event Listeners

from crewai import Task, Agent

def on_task_complete(output): print(f"Task completed with output: {output}") # Log, notify, or process output

def on_task_error(error): print(f"Task failed with error: {error}") # Handle error, retry, or notify

task = Task( description='Analyze data', expected_output='Analysis report', agent=analyst, callback=on_task_complete )

Agent-level callbacks

agent = Agent( role='Analyst', step_callback=lambda step: print(f"Agent step: {step}"), task_callback=on_task_complete )

Enterprise Deployment

Environment Configuration

import os

API keys

os.environ['OPENAI_API_KEY'] = 'your-key' os.environ['SERPER_API_KEY'] = 'your-key'

CrewAI+ (Enterprise)

os.environ['CREWAI_API_KEY'] = 'your-enterprise-key'

Observability

os.environ['LANGCHAIN_TRACING_V2'] = 'true' os.environ['LANGCHAIN_API_KEY'] = 'your-langchain-key'

Project Structure

my_crew_project/ ├── src/ │ └── my_crew_project/ │ ├── init.py │ ├── main.py │ ├── crew.py │ ├── config/ │ │ ├── agents.yaml │ │ └── tasks.yaml │ └── tools/ │ └── custom_tool.py ├── tests/ │ └── test_crew.py ├── pyproject.toml └── README.md

YAML Configuration

agents.yaml

researcher: role: > Senior Research Analyst goal: > Uncover cutting-edge developments in {topic} backstory: > You are an expert researcher with deep knowledge in {topic}

writer: role: > Content Writer goal: > Create engaging content about {topic} backstory: > You are a skilled writer who makes complex topics accessible

tasks.yaml

research_task: description: > Conduct comprehensive research on {topic} expected_output: > A detailed research report with key findings agent: researcher

writing_task: description: > Write an article based on the research expected_output: > A well-structured article agent: writer context: - research_task

Best Practices

Agent Design

✅ Good Practices:

• Give agents clear, specific roles

• Provide detailed backstories for context

• Limit tools to what's necessary

• Enable delegation for managers

• Use verbose mode during development

❌ Avoid:

• Vague or overlapping roles

• Too many tools (causes confusion)

• Missing backstories

• Overly complex goals

Task Design

✅ Good Practices:

• Write clear, actionable descriptions

• Specify expected output format

• Set up proper task dependencies

• Use context for task chaining

• Enable human input for critical decisions

❌ Avoid:

• Ambiguous descriptions

• Missing expected output

• Circular dependencies

• Overly complex single tasks

Crew Organization

✅ Good Practices:

• Start with sequential process

• Use hierarchical for complex coordination

• Enable memory for context retention

• Set reasonable rate limits

• Test with small datasets first

❌ Avoid:

• Too many agents (3-5 is optimal)

• Complex hierarchies without testing

• Disabled memory in multi-step flows

• No rate limiting

Common Patterns

Research and Write Pipeline

1. Research agent gathers information

2. Analyst agent processes data

3. Writer agent creates content

4. Editor agent reviews and refines

researcher = Agent(role='Researcher', ...) analyst = Agent(role='Analyst', ...) writer = Agent(role='Writer', ...) editor = Agent(role='Editor', ...)

research = Task(agent=researcher, ...) analysis = Task(agent=analyst, context=[research], ...) draft = Task(agent=writer, context=[analysis], ...) final = Task(agent=editor, context=[draft], ...)

crew = Crew( agents=[researcher, analyst, writer, editor], tasks=[research, analysis, draft, final], process=Process.sequential )

Multi-Stage Approval Flow

class ApprovalFlow(Flow):

@start()
def create_draft(self):
    # Generate initial draft
    return draft_content

@listen(create_draft)
def request_review(self, draft):
    # Send for review
    return review_request

@router(request_review)
def check_approval(self, review):
    if review.approved:
        return "finalize"
    else:
        return "revise"

@listen("revise")
def revise_draft(self):
    # Revise and loop back
    return revised_draft

@listen("finalize")
def finalize_content(self):
    return final_content

Quick Reference

Installation

Using uv (recommended)

uv pip install crewai crewai-tools

Using pip

pip install crewai crewai-tools

With all extras

pip install 'crewai[all]'

CLI Commands

Create new project

crewai create crew my_project

Create flow

crewai create flow my_flow

Install dependencies

crewai install

Run project

crewai run

Train crew

crewai train

Replay task

crewai replay <task_id>

Test crew

crewai test

Essential Imports

from crewai import Agent, Task, Crew, Process from crewai.flow.flow import Flow, listen, start, router from crewai_tools import SerperDevTool, ScrapeWebsiteTool from pydantic import BaseModel

Resources

For advanced patterns, integration examples, and troubleshooting:

• Official Documentation: https://docs.crewai.com/

• API Reference: https://docs.crewai.com/api-reference

• GitHub: https://github.com/joaomdmoura/crewAI

• Community Forum: https://community.crewai.com/

Extended Reference

See references/advanced_patterns.md for:

• MCP (Model Context Protocol) integration

• Observability and tracing setup

• Production deployment strategies

• Advanced flow patterns

• Performance optimization