CrewAI Example

Build multi-agent AI crews with Composio and CrewAI using MCP

This example demonstrates how to use Composio with CrewAI to create collaborative AI agent teams that can execute complex tasks using MCP (Model Context Protocol).

Overview

In this example, you’ll learn how to:

• Integrate Composio with CrewAI using MCP
• Create agents with access to Composio tools
• Define tasks for agents to complete
• Execute crew workflows with tool access

Prerequisites

1. Install dependencies

   pip install composio crewai
   
2. Set up environment variables

   Create a .env file with your API keys:

   COMPOSIO_API_KEY=your_composio_api_key
   OPENAI_API_KEY=your_openai_api_key
   
3. Authenticate with Gmail

   composio add gmail
   

Complete Example

from crewai import Agent, Crew, Task
from crewai.mcp import MCPServerHTTP
from composio import Composio

# Initialize Composio and create a session
composio = Composio()
session = composio.create(
    user_id="user_123",
)

# Create an agent with Composio MCP server
agent = Agent(
    role="Gmail agent",
    goal="helps with gmail related queries",
    backstory="You are a helpful assistant that can use the tools provided to you.",
    mcps=[
        MCPServerHTTP(
            url=session.mcp.url,
            headers=session.mcp.headers,
        )
    ],
)

# Define task
task = Task(
    description=("Find the last email and summarize it."),
    expected_output="A summary of the last email including sender, subject, and key points.",
    agent=agent,
)

# Create and run the crew
my_crew = Crew(agents=[agent], tasks=[task])
result = my_crew.kickoff()
print(result)

How It Works

1. Create Composio Session

   Initialize Composio and create a session for your user. This provides an MCP server endpoint with all connected tools.

   composio = Composio()
   session = composio.create(user_id="user_123")
   
2. Configure MCP Server

   Create an MCPServerHTTP instance that connects to Composio’s MCP endpoint. This gives the agent access to all Composio tools.

   MCPServerHTTP(
       url=session.mcp.url,
       headers=session.mcp.headers,
   )
   
3. Create Agent with Tools

   Define an agent with a specific role, goal, and backstory. Add the MCP server to the mcps parameter to give it access to Composio tools.

4. Define Tasks

   Create tasks with clear descriptions and expected outputs. CrewAI will automatically select the appropriate tools to complete each task.

5. Execute Crew

   Create a Crew with your agents and tasks, then call kickoff() to start execution.

Multi-Agent Example

Build a crew with multiple specialized agents:

from crewai import Agent, Crew, Task, Process
from crewai.mcp import MCPServerHTTP
from composio import Composio

composio = Composio()
session = composio.create(user_id="user_123")

mcp_server = MCPServerHTTP(
    url=session.mcp.url,
    headers=session.mcp.headers,
)

# Email agent - handles email operations
email_agent = Agent(
    role="Email Specialist",
    goal="Manage and analyze emails efficiently",
    backstory="Expert in email management and communication",
    mcps=[mcp_server],
)

# Calendar agent - handles scheduling
calendar_agent = Agent(
    role="Calendar Manager",
    goal="Organize and optimize schedules",
    backstory="Expert in time management and scheduling",
    mcps=[mcp_server],
)

# Define tasks
email_task = Task(
    description="Find all unread emails from this week and create a summary",
    expected_output="A categorized summary of unread emails",
    agent=email_agent,
)

calendar_task = Task(
    description="Review this week's calendar and identify any scheduling conflicts",
    expected_output="A list of scheduling conflicts with suggested resolutions",
    agent=calendar_agent,
)

# Create crew with sequential process
crew = Crew(
    agents=[email_agent, calendar_agent],
    tasks=[email_task, calendar_task],
    process=Process.sequential,  # Tasks run in order
    verbose=True,
)

result = crew.kickoff()
print(result)

Expected Output

Starting crew execution...

[Gmail agent] Starting task: Find the last email and summarize it.
[Gmail agent] Using tool: gmail_get_latest_email
[Gmail agent] Task completed!

Result:
{
  "summary": {
    "from": "[e-mail]",
    "subject": "Q1 Project Update",
    "received": "2 hours ago",
    "key_points": [
      "Project is on track for Q1 delivery",
      "Budget review scheduled for next week",
      "Team needs additional resources for testing phase"
    ]
  }
}

Agent Configuration

role · string · required

The role or title of the agent (e.g., “Email Specialist”, “Data Analyst”)

goal · string · required

The agent’s objective and what it aims to accomplish

backstory · string · required

The agent’s background and expertise context

mcps · list[MCPServer]

List of MCP servers providing tools to the agent

verbose · bool · default: "false"

Whether to print detailed execution logs

Task Configuration

description · string · required

Detailed description of what the task should accomplish

expected_output · string · required

Clear specification of what the task output should look like

agent · Agent · required

The agent responsible for executing this task

Process Types

CrewAI supports different execution processes:

Sequential

# Tasks execute one after another
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    process=Process.sequential,
)

Crew Collaboration

Agents can collaborate on complex tasks:

research_task = Task(
    description="Research the latest AI trends in email automation",
    expected_output="A detailed report on AI email automation trends",
    agent=researcher_agent,
)

write_task = Task(
    description="Write an email summary based on the research",
    expected_output="A professional email summarizing the findings",
    agent=writer_agent,
    context=[research_task],  # Uses output from research_task
)

send_task = Task(
    description="Send the email to the team",
    expected_output="Confirmation that email was sent",
    agent=email_agent,
    context=[write_task],  # Uses output from write_task
)

crew = Crew(
    agents=[researcher_agent, writer_agent, email_agent],
    tasks=[research_task, write_task, send_task],
    process=Process.sequential,
)

Custom MCP Configuration

Configure MCP servers with advanced options:

mcp_server = MCPServerHTTP(
    url=session.mcp.url,
    headers=session.mcp.headers,
    timeout=30,  # Request timeout in seconds
    retry_attempts=3,  # Number of retry attempts
)

Error Handling

try:
    crew = Crew(
        agents=[agent],
        tasks=[task],
        verbose=True,
    )
    result = crew.kickoff()
    print(f"Success: {result}")
except Exception as e:
    print(f"Crew execution failed: {e}")
    # Handle errors (e.g., authentication, network issues)

Best Practices

Specific Roles: Give agents clear, specific roles to improve task execution quality

Clear Expected Outputs: Define precise expected outputs to guide agent behavior

Task Dependencies: Use context parameter to create task dependencies when needed

Verbose Mode: Enable verbose mode during development to understand agent decisions

Advanced Features

Memory and Learning

CrewAI agents can maintain memory across executions:

agent = Agent(
    role="Email Specialist",
    goal="Manage emails",
    backstory="Expert email manager",
    mcps=[mcp_server],
    memory=True,  # Enable memory
)

Custom Tools

Combine Composio tools with custom CrewAI tools:

from crewai_tools import tool

@tool
def custom_analysis(text: str) -> str:
    """Perform custom text analysis"""
    return f"Analyzed: {text}"

agent = Agent(
    role="Analyst",
    goal="Analyze data",
    backstory="Data expert",
    mcps=[mcp_server],
    tools=[custom_analysis],
)

Callbacks

Monitor crew execution with callbacks:

def step_callback(output):
    print(f"Step completed: {output}")

crew = Crew(
    agents=[agent],
    tasks=[task],
    step_callback=step_callback,
)

Next Steps

Custom Tools

Create custom Python tools for CrewAI

LangChain Example

Use Composio with LangChain