10 Practical MCP Use Cases (With Code)

from fastmcp import FastMCP
from pathlib import Path

mcp = FastMCP("knowledge")

@mcp.tool()
def search_obsidian(query: str) -> str:
    """Search Obsidian vault for notes matching query."""
    vault = Path.home() / "Documents/Obsidian"
    matches = []
    for md in vault.rglob("*.md"):
        content = md.read_text()
        if query.lower() in content.lower():
            matches.append(f"**{md.stem}**\n{content[:300]}...")
    return "\n\n".join(matches[:5]) or "No matches found"

@mcp.tool()
def get_note(title: str) -> str:
    """Get full content of a specific note."""
    vault = Path.home() / "Documents/Obsidian"
    for md in vault.rglob("*.md"):
        if title.lower() in md.stem.lower():
            return md.read_text()
    return "Note not found"

import subprocess
from fastmcp import FastMCP

mcp = FastMCP("dev-tools")

@mcp.tool()
def git_status(repo_path: str) -> str:
    """Get git status for a repository."""
    result = subprocess.run(
        ["git", "status", "--short"],
        cwd=repo_path,
        capture_output=True,
        text=True
    )
    return result.stdout or "Clean working directory"

@mcp.tool()
def run_tests(repo_path: str, test_pattern: str = "") -> str:
    """Run pytest in a repository."""
    cmd = ["pytest", "-v"]
    if test_pattern:
        cmd.extend(["-k", test_pattern])
    result = subprocess.run(cmd, cwd=repo_path, capture_output=True, text=True)
    return result.stdout + result.stderr

@mcp.tool()
def list_branches(repo_path: str) -> str:
    """List git branches."""
    result = subprocess.run(
        ["git", "branch", "-a"],
        cwd=repo_path,
        capture_output=True,
        text=True
    )
    return result.stdout

import imaplib
import email
from fastmcp import FastMCP

mcp = FastMCP("email")

@mcp.tool()
def get_unread_emails(max_count: int = 10) -> str:
    """Get unread emails from inbox."""
    mail = imaplib.IMAP4_SSL("imap.gmail.com")
    mail.login(EMAIL, PASSWORD)  # Use app password
    mail.select("inbox")
    
    _, messages = mail.search(None, "UNSEEN")
    email_ids = messages[0].split()[-max_count:]
    
    summaries = []
    for eid in email_ids:
        _, data = mail.fetch(eid, "(RFC822)")
        msg = email.message_from_bytes(data[0][1])
        summaries.append(f"From: {msg['from']}\nSubject: {msg['subject']}")
    
    mail.logout()
    return "\n\n".join(summaries)

@mcp.tool()
def search_emails(query: str, folder: str = "inbox") -> str:
    """Search emails by subject or sender."""
    # Implementation searches email headers
    pass

import sqlite3
from fastmcp import FastMCP

mcp = FastMCP("database")

@mcp.tool()
def query_database(sql: str) -> str:
    """Run read-only SQL query. SELECT only."""
    if not sql.strip().upper().startswith("SELECT"):
        return "Error: Only SELECT queries allowed"
    
    conn = sqlite3.connect("app.db")
    cursor = conn.execute(sql)
    columns = [d[0] for d in cursor.description]
    rows = cursor.fetchall()
    conn.close()
    
    # Format as markdown table
    result = "| " + " | ".join(columns) + " |\n"
    result += "| " + " | ".join(["---"] * len(columns)) + " |\n"
    for row in rows[:50]:  # Limit rows
        result += "| " + " | ".join(str(v) for v in row) + " |\n"
    return result

@mcp.tool()
def describe_tables() -> str:
    """List all tables and their schemas."""
    conn = sqlite3.connect("app.db")
    cursor = conn.execute(
        "SELECT name, sql FROM sqlite_master WHERE type='table'"
    )
    tables = cursor.fetchall()
    conn.close()
    return "\n\n".join(f"**{t[0]}**\n```sql\n{t[1]}\n```" for t in tables)

import httpx
from fastmcp import FastMCP

mcp = FastMCP("api-docs")

@mcp.tool()
def get_openapi_spec(api_url: str) -> str:
    """Fetch and summarize an OpenAPI specification."""
    response = httpx.get(f"{api_url}/openapi.json")
    spec = response.json()
    
    endpoints = []
    for path, methods in spec.get("paths", {}).items():
        for method, details in methods.items():
            endpoints.append(
                f"{method.upper()} {path}: {details.get('summary', 'No description')}"
            )
    
    return f"**{spec.get('info', {}).get('title', 'API')}**\n\n" + "\n".join(endpoints)

@mcp.tool()
def test_endpoint(method: str, url: str, data: str = "") -> str:
    """Test an API endpoint."""
    response = httpx.request(method, url, json=eval(data) if data else None)
    return f"Status: {response.status_code}\n\n{response.text[:1000]}"

from datetime import datetime, timedelta
from fastmcp import FastMCP
import caldav

mcp = FastMCP("calendar")

@mcp.tool()
def get_today_events() -> str:
    """Get today's calendar events."""
    client = caldav.DAVClient(url=CALDAV_URL, username=USER, password=PASS)
    calendar = client.principal().calendars()[0]
    
    today = datetime.now().date()
    events = calendar.date_search(today, today + timedelta(days=1))
    
    return "\n".join(
        f"{e.vobject_instance.vevent.dtstart.value}: "
        f"{e.vobject_instance.vevent.summary.value}"
        for e in events
    )

@mcp.tool()
def find_free_time(duration_hours: int, within_days: int = 7) -> str:
    """Find free time slots in calendar."""
    # Implementation checks for gaps between events
    pass

from pathlib import Path
import shutil
from fastmcp import FastMCP

mcp = FastMCP("files")
ALLOWED_DIR = Path.home() / "Documents"

def safe_path(filepath: str) -> Path:
    """Ensure path is within allowed directory."""
    full = (ALLOWED_DIR / filepath).resolve()
    if not full.is_relative_to(ALLOWED_DIR):
        raise ValueError("Path outside allowed directory")
    return full

@mcp.tool()
def find_files(pattern: str, directory: str = ".") -> str:
    """Find files matching a glob pattern."""
    base = safe_path(directory)
    matches = list(base.glob(pattern))[:50]
    return "\n".join(str(m.relative_to(ALLOWED_DIR)) for m in matches)

@mcp.tool()
def read_file(filepath: str) -> str:
    """Read file contents."""
    return safe_path(filepath).read_text()[:10000]

@mcp.tool()
def move_file(source: str, destination: str) -> str:
    """Move a file to a new location."""
    src = safe_path(source)
    dst = safe_path(destination)
    shutil.move(src, dst)
    return f"Moved {source} to {destination}"

import httpx
from fastmcp import FastMCP

mcp = FastMCP("monitoring")

@mcp.tool()
def check_website(url: str) -> str:
    """Check if a website is up and measure response time."""
    try:
        response = httpx.get(url, timeout=10)
        return f"Status: {response.status_code}\n" \
               f"Response time: {response.elapsed.total_seconds():.2f}s"
    except Exception as e:
        return f"Error: {e}"

@mcp.tool()
def get_server_metrics(host: str) -> str:
    """Get CPU, memory, disk usage from a server."""
    # Implementation via SSH or monitoring API
    pass

@mcp.tool()
def query_logs(service: str, hours: int = 1, level: str = "error") -> str:
    """Search logs for a service."""
    # Implementation queries your log aggregator
    pass

import httpx
from fastmcp import FastMCP

mcp = FastMCP("research")

@mcp.tool()
def search_arxiv(query: str, max_results: int = 5) -> str:
    """Search arXiv for papers."""
    response = httpx.get(
        "http://export.arxiv.org/api/query",
        params={"search_query": f"all:{query}", "max_results": max_results}
    )
    # Parse XML response and extract titles, authors, abstracts
    return parse_arxiv_response(response.text)

@mcp.tool()
def get_paper_summary(arxiv_id: str) -> str:
    """Get abstract and details for an arXiv paper."""
    pass

import httpx
from fastmcp import FastMCP

mcp = FastMCP("home")

@mcp.tool()
def set_light(room: str, state: str, brightness: int = 100) -> str:
    """Control a light. state: on/off"""
    response = httpx.post(
        f"{HASS_URL}/api/services/light/turn_{state}",
        headers={"Authorization": f"Bearer {HASS_TOKEN}"},
        json={"entity_id": f"light.{room}", "brightness_pct": brightness}
    )
    return f"Light {room} turned {state}"

@mcp.tool()
def get_temperature(room: str) -> str:
    """Get temperature sensor reading."""
    pass

@mcp.tool()
def run_scene(scene_name: str) -> str:
    """Activate a home automation scene."""
    pass