Real-World MCP Servers: Databases, APIs, and Files

User: "How many orders were placed last month?"
  ↓
Claude → calls query_database tool
  ↓
MCP Server → executes SQL against PostgreSQL
  ↓
Returns: "1,247 orders totaling $89,340"

import asyncpg
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Database Assistant")

# Connection pool (created on startup)
pool = None

@mcp.tool()
async def query_database(sql: str) -> str:
    """Execute a read-only SQL query and return results."""
    # Safety: block write operations
    forbidden = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "TRUNCATE"]
    sql_upper = sql.upper().strip()
    for keyword in forbidden:
        if keyword in sql_upper:
            return f"Error: {keyword} operations are not allowed. This tool is read-only."

    try:
        async with pool.acquire() as conn:
            rows = await conn.fetch(sql)
            if not rows:
                return "Query returned no results."
            # Format as table
            headers = list(rows[0].keys())
            lines = [" | ".join(headers)]
            lines.append("-" * len(lines[0]))
            for row in rows[:50]:  # Limit to 50 rows
                lines.append(" | ".join(str(row[h]) for h in headers))
            return "\n".join(lines)
    except Exception as e:
        return f"Query error: {str(e)}"

@mcp.resource("db://schema")
def get_schema() -> str:
    """Database table schemas for reference."""
    return open("schema.sql").read()

import httpx
import os
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("GitHub Assistant")

GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN")
BASE_URL = "https://api.github.com"

@mcp.tool()
async def list_issues(repo: str, state: str = "open", limit: int = 10) -> str:
    """List GitHub issues for a repository (format: owner/repo)."""
    if "/" not in repo:
        return "Error: repo must be in 'owner/repo' format"

    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{BASE_URL}/repos/{repo}/issues",
            headers={"Authorization": f"Bearer {GITHUB_TOKEN}"},
            params={"state": state, "per_page": min(limit, 30)}
        )
        if response.status_code == 404:
            return f"Repository '{repo}' not found"
        if response.status_code != 200:
            return f"GitHub API error: {response.status_code}"

        issues = response.json()
        if not issues:
            return f"No {state} issues in {repo}"

        lines = [f"Issues in {repo} ({state}):"]
        for issue in issues:
            lines.append(f"  #{issue['number']}: {issue['title']}")
        return "\n".join(lines)

@mcp.tool()
async def create_issue(repo: str, title: str, body: str = "") -> str:
    """Create a new GitHub issue."""
    if "/" not in repo:
        return "Error: repo must be in 'owner/repo' format"

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{BASE_URL}/repos/{repo}/issues",
            headers={"Authorization": f"Bearer {GITHUB_TOKEN}"},
            json={"title": title, "body": body}
        )
        if response.status_code == 201:
            issue = response.json()
            return f"Created issue #{issue['number']}: {issue['html_url']}"
        return f"Failed to create issue: {response.status_code}"

{
  "mcpServers": {
    "github": {
      "command": "python",
      "args": ["/path/to/github_server.py"],
      "env": {
        "GITHUB_TOKEN": "ghp_your_token_here"
      }
    }
  }
}

import os
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("File Assistant")

# SAFETY: Only allow access within this directory
ALLOWED_DIR = os.path.expanduser("~/projects")

def safe_path(filepath: str) -> str:
    """Resolve and validate a file path is within the allowed directory."""
    resolved = os.path.realpath(os.path.join(ALLOWED_DIR, filepath))
    if not resolved.startswith(os.path.realpath(ALLOWED_DIR)):
        raise ValueError(f"Access denied: path outside {ALLOWED_DIR}")
    return resolved

@mcp.tool()
def read_file(filepath: str) -> str:
    """Read a file from the projects directory."""
    try:
        full_path = safe_path(filepath)
        with open(full_path, 'r') as f:
            content = f.read()
        # Truncate very large files
        if len(content) > 10000:
            return content[:10000] + f"\n\n[Truncated — file is {len(content)} characters]"
        return content
    except ValueError as e:
        return f"Error: {str(e)}"
    except FileNotFoundError:
        return f"File not found: {filepath}"

@mcp.tool()
def list_directory(dirpath: str = ".") -> str:
    """List files and directories in a given path."""
    try:
        full_path = safe_path(dirpath)
        entries = os.listdir(full_path)
        dirs = sorted(e + "/" for e in entries if os.path.isdir(os.path.join(full_path, e)))
        files = sorted(e for e in entries if os.path.isfile(os.path.join(full_path, e)))
        return "Directories:\n" + "\n".join(f"  {d}" for d in dirs) + \
               "\n\nFiles:\n" + "\n".join(f"  {f}" for f in files)
    except ValueError as e:
        return f"Error: {str(e)}"

mcp = FastMCP("DevOps Assistant")

# Database tools
@mcp.tool()
async def query_production_db(sql: str) -> str: ...

# GitHub tools
@mcp.tool()
async def list_open_prs(repo: str) -> str: ...

# File system tools
@mcp.tool()
def read_config(filename: str) -> str: ...

# Context resources
@mcp.resource("project://architecture")
def get_architecture_docs() -> str: ...