CrewAI — Build Collaborative Multi-Agent Systems in Python (2026)

pip install crewai crewai-tools

import os
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileReadTool, WebsiteSearchTool
 
os.environ["OPENAI_API_KEY"] = "your-openai-key"
os.environ["SERPER_API_KEY"] = "your-serper-key"  # For web search tool

from crewai import Agent
from crewai_tools import SerperDevTool
 
search_tool = SerperDevTool()
 
# Research agent: finds information
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find accurate, up-to-date information on the assigned topic and identify key trends",
    backstory="""You are a former academic researcher turned industry analyst.
You have 15 years of experience synthesizing complex information from multiple sources.
You always cite your sources and flag when information is uncertain or contested.
You are skeptical of marketing claims and prioritize peer-reviewed or primary sources.""",
    tools=[search_tool],
    verbose=True,
    allow_delegation=False,    # This agent does not delegate to others
    max_iter=5,                # Maximum tool-use iterations per task
    llm="gpt-4o"
)
 
# Writing agent: produces polished content
writer = Agent(
    role="Technical Content Writer",
    goal="Transform research findings into clear, engaging, and technically accurate content",
    backstory="""You are a technical writer with a computer science background.
You write for developers and engineers who value accuracy over marketing fluff.
You use concrete examples, avoid jargon without explanation, and structure content for scannability.
You never invent facts — if the research does not cover something, you say so.""",
    verbose=True,
    allow_delegation=False,
    llm="gpt-4o"
)
 
# Editor agent: quality control
editor = Agent(
    role="Editorial Director",
    goal="Ensure content is factually accurate, well-structured, and meets publication standards",
    backstory="""You have edited technical content for major developer publications for 10 years.
You check for factual accuracy, logical flow, consistent terminology, and readability.
You provide specific, actionable feedback. You approve content only when it meets high standards.""",
    verbose=True,
    allow_delegation=True,    # Can send work back to writer
    llm="gpt-4o"
)

from crewai import Task
 
def create_content_pipeline(topic: str) -> tuple:
    """Create a three-stage content pipeline for a given topic."""
 
    research_task = Task(
        description=f"""Research the following topic thoroughly: {topic}
 
Your deliverable must include:
1. A factual overview (what it is, why it matters)
2. Current state in 2026 (recent developments, adoption)
3. Key technical concepts (3-5 core ideas with brief explanations)
4. Real-world use cases (at least 3 concrete examples)
5. Limitations and challenges
6. Comparison with alternatives if relevant
 
Cite specific sources where possible. Flag any claims you are uncertain about.""",
        expected_output="A structured research report with sections, facts, sources, and flagged uncertainties",
        agent=researcher,
    )
 
    writing_task = Task(
        description=f"""Using the research provided, write a comprehensive technical blog post about: {topic}
 
Requirements:
- Target audience: senior software engineers
- Length: 1000-1200 words
- Structure: Introduction, 4-5 main sections with h2 headings, Conclusion
- Include at least one Python code example that is complete and runnable
- Use concrete numbers and benchmarks from the research
- Avoid fluff sentences — every sentence must add information""",
        expected_output="A complete, publication-ready technical blog post in markdown",
        agent=writer,
        context=[research_task],        # This task uses research_task's output
    )
 
    editing_task = Task(
        description="""Review the drafted blog post for:
1. Factual accuracy against the research report
2. Technical correctness of any code examples
3. Logical flow and structure
4. Readability (flag sentences over 25 words)
5. Missing context or unexplained jargon
 
Provide: (a) specific line-level feedback, and (b) a revised final version if changes are needed.
Approve with 'APPROVED' only when the post meets publication standards.""",
        expected_output="Editorial feedback and a final approved version of the blog post",
        agent=editor,
        context=[research_task, writing_task],
    )
 
    return research_task, writing_task, editing_task

from crewai import Crew, Process
 
def run_content_crew(topic: str) -> str:
    """Run the full content pipeline for a given topic."""
    research_task, writing_task, editing_task = create_content_pipeline(topic)
 
    crew = Crew(
        agents=[researcher, writer, editor],
        tasks=[research_task, writing_task, editing_task],
        process=Process.sequential,    # Tasks run in order
        verbose=True,
        memory=True,                   # Agents share short-term memory
        max_rpm=20,                    # Rate limit API calls
    )
 
    result = crew.kickoff()
    return str(result)
 
output = run_content_crew("vector databases and their role in RAG systems")
print(output)

from crewai import Agent, Crew, Process
 
manager = Agent(
    role="Engineering Manager",
    goal="Coordinate the team to deliver high-quality software solutions efficiently",
    backstory="You are a senior engineering manager who excels at breaking down problems and delegating effectively.",
    allow_delegation=True,
    llm="gpt-4o"
)
 
backend_dev = Agent(
    role="Backend Engineer",
    goal="Write efficient, secure Python backend code",
    backstory="Expert in FastAPI, PostgreSQL, and distributed systems.",
    allow_delegation=False,
    llm="gpt-4o"
)
 
qa_engineer = Agent(
    role="QA Engineer",
    goal="Write comprehensive test suites that catch bugs and regressions",
    backstory="Expert in pytest, test coverage analysis, and property-based testing.",
    allow_delegation=False,
    llm="gpt-4o"
)
 
build_task = Task(
    description="Build a rate-limiting middleware for a FastAPI application. Limit: 100 requests per minute per IP.",
    expected_output="Production-ready Python code with tests",
    agent=manager,  # Manager receives the task and delegates
)
 
crew = Crew(
    agents=[manager, backend_dev, qa_engineer],
    tasks=[build_task],
    process=Process.hierarchical,  # Manager delegates to workers
    manager_llm="gpt-4o",
    verbose=True,
)
 
result = crew.kickoff()

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
import httpx
 
class GitHubSearchInput(BaseModel):
    query: str = Field(description="GitHub search query")
    language: str = Field(default="python", description="Programming language filter")
 
class GitHubSearchTool(BaseTool):
    name: str = "GitHub Code Search"
    description: str = "Search GitHub for code examples and repositories"
    args_schema: type[BaseModel] = GitHubSearchInput
 
    def _run(self, query: str, language: str = "python") -> str:
        """Search GitHub and return formatted results."""
        # Replace with real GitHub API call using your token
        url = f"https://api.github.com/search/repositories?q={query}+language:{language}&sort=stars"
        headers = {"Authorization": "token YOUR_GITHUB_TOKEN"}
        try:
            response = httpx.get(url, headers=headers, timeout=10)
            data = response.json()
            items = data.get("items", [])[:5]
            results = [
                f"- {item['full_name']}: {item['description']} ({item['stargazers_count']} stars)"
                for item in items
            ]
            return "\n".join(results) if results else "No results found"
        except Exception as e:
            return f"Search failed: {e}"
 
# Attach to an agent
researcher_with_github = Agent(
    role="Open Source Researcher",
    goal="Find the best open-source solutions on GitHub",
    backstory="Expert at discovering and evaluating open-source projects.",
    tools=[GitHubSearchTool()],
    llm="gpt-4o"
)