Building CrewAI Agents

CrewAI is a framework for orchestrating multi-agent AI systems where specialized agents collaborate to complete complex tasks. AgentVisor™ supports CrewAI natively — your crews run inside a hardened sandbox with policy enforcement, credential brokering, and durable execution.

This guide covers building CrewAI agents for AgentVisor.

Prerequisites

• Python 3.10+
• CrewAI installed (pip install crewai)
• AgentVisor CLI installed

Basic Structure

Every CrewAI project for AgentVisor needs:

1. crewai.yaml: Discovery file pointing AgentVisor to your crew class(es)
2. Crew module: Python file defining agents, tasks, and the crew
3. requirements.txt: Python dependencies

crewai.yaml

The discovery file tells AgentVisor how to find your crews:

crews:
  agent: "./crew.py:MyCrew"
dependencies:
  - "."

The key under crews: is the crew name (used in API calls). The value is a module:ClassName path — either a file path or importable module.

Multiple crews:

crews:
  research: "./research_crew.py:ResearchCrew"
  summary: "./summary_crew.py:SummaryCrew"
dependencies:
  - "."

Minimal Crew Example

from crewai import LLM, Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task

@CrewBase
class MyCrew:
    """A simple single-agent crew."""

    @agent
    def analyst(self) -> Agent:
        return Agent(
            role="Analyst",
            goal="Analyze the provided topic and give insights",
            backstory="You are an expert analyst with broad domain knowledge.",
            llm=LLM(model="anthropic/claude-haiku-4-5-20251001"),
            verbose=True,
        )

    @task
    def analyze_task(self) -> Task:
        return Task(
            description="Analyze the following topic: {topic}",
            expected_output="A concise analysis with key insights and conclusions.",
            agent=self.analyst(),
        )

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True,
        )

Input / Output

CrewAI agents in AgentVisor receive input as a JSON object passed to crew.kickoff(inputs=...).

The {placeholder} syntax in task descriptions maps directly to input fields:

# Task description with placeholders
description="Research the following topic: {topic}\n\nFocus area: {focus}"

Calling the agent:

THREAD=$(curl -sX POST http://localhost:8090/threads | jq -r '.thread_id')

curl -sX POST "http://localhost:8090/threads/$THREAD/runs?wait=120s" \
    -H "Content-Type: application/json" \
    -d '{"input": {"topic": "quantum computing", "focus": "recent hardware advances"}}'

The output object contains:

{
  "output": "...",           // Final crew output (raw text)
  "tasks_output": [...],     // Per-task output details
  "token_usage": {...}       // Token consumption metrics
}

Multi-Agent Crews

The power of CrewAI is composing multiple specialized agents:

import os
from crewai import LLM, Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task

def _get_llm() -> LLM:
    model = os.environ.get("ANTHROPIC_MODEL", "claude-haiku-4-5-20251001")
    if not model.startswith("anthropic/"):
        model = f"anthropic/{model}"
    return LLM(model=model)


@CrewBase
class ResearchCrew:
    """Multi-agent crew: researcher gathers data, writer creates report."""

    @agent
    def researcher(self) -> Agent:
        return Agent(
            role="Senior Researcher",
            goal="Conduct thorough research on the given topic",
            backstory=(
                "You are an expert researcher with years of experience gathering "
                "and synthesizing information from diverse sources."
            ),
            llm=_get_llm(),
            verbose=True,
        )

    @agent
    def writer(self) -> Agent:
        return Agent(
            role="Technical Writer",
            goal="Create clear, well-structured reports from research findings",
            backstory=(
                "You are a skilled technical writer who transforms complex research "
                "into accessible, well-organized reports."
            ),
            llm=_get_llm(),
            verbose=True,
        )

    @task
    def research_task(self) -> Task:
        return Task(
            description=(
                "Research the following topic thoroughly: {topic}\n\n"
                "Gather key facts, background context, recent developments, "
                "and authoritative references."
            ),
            expected_output=(
                "Comprehensive research notes with key facts, statistics, "
                "and source references."
            ),
            agent=self.researcher(),
        )

    @task
    def write_task(self) -> Task:
        return Task(
            description=(
                "Write a comprehensive report on: {topic}\n\n"
                "Based on the research, create a structured markdown report "
                "with an executive summary, key findings, and conclusions."
            ),
            expected_output=(
                "A well-structured markdown report with executive summary, "
                "key findings sections, and conclusions."
            ),
            agent=self.writer(),
        )

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True,
        )

Using MCP Tools with CrewAI

AgentVisor's MCP Gateway provides tools to CrewAI agents via the AgentVisor Python SDK. Use get_mcp_tools() to convert MCP server tools into CrewAI-compatible tools that CrewAI agents can call:

from agentvisor.crewai.mcp import get_mcp_tools

@agent
def researcher(self) -> Agent:
    # Load tools from a configured MCP server
    tools = get_mcp_tools("fetch")    # name matches mcp.servers[].name in agentvisor.yaml
    return Agent(
        role="Senior Researcher",
        goal="Research topics using web tools",
        backstory="Expert researcher with web access.",
        tools=tools,
        llm=_llm,
        verbose=True,
    )

Configure the MCP server in agentvisor.yaml:

mcp:
  servers:
    - name: fetch
      transport: stdio
      command: ["uvx", "mcp-server-fetch"]

The policy engine controls which MCP tools agents can call. See Policy Configuration.

Using A2A Tools with CrewAI

Agents can call external A2A agents via get_a2a_tools():

from agentvisor.langchain.a2a import get_a2a_tools

@agent
def coordinator(self) -> Agent:
    a2a_tools = get_a2a_tools("analyzer")    # name matches a2a_gateway.agents[].name
    return Agent(
        role="Coordinator",
        goal="Coordinate research using external agents",
        backstory="Expert coordinator who delegates specialized work.",
        tools=a2a_tools,
        llm=_llm,
        verbose=True,
    )

Project Configuration

mav-agent-config.yaml

Optional agent-level configuration:

framework:
  provider: "crewai"    # Explicit provider (optional if crewai.yaml is present)
  env:
    ANTHROPIC_MODEL: "claude-haiku-4-5-20251001"

agentvisor.yaml

authz:
  type: embedded
  embedded:
    policy_domain_files:
      - "./policies/domain.yml"

proxy:
  credentials:
    - name: anthropic-key
      guest_env_var: ANTHROPIC_API_KEY
      destinations:
        - "api\\.anthropic\\.com"
      resolver:
        type: bearer_token
        source: env
        env_var: ANTHROPIC_API_KEY

mcp:
  servers:
    - name: fetch
      transport: stdio
      command: ["uvx", "mcp-server-fetch"]

requirements.txt

crewai[anthropic]>=1.0.0,<1.10.0

Provider Selection

AgentVisor automatically selects the CrewAI provider when crewai.yaml is present — as long as no higher-priority discovery file also matches. The guest runtime checks, in order: an explicit framework.provider, then langgraph.json, then agent.yaml, then crewai.yaml, then (in agentvisor exec mode only) the interactive provider, then errors:

framework section	langgraph.json / agent.yaml present	crewai.yaml exists	Result
Absent	No	Yes	crewai provider (implicit)
Absent	Yes	Any	langgraph or adk provider wins — checked before crewai.yaml
Absent	No	No	Error: no provider found (unless agentvisor exec falls back to interactive)
provider: crewai	Any	Any	CrewAI provider (explicit, always wins)

A project should generally only ship one of langgraph.json, agent.yaml, or crewai.yaml to avoid relying on this ordering.

Schema Discovery

AgentVisor uses a three-tier fallback to discover crew schemas:

1. Build-time cache (.agentvisor/schema.json) — used if newer than crewai.yaml
2. Dynamic extraction (python3 -m agentvisor.schema_cli) — introspects crew classes
3. Fallback — crew names from crewai.yaml only (no input/output schema)

Input schemas are derived from {placeholder} tokens in task descriptions.

Testing Locally

# Start development services
temporal server start-dev

# Run the agent (no sandbox for local dev)
agentvisor serve ./my-crewai-agent --sandbox=none

# Create a thread
THREAD=$(curl -sX POST http://localhost:8090/threads | jq -r '.thread_id')

# Invoke the crew
curl -sX POST "http://localhost:8090/threads/$THREAD/runs?wait=120s" \
    -H "Content-Type: application/json" \
    -d '{"input": {"topic": "the Rust programming language"}}' | jq

Best Practices

Keep Tasks Focused

Each task should have a clear, single responsibility:

# Good
@task
def research_task(self) -> Task:
    return Task(
        description="Research {topic}: gather facts, sources, and key data.",
        expected_output="Research notes with sources.",
        agent=self.researcher(),
    )

@task
def write_task(self) -> Task:
    return Task(
        description="Write a report on {topic} based on research findings.",
        expected_output="Structured markdown report.",
        agent=self.writer(),
    )

# Avoid combining research + writing in a single task

Define Clear Expected Output

Specific expected_output strings improve reliability:

return Task(
    description="Analyze {data} for anomalies.",
    expected_output=(
        "A JSON object with fields: 'anomalies' (list of findings), "
        "'severity' (high/medium/low), 'recommendations' (list of actions)."
    ),
    agent=self.analyst(),
)

Use Structured Input

Design task descriptions with consistent {placeholder} names that map directly to your API input:

# API input: {"topic": "...", "depth": "brief|detailed"}
description="Research {topic} at {depth} depth."

Related Guides

• MCP Gateway: Configuring MCP servers for agent tool access
• Policy Configuration: Authorization policies for crew tool calls
• Agent Logging: Structured logging for CrewAI agents