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 klavis import Klavis
from klavis.types import McpServerName, ToolFormat
import webbrowser

klavis_client = Klavis(api_key=os.getenv("KLAVIS_API_KEY"))

response = klavis_client.mcp_server.create_strata_server(
    servers=[McpServerName.GMAIL, McpServerName.SLACK], 
    user_id="1234"
)

# Handle OAuth authorization for each services
if response.oauth_urls:
    for server_name, oauth_url in response.oauth_urls.items():
        webbrowser.open(oauth_url)
        print(f"Or please open this URL to complete {server_name} OAuth authorization: {oauth_url}")

import json
from crewai import Agent, Task, Crew, Process
from crewai_tools import MCPServerAdapter

def crew_with_mcp_server(mcp_server_url: str, user_query: str):
    klavis_server_params = [
            {
                "url": mcp_server_url,
                "transport": "streamable-http"
            }
        ]

    with MCPServerAdapter(klavis_server_params) as all_mcp_tools:
        print(f"Available tools: {[tool.name for tool in all_mcp_tools]}")

        klavis_agent = Agent(
            role="Klavis Query Assistant",
            goal="Assist the user with their query using available tools",
            backstory="Expert at assisting users with their queries using available tools",
            tools=all_mcp_tools,
            verbose=False,
            llm="gpt-4o"
        )

        klavis_task = Task(
            description=f"Answer the user's query: {user_query}",
            expected_output="Provide a detailed response to the user's query",
            agent=klavis_agent
        )

        crew = Crew(
            agents = [klavis_agent],
            tasks = [klavis_task],
            process=Process.sequential,
            verbose=True
        )

        result = crew.kickoff()
        return result

result = crew_with_mcp_server(
    mcp_server_url=response.strata_server_url, 
    user_query="Check my latest 5 emails and summarize them in a Slack message to #general"
)

print(f"\nFinal Response: {result}")

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")