CrewAI SDK Track Part 8: MCP Integration

1. MCP in CrewAI

The Model Context Protocol (MCP) is an open standard that allows AI agents to connect to external tools and data sources through a unified interface. CrewAI’s MCP integration lets your agents access any MCP-compatible server — from file systems and databases to custom enterprise services — without writing tool-specific code.

1.1 Architecture Overview

The flow is: Agent → MCPServerAdapter → MCP Server → External Service. CrewAI’s adapter translates MCP tool definitions into native CrewAI tools automatically.

from crewai import Agent, Task, Crew
from crewai.tools import MCPServerAdapter

# Basic MCP connection — connects to a local file system server
mcp_server = MCPServerAdapter(
    server_params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", "./workspace"]
    }
)

# Agent automatically discovers all tools from the MCP server
file_agent = Agent(
    role="File Manager",
    goal="Manage and organize workspace files efficiently",
    backstory="Expert file system operator",
    tools=mcp_server.tools,  # Auto-discovered tools from MCP
    verbose=True
)

task = Task(
    description="List all Python files in the workspace and summarize their purposes",
    expected_output="A list of .py files with brief descriptions",
    agent=file_agent
)

crew = Crew(agents=[file_agent], tasks=[task])
result = crew.kickoff()
print(result.raw)

1.2 When to Use MCP vs Native Tools

2. DSL Integration (Recommended)

The simplest way to connect MCP servers is via CrewAI’s DSL — just add an mcps field to your agent definition. This handles connection lifecycle, tool discovery, and cleanup automatically.

2.1 Agent with MCP DSL

from crewai import Agent, Task, Crew

# DSL approach — simplest MCP integration
researcher = Agent(
    role="Research Analyst",
    goal="Gather information from multiple sources",
    backstory="Expert researcher with access to various data sources",
    mcps=[
        {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "./data"]
        },
        {
            "url": "http://localhost:3001/sse",
            "transport": "sse"
        }
    ],
    verbose=True
)

task = Task(
    description="Read the research brief from ./data/brief.txt and gather supplementary information",
    expected_output="Comprehensive research report combining file data and web sources",
    agent=researcher
)

crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()
print(result.raw)

3. Transport Mechanisms

MCP supports three transport protocols. Choose based on your server deployment model:

3.1 Stdio Transport (Local Processes)

Best for local MCP servers that run as child processes. Communication happens via stdin/stdout:

from crewai import Agent, Task, Crew
from crewai.tools import MCPServerAdapter

# Stdio: spawn a local process (e.g., filesystem server)
filesystem_mcp = MCPServerAdapter(
    server_params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", "/home/user/projects"]
    }
)

# Stdio: Python-based MCP server
python_mcp = MCPServerAdapter(
    server_params={
        "command": "python",
        "args": ["-m", "my_mcp_server"],
        "env": {"DATABASE_URL": "postgresql://localhost/mydb"}
    }
)

agent = Agent(
    role="Developer Assistant",
    goal="Help manage project files and database operations",
    backstory="Full-stack developer assistant",
    tools=filesystem_mcp.tools + python_mcp.tools,
    verbose=True
)

task = Task(
    description="List the project structure and check database connectivity",
    expected_output="Project file tree and database status",
    agent=agent
)

crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()
print(result.raw)

3.2 SSE Transport (Server-Sent Events)

For MCP servers running as persistent HTTP services with real-time streaming:

from crewai import Agent, Task, Crew
from crewai.tools import MCPServerAdapter

# SSE: connect to a running MCP server via Server-Sent Events
sse_mcp = MCPServerAdapter(
    server_params={
        "url": "http://localhost:8080/sse",
        "transport": "sse"
    }
)

# SSE with authentication headers
auth_sse_mcp = MCPServerAdapter(
    server_params={
        "url": "https://mcp.example.com/sse",
        "transport": "sse",
        "headers": {
            "Authorization": "Bearer your-api-token"
        }
    }
)

agent = Agent(
    role="Data Analyst",
    goal="Query real-time data streams for analysis",
    backstory="Real-time data specialist",
    tools=sse_mcp.tools,
    verbose=True
)

task = Task(
    description="Fetch the latest metrics from the streaming data source",
    expected_output="Current metrics summary with trends",
    agent=agent
)

crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()
print(result.raw)

3.3 Streamable HTTP Transport

The most flexible transport — supports both request/response and streaming over standard HTTP:

from crewai import Agent, Task, Crew
from crewai.tools import MCPServerAdapter

# Streamable HTTP: modern flexible transport
http_mcp = MCPServerAdapter(
    server_params={
        "url": "https://mcp-api.example.com/mcp",
        "transport": "streamable-http",
        "headers": {
            "Authorization": "Bearer your-api-token",
            "X-Client-ID": "crewai-production"
        }
    }
)

agent = Agent(
    role="Enterprise Agent",
    goal="Interface with enterprise systems via MCP",
    backstory="Enterprise integration specialist",
    tools=http_mcp.tools,
    verbose=True
)

task = Task(
    description="Query the enterprise CRM for top accounts and their recent activities",
    expected_output="Top 5 accounts with recent interaction summary",
    agent=agent
)

crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()
print(result.raw)

4. Connecting to Multiple Servers

Real-world agents often need tools from multiple MCP servers. CrewAI supports aggregating tools from several servers into a single agent.

4.1 Multi-Server Agent

from crewai import Agent, Task, Crew
from crewai.tools import MCPServerAdapter

# Server 1: File system access
fs_server = MCPServerAdapter(
    server_params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", "./workspace"]
    }
)

# Server 2: GitHub integration
github_server = MCPServerAdapter(
    server_params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-github"],
        "env": {"GITHUB_TOKEN": "ghp_your_token_here"}
    }
)

# Server 3: Database access
db_server = MCPServerAdapter(
    server_params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-postgres"],
        "env": {"DATABASE_URL": "postgresql://localhost/app"}
    }
)

# Combine tools from all servers
all_tools = fs_server.tools + github_server.tools + db_server.tools

# Agent with access to all three servers
fullstack_agent = Agent(
    role="Full-Stack Developer",
    goal="Manage code, files, and database for the project",
    backstory="""Senior developer who coordinates across file system,
    version control, and database layers.""",
    tools=all_tools,
    verbose=True
)

task = Task(
    description="""Perform a code review workflow:
    1. List open PRs on the repository
    2. Read the changed files from the filesystem
    3. Check if database migrations are needed
    4. Provide a summary with recommendations""",
    expected_output="Code review summary with actionable recommendations",
    agent=fullstack_agent
)

crew = Crew(agents=[fullstack_agent], tasks=[task])
result = crew.kickoff()
print(result.raw)

5. MCP Security Considerations

MCP servers can access sensitive systems. Follow these security practices to protect your infrastructure.

5.1 Secure MCP Configuration

import os
from crewai import Agent, Task, Crew
from crewai.tools import MCPServerAdapter

# NEVER hardcode secrets — use environment variables
github_token = os.environ.get("GITHUB_TOKEN")
db_url = os.environ.get("DATABASE_URL")

if not github_token:
    raise ValueError("GITHUB_TOKEN environment variable required")

# Secure Stdio server with environment-injected secrets
secure_github = MCPServerAdapter(
    server_params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-github"],
        "env": {"GITHUB_TOKEN": github_token}
    }
)

# Secure HTTP server with proper auth headers
secure_api = MCPServerAdapter(
    server_params={
        "url": "https://internal-mcp.company.com/mcp",
        "transport": "streamable-http",
        "headers": {
            "Authorization": f"Bearer {os.environ.get('MCP_API_KEY', '')}",
            "X-Request-Source": "crewai-agent"
        }
    }
)

# Agent with security-conscious configuration
secure_agent = Agent(
    role="Security-Aware Developer",
    goal="Perform tasks while following security best practices",
    backstory="Developer with strong security awareness",
    tools=secure_github.tools + secure_api.tools,
    max_iter=10,  # Limit iterations to prevent runaway
    verbose=True
)

task = Task(
    description="Check repository security alerts and summarize critical issues",
    expected_output="Security alert summary with severity levels",
    agent=secure_agent
)

crew = Crew(agents=[secure_agent], tasks=[task])
result = crew.kickoff()
print(result.raw)