MCP Integration Patterns

import asyncio
import os
from agents import Agent, OpenAIChatCompletionsModel, Runner
from agents.mcp import MCPServer, MCPServerStdio
from openai import AsyncOpenAI
from dotenv import load_dotenv

load_dotenv()

# Configure OpenAI client (using Nebius in this example)
api_key = os.environ["NEBIUS_API_KEY"]
base_url = "https://api.tokenfactory.nebius.com/v1"
client = AsyncOpenAI(base_url=base_url, api_key=api_key)

async def run(mcp_server: MCPServer, repo_url: str):
    parts = repo_url.strip("/").split("/")
    owner = parts[-2] if len(parts) >= 2 else None
    repo = parts[-1] if len(parts) >= 1 else None
    
    if not owner or not repo:
        print("Invalid repository URL")
        return
    
    # Create agent with MCP server
    agent = Agent(
        name="GitHub Assistant",
        instructions=f"""
        You are a GitHub repository analyzer for {repo_url}.
        Use list_issues with sort='created' and direction='desc' for latest issues.
        Use list_commits to get latest commits.
        Numeric parameters like per_page should be numbers, not strings.
        """,
        mcp_servers=[mcp_server],  # Attach MCP server
        model=OpenAIChatCompletionsModel(
            model="meta-llama/Meta-Llama-3.1-8B-Instruct",
            openai_client=client
        )
    )
    
    # Run agent with MCP tools available
    result = await Runner.run(
        starting_agent=agent,
        input="Analyze the most recent issue and commit"
    )
    print(result.final_output)

async def main():
    # Connect to GitHub MCP server
    async with MCPServerStdio(
        cache_tools_list=True,
        params={
            "command": "npx",
            "args": [
                "-y",
                "@modelcontextprotocol/server-github"
            ],
            "env": {
                "GITHUB_PERSONAL_ACCESS_TOKEN": os.environ["GITHUB_PERSONAL_ACCESS_TOKEN"],
            }
        },
    ) as server:
        await run(server, "owner/repo")

if __name__ == "__main__":
    asyncio.run(main())

from strands import Agent
from strands.mcp import MCPServerStdio
import os

async def main():
    # Initialize MCP server
    async with MCPServerStdio(
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-github"],
            "env": {
                "GITHUB_PERSONAL_ACCESS_TOKEN": os.getenv("GITHUB_TOKEN")
            }
        }
    ) as mcp_server:
        # Create agent with MCP tools
        agent = Agent(
            model=model,
            system_prompt="You are a GitHub assistant",
            mcp_servers=[mcp_server]
        )
        
        result = agent("List recent issues in owner/repo")
        print(result)

from typing import Optional, Dict, Any
from mcp.server.fastmcp import FastMCP
import smtplib
from email.message import EmailMessage

# Initialize FastMCP server
mcp = FastMCP("email")

# Global configuration
SENDER_NAME: Optional[str] = None
SENDER_EMAIL: Optional[str] = None
SENDER_PASSKEY: Optional[str] = None

@mcp.tool()
def configure_email(
    sender_name: str,
    sender_email: str,
    sender_passkey: str
) -> Dict[str, Any]:
    """Configure email sender details.
    
    Args:
        sender_name: Name of the email sender
        sender_email: Email address of the sender
        sender_passkey: App password for authentication
    """
    global SENDER_NAME, SENDER_EMAIL, SENDER_PASSKEY
    SENDER_NAME = sender_name
    SENDER_EMAIL = sender_email
    SENDER_PASSKEY = sender_passkey
    
    return {
        "success": True,
        "message": "Email configuration updated successfully"
    }

@mcp.tool()
def send_email(
    receiver_email: str,
    subject: str,
    body: str
) -> Dict[str, Any]:
    """Send an email to specified recipient.
    
    Args:
        receiver_email: Email address of the recipient
        subject: Subject line of the email
        body: Main content of the email
    
    Returns:
        Dictionary containing success status and message
    """
    if not all([SENDER_NAME, SENDER_EMAIL, SENDER_PASSKEY]):
        return {
            "success": False,
            "message": "Email sender not configured. Use configure_email first."
        }
    
    try:
        msg = EmailMessage()
        msg["Subject"] = subject
        msg["From"] = f"{SENDER_NAME} <{SENDER_EMAIL}>"
        msg["To"] = receiver_email
        msg.set_content(body)
        
        with smtplib.SMTP_SSL("smtp.gmail.com", 465) as smtp:
            smtp.login(SENDER_EMAIL, SENDER_PASSKEY)
            smtp.send_message(msg)
        
        return {
            "success": True,
            "message": "Email sent successfully"
        }
    except Exception as e:
        return {
            "success": False,
            "message": f"Error sending email: {str(e)}"
        }

if __name__ == "__main__":
    mcp.run(transport='stdio')

import asyncio
from agents import Agent, Runner
from agents.mcp import MCPServerStdio

async def main():
    async with MCPServerStdio(
        params={
            "command": "python",
            "args": ["mcp-server.py"],
            "env": {}  # Add any environment variables needed
        }
    ) as server:
        agent = Agent(
            name="Email Assistant",
            instructions="Help users send emails using the MCP tools",
            mcp_servers=[server],
            model=model
        )
        
        # Agent can now use configure_email and send_email tools
        result = await Runner.run(
            starting_agent=agent,
            input="Send an email to [email protected] about the meeting"
        )
        print(result.final_output)

asyncio.run(main())

from mcp.server.fastmcp import FastMCP
import sqlite3
from typing import List, Dict, Any

mcp = FastMCP("database")

# Database connection
DB_PATH = "data/app.db"

@mcp.tool()
def query_database(sql: str) -> List[Dict[str, Any]]:
    """
    Execute a SELECT query on the database.
    
    Args:
        sql: SQL SELECT statement
    
    Returns:
        List of rows as dictionaries
    """
    # Validate it's a SELECT query
    if not sql.strip().upper().startswith("SELECT"):
        return [{"error": "Only SELECT queries are allowed"}]
    
    try:
        conn = sqlite3.connect(DB_PATH)
        conn.row_factory = sqlite3.Row  # Return rows as dicts
        cursor = conn.cursor()
        
        cursor.execute(sql)
        rows = [dict(row) for row in cursor.fetchall()]
        
        conn.close()
        return rows
    except Exception as e:
        return [{"error": str(e)}]

@mcp.tool()
def get_schema() -> Dict[str, List[str]]:
    """
    Get database schema (table names and columns).
    
    Returns:
        Dictionary mapping table names to column lists
    """
    try:
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
        
        # Get all tables
        cursor.execute(
            "SELECT name FROM sqlite_master WHERE type='table'"
        )
        tables = [row[0] for row in cursor.fetchall()]
        
        schema = {}
        for table in tables:
            cursor.execute(f"PRAGMA table_info({table})")
            columns = [row[1] for row in cursor.fetchall()]
            schema[table] = columns
        
        conn.close()
        return schema
    except Exception as e:
        return {"error": str(e)}

if __name__ == "__main__":
    mcp.run(transport='stdio')

async def database_agent():
    async with MCPServerStdio(
        params={
            "command": "python",
            "args": ["database_mcp_server.py"]
        }
    ) as server:
        agent = Agent(
            name="Database Analyst",
            instructions="""
            You are a database analyst. Help users query the database.
            
            Always:
            1. First call get_schema() to understand the database structure
            2. Construct safe SELECT queries based on user questions
            3. Explain the results in plain language
            4. Never use DELETE, UPDATE, or DROP statements
            """,
            mcp_servers=[server],
            model=model
        )
        
        result = await Runner.run(
            starting_agent=agent,
            input="Show me the top 10 customers by total orders"
        )
        print(result.final_output)

from mcp.server.fastmcp import FastMCP
from couchbase.cluster import Cluster
from couchbase.auth import PasswordAuthenticator
from couchbase.options import ClusterOptions
from couchbase.vector_search import VectorSearch, VectorQuery
import os

mcp = FastMCP("couchbase")

# Connect to Couchbase
auth = PasswordAuthenticator(
    os.getenv("COUCHBASE_USER"),
    os.getenv("COUCHBASE_PASSWORD")
)
cluster = Cluster(
    os.getenv("COUCHBASE_CONNECTION_STRING"),
    ClusterOptions(auth)
)
bucket = cluster.bucket("knowledge_base")
collection = bucket.default_collection()

@mcp.tool()
def vector_search(
    query_text: str,
    limit: int = 5
) -> List[Dict[str, Any]]:
    """
    Perform vector similarity search.
    
    Args:
        query_text: Text to search for
        limit: Maximum number of results
    
    Returns:
        List of matching documents with scores
    """
    # Generate embedding for query (using your embedding model)
    query_embedding = generate_embedding(query_text)
    
    # Vector search
    search_req = VectorSearch.from_vector_query(
        VectorQuery(
            "embedding",  # field name
            query_embedding,
            num_candidates=limit * 3
        )
    )
    
    results = collection.search(
        search_req,
        limit=limit
    )
    
    return [
        {
            "id": hit.id,
            "score": hit.score,
            "content": hit.fields.get("content"),
            "metadata": hit.fields.get("metadata")
        }
        for hit in results.rows()
    ]

if __name__ == "__main__":
    mcp.run(transport='stdio')

async def multi_mcp_agent():
    # Connect to multiple MCP servers
    async with MCPServerStdio(
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-github"],
            "env": {"GITHUB_PERSONAL_ACCESS_TOKEN": os.getenv("GITHUB_TOKEN")}
        }
    ) as github_server:
        async with MCPServerStdio(
            params={
                "command": "python",
                "args": ["database_mcp_server.py"]
            }
        ) as db_server:
            async with MCPServerStdio(
                params={
                    "command": "python",
                    "args": ["email_mcp_server.py"]
                }
            ) as email_server:
                # Agent has access to all MCP servers
                agent = Agent(
                    name="Multi-Tool Assistant",
                    instructions="""
                    You have access to:
                    - GitHub (issues, PRs, commits)
                    - Database (query customer data)
                    - Email (send notifications)
                    
                    Use the appropriate tool for each task.
                    """,
                    mcp_servers=[
                        github_server,
                        db_server,
                        email_server
                    ],
                    model=model
                )
                
                result = await Runner.run(
                    starting_agent=agent,
                    input="""
                    1. Check for critical GitHub issues
                    2. Query database for affected customers
                    3. Send email notifications to the team
                    """
                )
                print(result.final_output)

# ✅ Good: Validate and sanitize inputs
@mcp.tool()
def query_database(sql: str) -> List[Dict]:
    # Validate it's a SELECT
    if not sql.strip().upper().startswith("SELECT"):
        raise ValueError("Only SELECT queries allowed")
    
    # Prevent SQL injection
    forbidden_keywords = ["DROP", "DELETE", "UPDATE", "INSERT"]
    if any(keyword in sql.upper() for keyword in forbidden_keywords):
        raise ValueError("Forbidden SQL keyword detected")
    
    return execute_query(sql)

# ❌ Bad: No validation
@mcp.tool()
def query_database(sql: str):
    return execute_query(sql)  # Unsafe!

# ✅ Good: Graceful error handling
async def run_with_mcp():
    try:
        async with MCPServerStdio(
            params={...},
            timeout=30.0  # Set timeout
        ) as server:
            agent = Agent(mcp_servers=[server], ...)
            return await Runner.run(agent, input=query)
    except TimeoutError:
        return "MCP server connection timeout"
    except Exception as e:
        logger.error(f"MCP error: {e}")
        return "Error connecting to external tools"

# ❌ Bad: No error handling
async def run_with_mcp():
    async with MCPServerStdio(params={...}) as server:
        # Might crash if server fails
        return await Runner.run(agent, input=query)

# ✅ Good: Cache available tools
async with MCPServerStdio(
    cache_tools_list=True,  # Cache tool discovery
    params={...}
) as server:
    agent = Agent(mcp_servers=[server])

# ❌ Bad: Re-discover tools on every request
async with MCPServerStdio(
    cache_tools_list=False,  # Slower
    params={...}
) as server:
    agent = Agent(mcp_servers=[server])

# ✅ Good: Credentials from environment
async with MCPServerStdio(
    params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-github"],
        "env": {
            "GITHUB_PERSONAL_ACCESS_TOKEN": os.getenv("GITHUB_TOKEN")
        }
    }
) as server:
    pass

# ❌ Bad: Hardcoded credentials
async with MCPServerStdio(
    params={
        "env": {
            "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_hardcoded_token"  # Never!
        }
    }
) as server:
    pass