CrewAI

pip install crewai 'crewai-tools[mcp]' klavis openai

import os

# Set environment variables
os.environ["OPENAI_API_KEY"] = "your-openai-api-key-here"  # Replace with your actual OpenAI API key
os.environ["KLAVIS_API_KEY"] = "your-klavis-api-key-here"   # Replace with your actual Klavis API key

from crewai import Agent, Task, Crew, Process
from crewai_tools import MCPServerAdapter
from klavis import Klavis
from klavis.types import McpServerName

# Initialize clients
klavis_client = Klavis(api_key=os.getenv("KLAVIS_API_KEY"))

# Create YouTube MCP server
youtube_mcp_instance = klavis_client.mcp_server.create_server_instance(
    server_name=McpServerName.YOUTUBE,
    user_id="1234",
    platform_name="Klavis",
)

print(f"🔗 YouTube MCP server created at: {youtube_mcp_instance.server_url}")

# Configure MCP server parameters for CrewAI
server_params = {
    "url": youtube_mcp_instance.server_url,
    "transport": "streamable-http"
}

YOUTUBE_VIDEO_URL = "https://www.youtube.com/watch?v=LCEmiRjPEtQ"

try:
    with MCPServerAdapter(server_params) as mcp_tools:
        print(f"✅ Available tools: {[tool.name for tool in mcp_tools]}")
        
        # Create a YouTube Analysis Agent
        youtube_agent = Agent(
            role="YouTube Content Analyst",
            goal="Research and analyze YouTube videos to extract comprehensive insights",
            backstory="You are an expert at analyzing video content and creating professional summaries.",
            tools=mcp_tools,
            reasoning=True,
            verbose=False
        )
        
        # Define Task
        analysis_task = Task(
            description=f"Research the YouTube video at {YOUTUBE_VIDEO_URL}. Extract the video transcript, analyze the content, and create a comprehensive summary with key points, timestamps, and main takeaways.",
            expected_output="Complete video analysis with transcript, structured summary, key insights, timestamps, and main takeaways",
            agent=youtube_agent,
            markdown=True
        )
        
        # Create and execute the crew
        youtube_crew = Crew(
            agents=[youtube_agent],
            tasks=[analysis_task],
            verbose=False,
            process=Process.sequential
        )
        
        result = youtube_crew.kickoff()
        print(result)
        
except Exception as e:
    print(f"❌ Error connecting to YouTube MCP server: {e}")

import webbrowser

# Create YouTube and Gmail MCP servers
youtube_mcp_instance = klavis_client.mcp_server.create_server_instance(
    server_name=McpServerName.YOUTUBE,
    user_id="1234",
    platform_name="Klavis",
)

gmail_mcp_instance = klavis_client.mcp_server.create_server_instance(
    server_name=McpServerName.GMAIL,
    user_id="1234",
    platform_name="Klavis",
)

# Handle OAuth for Gmail
webbrowser.open(gmail_mcp_instance.oauth_url)
print(f"🔐 Opening OAuth authorization for Gmail...")
print(f"📱 If not redirected automatically, open: {gmail_mcp_instance.oauth_url}")

VIDEO_URL = "https://www.youtube.com/watch?v=LCEmiRjPEtQ"
RECIPIENT_EMAIL = "your-email@example.com"

# Configure multiple MCP servers
multiple_server_params = [
    {
        "url": youtube_mcp_instance.server_url,
        "transport": "streamable-http"
    },
    {
        "url": gmail_mcp_instance.server_url,
        "transport": "streamable-http"
    }
]

try:
    with MCPServerAdapter(multiple_server_params) as all_mcp_tools:
        print(f"✅ Available tools from all MCP servers: {[tool.name for tool in all_mcp_tools]}")
        
        # Create YouTube Research Agent
        youtube_research_agent = Agent(
            role="YouTube Content Analyst",
            goal="Research and analyze YouTube videos to extract comprehensive insights",
            backstory="You are an expert at analyzing video content and extracting key insights.",
            tools=all_mcp_tools,
            reasoning=False,
            verbose=False,
        )
        
        # Create Email Communication Agent
        email_agent = Agent(
            role="Email Communications Specialist",
            goal="Draft and send professional email communications based on research findings",
            backstory="You are skilled at crafting professional emails with clear, impactful messaging.",
            tools=all_mcp_tools,
            reasoning=True,
            verbose=False,
        )
        
        # Define workflow tasks
        youtube_research_task = Task(
            description=f"Research the YouTube video at {VIDEO_URL}. Extract transcript, analyze the content for key insights about AI and software development, and create a comprehensive analysis report with key takeaways and recommendations.",
            expected_output="Complete video analysis report with transcript, key insights, recommendations, and strategic takeaways",
            agent=youtube_research_agent
        )
        
        send_email_task = Task(
            description=f"Based on the youtube analysis, draft and send a professional email to {RECIPIENT_EMAIL} with the subject 'YouTube Video AI Analysis'. Include content of the youtube video in the email.",
            expected_output="Confirmation that a professional email has been sent with the research insights",
            agent=email_agent,
            context=[youtube_research_task]
        )
        
        # Create and execute the crew
        multi_service_crew = Crew(
            agents=[youtube_research_agent, email_agent],
            tasks=[youtube_research_task, send_email_task],
            verbose=False,
            process=Process.sequential
        )
        
        result = multi_service_crew.kickoff()
        print(result)
        
except Exception as e:
    print(f"❌ Error with multi-service MCP integration: {e}")

def create_secure_crew():
    """Demonstrates secure MCP server integration with CrewAI"""
    
    # 1. Use environment variables for sensitive data
    api_key = os.getenv("KLAVIS_API_KEY")
    if not api_key:
        raise ValueError("KLAVIS_API_KEY environment variable is required")
    
    # 2. Validate server URLs (use HTTPS in production)
    server_params = [{
        "url": server_instance.server_url,
        "transport": "streamable-http"
    }]
    
    # 3. Always use context managers for proper resource cleanup
    try:
        with MCPServerAdapter(server_params) as mcp_tools:
            # 4. Validate available tools before use
            if not mcp_tools:
                raise ValueError("No tools available from MCP server")
            
            print(f"✅ Securely connected with {len(mcp_tools)} tools")
            
            # 5. Create agents with limited scope
            agent = Agent(
                role="Data Analyst",
                goal="Analyze data within defined parameters",
                backstory="You operate within strict security guidelines.",
                tools=mcp_tools,
                reasoning=False,  # Disable for production
                verbose=False     # Disable verbose logging in production
            )
            
            return agent
            
    except Exception as e:
        print(f"🔒 Security check failed: {e}")
        return None

# Example usage
secure_agent = create_secure_crew()
if secure_agent:
    print("✅ Secure crew created successfully")