CrewAI structures multi-agent systems around Crews — groups of role-specialized Agents that execute Tasks in collaboration. Each Agent has a role, goal, and backstory that shape its LLM behavior. Tasks define what needs to be done and which Agent does it. The Process controls execution order: sequential for pipelines, hierarchical for manager-subordinate workflows. Agents use tools to interact with APIs, databases, and the web. Claude Code generates CrewAI agent definitions, task workflows, custom tools, and the orchestration layer for production autonomous systems.
CLAUDE.md for CrewAI Projects
## CrewAI Stack
- Version: crewai >= 0.80, crewai-tools >= 0.15
- LLM: anthropic/claude-sonnet-4-6 or openai/gpt-4o per agent (different for different tasks)
- Process: Process.sequential (default), Process.hierarchical (use when agents need to delegate)
- Tools: @tool decorator or BaseTool subclass — keep tools focused and well-documented
- Memory: memory=True on Crew enables short-term + entity memory via ChromaDB
- Output: Use output_pydantic on Task for structured, validated results
- Async: kickoff_async() for non-blocking execution in FastAPI/async contextsBasic Crew Definition
# crews/research_crew.py — research and reporting crew
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool
from langchain_anthropic import ChatAnthropic
from langchain_openai import ChatOpenAI
import os
# Different LLMs for different agents
research_llm = ChatAnthropic(
model="claude-sonnet-4-6",
api_key=os.environ["ANTHROPIC_API_KEY"],
temperature=0.1,
)
writer_llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
# Tools
search_tool = SerperDevTool(n_results=5)
web_tool = WebsiteSearchTool()
# Agents — each has a role, goal, and backstory shaping LLM behavior
researcher = Agent(
role="Senior Research Analyst",
goal="Gather comprehensive, accurate information about {topic} from multiple sources",
backstory="""You are an expert researcher with 10 years of experience in technology
analysis. You excel at finding authoritative sources, cross-referencing information,
and identifying key trends. You never fabricate information.""",
tools=[search_tool, web_tool],
llm=research_llm,
verbose=True,
max_iter=5, # Max tool call iterations before giving up
memory=True,
)
analyst = Agent(
role="Data Analyst",
goal="Analyze research findings and extract actionable insights about {topic}",
backstory="""You are a sharp analytical mind who transforms raw research into
structured insights. You identify patterns, compare options, and provide
data-driven recommendations.""",
llm=research_llm,
verbose=True,
allow_delegation=False, # This agent executes, doesn't delegate
)
writer = Agent(
role="Technical Content Writer",
goal="Create clear, engaging, well-structured content about {topic} for developers",
backstory="""You are a technical writer who makes complex topics accessible.
You write concisely, use concrete examples, and structure content for
maximum developer comprehension.""",
llm=writer_llm,
verbose=True,
allow_delegation=False,
)
# Tasks — define what each agent must produce
research_task = Task(
description="""Research {topic} thoroughly:
1. Find the latest developments and key frameworks/tools
2. Identify best practices and common patterns
3. Note performance comparisons and trade-offs
4. Gather real-world use cases and adoption statistics
Cite all sources with URLs.""",
expected_output="Comprehensive research report with cited sources, covering current state, best practices, and trade-offs",
agent=researcher,
)
analysis_task = Task(
description="""Analyze the research provided and:
1. Summarize the 5 most important insights
2. Compare top 3 approaches/tools with pros/cons table
3. Identify which use case fits which approach
4. Provide a clear recommendation for typical developers""",
expected_output="Structured analysis with comparison table and clear recommendations",
agent=analyst,
context=[research_task], # This task receives research_task's output as context
)
writing_task = Task(
description="""Write a developer blog post about {topic} based on the research and analysis.
Requirements:
- Title with clear value proposition
- Introduction (2-3 sentences, what problem this solves)
- 3-4 main sections with code examples where relevant
- Comparison table from the analysis
- Conclusion with next steps
- Target: 800-1200 words, developer audience""",
expected_output="Complete blog post in Markdown format, ready to publish",
agent=writer,
context=[research_task, analysis_task],
)
def run_research_crew(topic: str) -> str:
"""Execute the research and writing crew."""
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, writing_task],
process=Process.sequential,
verbose=True,
memory=True, # Enable cross-task memory
max_rpm=20, # Rate limit API calls
)
result = crew.kickoff(inputs={"topic": topic})
return result.rawCustom Tools
# tools/custom_tools.py — build tools for CrewAI agents
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
import httpx
import json
class DatabaseQueryInput(BaseModel):
sql: str = Field(description="SQL query to execute against the analytics database")
limit: int = Field(default=100, description="Maximum rows to return")
class DatabaseQueryTool(BaseTool):
name: str = "Database Query Tool"
description: str = """Execute SQL queries against the analytics database.
Use for: product metrics, user statistics, sales data, performance KPIs.
Tables: users, orders, products, events, sessions.
Returns: JSON array of rows."""
args_schema: type[BaseModel] = DatabaseQueryInput
def _run(self, sql: str, limit: int = 100) -> str:
import sqlite3 # Replace with your DB driver
# Validate: only SELECT queries
if not sql.strip().upper().startswith("SELECT"):
return "Error: only SELECT queries are allowed"
# Add LIMIT if not present
if "LIMIT" not in sql.upper():
sql = f"{sql} LIMIT {limit}"
try:
conn = sqlite3.connect("analytics.db")
cursor = conn.execute(sql)
columns = [d[0] for d in cursor.description]
rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
conn.close()
return json.dumps(rows, default=str)
except Exception as e:
return f"Query error: {e}"
class APICallInput(BaseModel):
endpoint: str = Field(description="API endpoint path, e.g. /v1/customers")
params: dict = Field(default={}, description="Query parameters as key-value pairs")
class InternalAPITool(BaseTool):
name: str = "Internal API Tool"
description: str = """Call the internal business API.
Use for: customer data, order details, product catalog, inventory.
Supports GET requests only."""
args_schema: type[BaseModel] = APICallInput
base_url: str = "https://api.internal.company.com"
def _run(self, endpoint: str, params: dict = {}) -> str:
try:
response = httpx.get(
f"{self.base_url}{endpoint}",
params=params,
headers={"Authorization": f"Bearer {self._get_token()}"},
timeout=10,
)
response.raise_for_status()
return response.text
except httpx.HTTPError as e:
return f"API error: {e}"
def _get_token(self) -> str:
import os
return os.environ["INTERNAL_API_TOKEN"]
# Simple function-based tool
from crewai.tools import tool
@tool("Calculate Metrics")
def calculate_metrics(data: str) -> str:
"""Calculate key business metrics from a JSON data string.
Computes: mean, median, sum, count, min, max for numeric fields."""
import statistics
records = json.loads(data)
if not records:
return "No data provided"
numeric_fields = {
k for k in records[0].keys()
if isinstance(records[0][k], (int, float))
}
results = {}
for field in numeric_fields:
values = [r[field] for r in records if r.get(field) is not None]
results[field] = {
"count": len(values),
"sum": sum(values),
"mean": statistics.mean(values),
"median": statistics.median(values),
"min": min(values),
"max": max(values),
}
return json.dumps(results, default=str)Hierarchical Process with Manager
# crews/support_crew.py — hierarchical crew with manager agent
from crewai import Agent, Task, Crew, Process
from pydantic import BaseModel
class TicketResolution(BaseModel):
category: str
priority: str # low, medium, high, urgent
resolution: str
escalate_to_human: bool
follow_up_actions: list[str]
# Manager coordinates other agents
support_manager = Agent(
role="Customer Support Manager",
goal="Ensure customer issues are resolved efficiently by delegating to the right specialist",
backstory="""You manage a customer support team. You triage issues,
delegate to specialists, and ensure quality resolutions.
You escalate to humans when the issue is complex or emotional.""",
allow_delegation=True, # Manager CAN delegate to other agents
verbose=True,
)
billing_specialist = Agent(
role="Billing Specialist",
goal="Resolve billing issues, refund requests, and payment questions accurately",
backstory="You have deep knowledge of billing systems, refund policies, and payment processing.",
tools=[InternalAPITool(base_url="https://billing-api.internal.com")],
allow_delegation=False,
)
technical_specialist = Agent(
role="Technical Support Engineer",
goal="Diagnose and resolve technical issues with the product",
backstory="You understand the product deeply and can diagnose integration issues, bugs, and configuration problems.",
tools=[DatabaseQueryTool()],
allow_delegation=False,
)
resolve_ticket_task = Task(
description="""Resolve this customer support ticket:
Customer: {customer_name}
Issue: {ticket_description}
Investigate the issue, determine the root cause, and provide a resolution.
Determine if human escalation is needed.""",
expected_output="Complete ticket resolution with category, priority, resolution steps, and escalation decision",
agent=support_manager, # Manager decides who handles what
output_pydantic=TicketResolution,
)
def resolve_support_ticket(customer_name: str, ticket_description: str) -> TicketResolution:
"""Run hierarchical support crew to resolve a ticket."""
crew = Crew(
agents=[support_manager, billing_specialist, technical_specialist],
tasks=[resolve_ticket_task],
process=Process.hierarchical, # Manager delegates to specialists
manager_agent=support_manager,
verbose=True,
)
result = crew.kickoff(inputs={
"customer_name": customer_name,
"ticket_description": ticket_description,
})
return result.pydanticAsync Crew Execution in FastAPI
# api/crew_api.py — async CrewAI in a FastAPI service
from fastapi import FastAPI, BackgroundTasks
from crewai import Crew
import asyncio
import uuid
app = FastAPI()
active_runs: dict[str, dict] = {}
@app.post("/crew/research")
async def start_research(topic: str, background_tasks: BackgroundTasks):
"""Start an async research crew run."""
run_id = str(uuid.uuid4())
active_runs[run_id] = {"status": "running", "result": None}
async def run_crew():
crew = build_research_crew() # Returns Crew instance
result = await crew.kickoff_async(inputs={"topic": topic})
active_runs[run_id] = {"status": "completed", "result": result.raw}
background_tasks.add_task(run_crew)
return {"run_id": run_id, "status": "started"}
@app.get("/crew/research/{run_id}")
async def get_result(run_id: str):
"""Poll for crew run result."""
run = active_runs.get(run_id)
if not run:
return {"error": "Run not found"}
return runFor the LangChain/LangGraph alternative that uses explicit graph-based agent orchestration instead of CrewAI’s role-based model, see the LangChain guide for agent graphs and tool nodes. For the DSPy approach that compiles LLM programs through optimization rather than workflow orchestration, the DSPy guide covers declarative LLM pipelines. The Claude Skills 360 bundle includes CrewAI skill sets covering agent design, multi-crew workflows, and tool development. Start with the free tier to try CrewAI agent generation.