AI Testing

CrewAI Testing Guide: Test Crew Pipelines, Mock LLMs & CI/CD Integration

HelpMeTest

HelpMeTest

— 8 min read
CrewAI Testing Guide: Test Crew Pipelines, Mock LLMs & CI/CD Integration

CrewAI makes it straightforward to build multi-agent workflows where AI agents with different roles collaborate on complex tasks. But without a solid testing strategy, CrewAI pipelines become black boxes — hard to debug, expensive to iterate on, and fragile in CI. This guide gives you a complete testing approach for CrewAI applications.

Why CrewAI Testing Is Different

CrewAI introduces several testing challenges that don't exist in simple LLM applications:

  • Role-based behavior — each agent has a defined role, goal, and backstory that shapes its outputs
  • Sequential and parallel task execution — tasks may run in order or in parallel, with outputs flowing between them
  • Tool use — agents call tools, and tool results affect subsequent agent behavior
  • Process types — sequential and hierarchical processes have different execution semantics
  • Non-deterministic LLM outputs — running the same crew twice may produce different results

The goal of CrewAI testing is not to test whether Claude or GPT-4 is smart enough to complete a task. It's to test that your crew's structure, task definitions, agent roles, and tool integrations work correctly — independent of LLM intelligence.

Setting Up the Testing Environment

pip install crewai crewai-tools pytest pytest-asyncio langchain-openai
# For test utilities
pip install pytest-mock

Project Structure

my_crew/
├── src/
│   ├── crew.py           # Crew definition
│   ├── agents.py         # Agent definitions
│   ├── tasks.py          # Task definitions
│   └── tools/
│       ├── search.py
│       └── database.py
├── tests/
│   ├── unit/
│   │   ├── test_agents.py
│   │   ├── test_tasks.py
│   │   └── test_tools.py
│   ├── integration/
│   │   └── test_crew_pipeline.py
│   └── conftest.py

Unit Testing CrewAI Agents

CrewAI agents are defined by their role, goal, backstory, and tools. Unit tests verify these properties are set correctly and that agents interact with their tools as expected.

Testing Agent Configuration

# src/agents.py
from crewai import Agent
from src.tools.search import web_search_tool
from src.tools.database import db_query_tool

def create_research_agent(llm=None):
    return Agent(
        role="Senior Research Analyst",
        goal="Uncover comprehensive information on any topic and synthesize findings clearly",
        backstory="""You are a meticulous researcher with 10 years of experience 
        in competitive intelligence and market analysis.""",
        tools=[web_search_tool],
        llm=llm,
        verbose=False,
        allow_delegation=False,
        max_iter=3,
        max_rpm=10
    )

def create_writer_agent(llm=None):
    return Agent(
        role="Content Writer",
        goal="Transform research findings into clear, engaging written content",
        backstory="You are an expert writer who turns complex research into accessible articles.",
        tools=[],
        llm=llm,
        verbose=False,
        allow_delegation=False
    )
# tests/unit/test_agents.py
import pytest
from src.agents import create_research_agent, create_writer_agent

def test_research_agent_has_correct_role():
    agent = create_research_agent()
    assert agent.role == "Senior Research Analyst"

def test_research_agent_has_search_tool():
    agent = create_research_agent()
    tool_names = [tool.name for tool in agent.tools]
    assert "web_search" in tool_names

def test_research_agent_has_max_iterations_set():
    agent = create_research_agent()
    assert agent.max_iter == 3

def test_writer_agent_has_no_tools():
    agent = create_writer_agent()
    assert len(agent.tools) == 0

def test_writer_agent_delegation_disabled():
    agent = create_writer_agent()
    assert agent.allow_delegation == False

def test_agents_accept_custom_llm():
    from unittest.mock import MagicMock
    mock_llm = MagicMock()
    agent = create_research_agent(llm=mock_llm)
    assert agent.llm == mock_llm

Unit Testing CrewAI Tasks

Tasks define what each agent should do, what inputs they receive, and what output they should produce.

# src/tasks.py
from crewai import Task

def create_research_task(agent, topic: str):
    return Task(
        description=f"""Research the following topic comprehensively: {topic}
        
        Your research must include:
        1. Current state of the field
        2. Key players and recent developments  
        3. Future trends and predictions
        
        Use the web search tool to find current information.""",
        expected_output="""A detailed research report with:
        - Executive summary (2-3 sentences)
        - Key findings (bullet points)
        - Sources consulted""",
        agent=agent,
        output_file="research_output.md"
    )

def create_writing_task(agent, research_task):
    return Task(
        description="""Based on the research provided, write a compelling blog article.
        
        The article should be:
        - 800-1000 words
        - Written for a technical audience
        - Include concrete examples and data points
        - Have clear section headings""",
        expected_output="A complete blog article in markdown format",
        agent=agent,
        context=[research_task],  # Takes research task output as input
    )
# tests/unit/test_tasks.py
import pytest
from unittest.mock import MagicMock
from src.tasks import create_research_task, create_writing_task
from src.agents import create_research_agent, create_writer_agent

@pytest.fixture
def research_agent():
    return create_research_agent()

@pytest.fixture
def writer_agent():
    return create_writer_agent()

def test_research_task_includes_topic_in_description(research_agent):
    task = create_research_task(research_agent, "quantum computing")
    assert "quantum computing" in task.description

def test_research_task_has_expected_output_defined(research_agent):
    task = create_research_task(research_agent, "any topic")
    assert task.expected_output
    assert len(task.expected_output) > 10

def test_research_task_assigned_to_correct_agent(research_agent):
    task = create_research_task(research_agent, "topic")
    assert task.agent == research_agent

def test_writing_task_has_research_task_as_context(research_agent, writer_agent):
    research_task = create_research_task(research_agent, "topic")
    writing_task = create_writing_task(writer_agent, research_task)
    assert research_task in writing_task.context

def test_writing_task_assigned_to_writer_agent(research_agent, writer_agent):
    research_task = create_research_task(research_agent, "topic")
    writing_task = create_writing_task(writer_agent, research_task)
    assert writing_task.agent == writer_agent

Unit Testing CrewAI Tools

Tools are pure functions — test them independently from agents and crews.

# src/tools/search.py
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

class WebSearchInput(BaseModel):
    query: str = Field(..., description="The search query to execute")
    num_results: int = Field(default=5, description="Number of results to return")

class WebSearchTool(BaseTool):
    name: str = "web_search"
    description: str = "Search the web for current information on any topic"
    args_schema: Type[BaseModel] = WebSearchInput
    
    def _run(self, query: str, num_results: int = 5) -> str:
        # Real implementation would call a search API
        from src.services.search_api import search
        results = search(query, limit=num_results)
        return self._format_results(results)
    
    def _format_results(self, results: list) -> str:
        formatted = []
        for r in results:
            formatted.append(f"**{r['title']}**\n{r['snippet']}\nSource: {r['url']}")
        return "\n\n".join(formatted)

web_search_tool = WebSearchTool()
# tests/unit/test_tools.py
import pytest
from unittest.mock import patch, MagicMock
from src.tools.search import WebSearchTool

@pytest.fixture
def tool():
    return WebSearchTool()

def test_tool_has_correct_name(tool):
    assert tool.name == "web_search"

def test_tool_has_description(tool):
    assert tool.description
    assert len(tool.description) > 10

def test_tool_returns_formatted_results(tool):
    mock_results = [
        {"title": "Test Title", "snippet": "Test snippet", "url": "https://example.com"},
        {"title": "Another Result", "snippet": "More info", "url": "https://example2.com"}
    ]
    
    with patch('src.services.search_api.search', return_value=mock_results):
        result = tool._run("test query", num_results=2)
    
    assert "Test Title" in result
    assert "https://example.com" in result
    assert "Another Result" in result

def test_tool_respects_num_results_param(tool):
    with patch('src.services.search_api.search') as mock_search:
        mock_search.return_value = []
        tool._run("test query", num_results=3)
        mock_search.assert_called_once_with("test query", limit=3)

def test_tool_handles_empty_results(tool):
    with patch('src.services.search_api.search', return_value=[]):
        result = tool._run("obscure query")
        assert isinstance(result, str)
        # Should not raise an exception

Mocking LLM Responses in CrewAI

The key to fast, deterministic CrewAI tests is mocking LLM responses. CrewAI uses LangChain under the hood, so you can inject mock LLMs:

# tests/conftest.py
import pytest
from unittest.mock import MagicMock, AsyncMock
from langchain_core.messages import AIMessage

class MockLLM:
    """Deterministic mock LLM for CrewAI testing."""
    
    def __init__(self, responses: dict = None):
        # Map of keyword → response for routing different prompts
        self.responses = responses or {}
        self.default_response = "Task completed successfully with mock data."
        self.calls = []
    
    def invoke(self, messages):
        # Extract the last human message content
        if isinstance(messages, list):
            content = " ".join(
                str(m.content) if hasattr(m, 'content') else str(m)
                for m in messages
            )
        else:
            content = str(messages)
        
        self.calls.append(content)
        
        # Route to specific response based on keywords
        for keyword, response in self.responses.items():
            if keyword.lower() in content.lower():
                return AIMessage(content=response)
        
        return AIMessage(content=self.default_response)
    
    # CrewAI may also call these
    def predict(self, text: str) -> str:
        return self.default_response
    
    def __call__(self, messages):
        return self.invoke(messages)

@pytest.fixture
def mock_llm():
    return MockLLM(responses={
        "research": "## Research Findings\n\n- Key finding 1\n- Key finding 2\n\nSources: example.com",
        "write": "# Article Title\n\nThis is the article content based on the research.",
    })

Testing the Full Crew with Mocked LLM

# src/crew.py
from crewai import Crew, Process
from src.agents import create_research_agent, create_writer_agent
from src.tasks import create_research_task, create_writing_task

def build_content_crew(topic: str, llm=None):
    research_agent = create_research_agent(llm=llm)
    writer_agent = create_writer_agent(llm=llm)
    
    research_task = create_research_task(research_agent, topic)
    writing_task = create_writing_task(writer_agent, research_task)
    
    crew = Crew(
        agents=[research_agent, writer_agent],
        tasks=[research_task, writing_task],
        process=Process.sequential,
        verbose=False
    )
    
    return crew
# tests/integration/test_crew_pipeline.py
import pytest
from unittest.mock import patch
from src.crew import build_content_crew

@pytest.fixture
def content_crew(mock_llm):
    with patch('src.tools.search.WebSearchTool._run') as mock_search:
        mock_search.return_value = "Search result: AI is transforming industries."
        crew = build_content_crew("artificial intelligence", llm=mock_llm)
        yield crew, mock_search

def test_crew_completes_without_error(content_crew, mock_llm):
    crew, _ = content_crew
    result = crew.kickoff()
    assert result is not None

def test_crew_produces_non_empty_output(content_crew, mock_llm):
    crew, _ = content_crew
    result = crew.kickoff()
    assert str(result).strip()

def test_crew_uses_research_tool(content_crew):
    crew, mock_search = content_crew
    crew.kickoff()
    # Research agent should have called the search tool
    mock_search.assert_called()

def test_crew_sequential_task_order(mock_llm):
    """Verify tasks execute in the correct order."""
    execution_order = []
    
    original_run = MockLLM.invoke
    def tracked_invoke(self, messages):
        content = str(messages)
        if "research" in content.lower():
            execution_order.append("research")
        elif "write" in content.lower() or "article" in content.lower():
            execution_order.append("write")
        return original_run(self, messages)
    
    mock_llm.invoke = lambda messages: tracked_invoke(mock_llm, messages)
    
    with patch('src.tools.search.WebSearchTool._run', return_value="search results"):
        crew = build_content_crew("topic", llm=mock_llm)
        crew.kickoff()
    
    # Research should come before writing
    if "research" in execution_order and "write" in execution_order:
        assert execution_order.index("research") < execution_order.index("write")

Testing Crew with Inputs (Dynamic Crews)

CrewAI supports passing inputs at kickoff time. Test that your crew correctly uses them:

# tests/integration/test_crew_inputs.py
import pytest
from src.crew import build_content_crew

def test_crew_uses_topic_input(mock_llm):
    """Verify the crew topic is incorporated into task descriptions."""
    with patch('src.tools.search.WebSearchTool._run') as mock_search:
        mock_search.return_value = "Results about quantum computing"
        
        crew = build_content_crew("quantum computing", llm=mock_llm)
        
        # Check the topic made it into task descriptions
        task_descriptions = [task.description for task in crew.tasks]
        assert any("quantum computing" in desc for desc in task_descriptions)

def test_different_topics_produce_different_task_descriptions(mock_llm):
    crew1 = build_content_crew("topic one", llm=mock_llm)
    crew2 = build_content_crew("topic two", llm=mock_llm)
    
    desc1 = crew1.tasks[0].description
    desc2 = crew2.tasks[0].description
    
    assert desc1 != desc2
    assert "topic one" in desc1
    assert "topic two" in desc2

CI/CD Integration for CrewAI

# .github/workflows/test-crewai.yml
name: Test CrewAI Pipelines

on: [push, pull_request]

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      
      - name: Install dependencies
        run: pip install -r requirements.txt
      
      - name: Run agent unit tests
        run: pytest tests/unit/test_agents.py -v
      
      - name: Run task unit tests
        run: pytest tests/unit/test_tasks.py -v
      
      - name: Run tool unit tests
        run: pytest tests/unit/test_tools.py -v

  integration-tests:
    runs-on: ubuntu-latest
    needs: unit-tests
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      
      - name: Install dependencies
        run: pip install -r requirements.txt
      
      - name: Run crew integration tests
        # No API keys needed — all LLM calls are mocked
        run: pytest tests/integration/ -v --timeout=30

  smoke-tests:
    runs-on: ubuntu-latest
    needs: integration-tests
    if: github.ref == 'refs/heads/main'
    steps:
      - uses: actions/checkout@v4
      
      - name: Run real LLM smoke test (once per main branch push)
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: pytest tests/smoke/ -v --timeout=120 -k "smoke"

Common Mistakes in CrewAI Testing

1. Testing with real LLM calls in every test. This is slow, expensive, and non-deterministic. Mock LLM calls in unit and integration tests. Reserve real LLM calls for smoke tests and evaluations.

2. Ignoring tool mocking. If your agents use tools that call external APIs, those calls must be mocked in CI. Otherwise your tests fail whenever the API is down or rate-limited.

3. Not testing agent/task configuration. Misconfigured agents (wrong tools, missing backstory, uncapped iterations) are a real source of bugs. Simple configuration tests catch these.

4. Testing crew output semantics with unit tests. Unit tests can verify that crew output is non-empty and structurally valid. They cannot meaningfully test whether the content is "good." That requires evals.

5. No timeout on integration tests. CrewAI crews can loop indefinitely if misconfigured. Always set --timeout in pytest for integration tests.

Production Monitoring for CrewAI Pipelines

Even with thorough testing, CrewAI pipelines in production can degrade when:

  • LLM providers update their models
  • Tool APIs change their response formats
  • Prompt injection attacks corrupt agent behavior
  • Input data distribution shifts

Setting up automated monitoring that periodically runs your critical crew workflows with known inputs — and validates that outputs meet quality thresholds — gives you a safety net. HelpMeTest can schedule these validation runs on any interval, alerting your team when outputs fall outside expected bounds before real users experience degraded results.

Summary

Testing CrewAI applications effectively means:

  1. Unit test agents — role, tools, limits, and configuration
  2. Unit test tasks — description content, context wiring, agent assignment
  3. Unit test tools — independently from agents, with mocked external calls
  4. Integration test crews — with mocked LLMs and mocked tool responses
  5. Smoke test with real LLMs — sparingly, on main branch only
  6. Never ship without CI — all three test layers must pass before deployment

The mock LLM pattern is the most important technique in this guide. Once you can run your entire crew deterministically with a MockLLM, your iteration speed increases dramatically and your CI pipelines become fast and reliable.

Read more

Start now free