M07: MCP — Model Context Protocol

{
  "mcpServers": {
    "filesystem": {
      "command": "python",
      "args": ["-m", "mcp_filesystem_server"],
      "env": {
        "ALLOWED_PATHS": "/home/user/projects"
      }
    },
    "database": {
      "command": "node",
      "args": ["mcp-db-server/index.js"],
      "env": {
        "DB_PATH": "./data/app.db"
      }
    },
    "web-api": {
      "command": "python",
      "args": ["-m", "mcp_web_server"],
      "env": {
        "API_BASE_URL": "https://api.example.com"
      }
    }
  }
}

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": { "GITHUB_TOKEN": "${GITHUB_TOKEN}" }
    },
    "postgres-readonly": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {
        "POSTGRES_CONNECTION_STRING": "${PG_READONLY_DSN}"
      }
    },
    "slack": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-slack"],
      "env": {
        "SLACK_BOT_TOKEN": "${SLACK_BOT_TOKEN}",
        "SLACK_CHANNELS": "eng-alerts,deploys"
      }
    },
    "jira": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-jira"],
      "env": {
        "JIRA_HOST": "${JIRA_HOST}",
        "JIRA_TOKEN": "${JIRA_TOKEN}",
        "JIRA_PROJECT_KEYS": "ENG,INFRA"
      }
    }
  }
}

# filesystem_server.py
# pip install "mcp[cli]>=1.0.0"
import os
import json
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("filesystem-server")

# Restrict to allowed paths for security
ALLOWED_ROOT = os.environ.get("ALLOWED_PATHS", os.getcwd())

def is_allowed(path: str) -> bool:
    """Check if path is within allowed directory."""
    resolved = Path(path).resolve()
    return str(resolved).startswith(str(Path(ALLOWED_ROOT).resolve()))

# --- TOOL: Read a file ---
@mcp.tool()
def read_file(path: str) -> str:
    """Read the contents of a file at the given path."""
    try:
        if not is_allowed(path):
            return json.dumps({"error": "Path outside allowed directory"})
        with open(path, "r", encoding="utf-8") as f:
            content = f.read()
        return json.dumps({"path": path, "content": content})
    except FileNotFoundError:
        return json.dumps({"error": f"File not found: {path}"})
    except PermissionError:
        return json.dumps({"error": f"Permission denied: {path}"})
    except Exception as e:
        return json.dumps({"error": f"Unexpected error: {str(e)}"})

# --- TOOL: Write a file ---
@mcp.tool()
def write_file(path: str, content: str) -> str:
    """Write content to a file at the given path."""
    try:
        if not is_allowed(path):
            return json.dumps({"error": "Path outside allowed directory"})
        os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            f.write(content)
        return json.dumps({"status": "ok", "path": path, "bytes": len(content)})
    except PermissionError:
        return json.dumps({"error": f"Permission denied: {path}"})
    except Exception as e:
        return json.dumps({"error": f"Unexpected error: {str(e)}"})

# --- TOOL: List directory ---
@mcp.tool()
def list_directory(path: str) -> str:
    """List files and subdirectories at the given path."""
    try:
        if not is_allowed(path):
            return json.dumps({"error": "Path outside allowed directory"})
        entries = []
        for entry in os.scandir(path):
            entries.append({
                "name": entry.name,
                "type": "dir" if entry.is_dir() else "file",
                "size": entry.stat().st_size if entry.is_file() else None
            })
        return json.dumps({"path": path, "entries": entries})
    except FileNotFoundError:
        return json.dumps({"error": f"Directory not found: {path}"})
    except Exception as e:
        return json.dumps({"error": f"Unexpected error: {str(e)}"})

# --- RESOURCE: File tree of working directory ---
@mcp.resource("file-tree://cwd")
def get_file_tree() -> str:
    """Browse the current working directory file tree."""
    tree = []
    for root, dirs, files in os.walk(ALLOWED_ROOT):
        depth = root.replace(ALLOWED_ROOT, "").count(os.sep)
        if depth > 3:  # Limit depth to prevent huge outputs
            continue
        indent = "  " * depth
        tree.append(f"{indent}{os.path.basename(root)}/")
        for f in files[:20]:  # Limit files per directory
            tree.append(f"{indent}  {f}")
    return "\n".join(tree)

# --- PROMPT: File summary ---
@mcp.prompt()
def file_summary(path: str) -> str:
    """Generate a structured file analysis prompt."""
    return (
        f"Analyze the file at '{path}'. Provide:\n"
        "1. A one-line summary of what the file does\n"
        "2. Key functions/classes and their purposes\n"
        "3. Dependencies and imports\n"
        "4. Potential issues or improvements\n"
        "5. A complexity rating (Low/Medium/High)"
    )

# --- Start server (stdio transport) ---
if __name__ == "__main__":
    mcp.run()

# database_server.py
# pip install "mcp[cli]>=1.0.0"
import os
import json
import sqlite3
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("database-server")
DB_PATH = os.environ.get("DB_PATH", "./data/app.db")

def get_conn():
    """Create a database connection with safety settings."""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA query_only = ON")  # Read-only by default
    return conn

# --- TOOL: Run a SQL query ---
@mcp.tool()
def query(sql: str) -> str:
    """Execute a read-only SQL query and return results."""
    try:
        conn = get_conn()
        cursor = conn.execute(sql)
        columns = [desc[0] for desc in cursor.description] if cursor.description else []
        rows = [dict(row) for row in cursor.fetchall()[:100]]  # Limit rows
        conn.close()
        return json.dumps({"columns": columns, "rows": rows, "count": len(rows)})
    except sqlite3.OperationalError as e:
        return json.dumps({"error": f"SQL error: {str(e)}"})
    except Exception as e:
        return json.dumps({"error": f"Unexpected error: {str(e)}"})

# --- TOOL: List tables ---
@mcp.tool()
def list_tables() -> str:
    """List all tables in the database."""
    try:
        conn = get_conn()
        cursor = conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
        )
        tables = [row["name"] for row in cursor.fetchall()]
        conn.close()
        return json.dumps({"tables": tables})
    except Exception as e:
        return json.dumps({"error": str(e)})

# --- TOOL: Describe table schema ---
@mcp.tool()
def describe_table(table_name: str) -> str:
    """Get the schema of a specific table."""
    try:
        conn = get_conn()
        cursor = conn.execute(f"PRAGMA table_info({table_name})")
        columns = [
            {"name": row["name"], "type": row["type"], "nullable": not row["notnull"]}
            for row in cursor.fetchall()
        ]
        conn.close()
        if not columns:
            return json.dumps({"error": f"Table not found: {table_name}"})
        return json.dumps({"table": table_name, "columns": columns})
    except Exception as e:
        return json.dumps({"error": str(e)})

# --- RESOURCE: Database schema overview ---
@mcp.resource("db://schema")
def get_schema() -> str:
    """Browse the full database schema."""
    try:
        conn = get_conn()
        tables = conn.execute(
            "SELECT sql FROM sqlite_master WHERE type='table'"
        ).fetchall()
        conn.close()
        return "\n\n".join(row["sql"] for row in tables if row["sql"])
    except Exception as e:
        return f"Error: {str(e)}"

# --- PROMPT: Data exploration ---
@mcp.prompt()
def explore_data(table_name: str) -> str:
    """Prompt template for exploring a database table."""
    return (
        f"Explore the '{table_name}' table. Provide:\n"
        "1. Row count and column summary\n"
        "2. Sample of 5 representative rows\n"
        "3. Distribution of key columns\n"
        "4. Any data quality issues (nulls, duplicates)\n"
        "5. Suggested queries for deeper analysis"
    )

# --- Start server (stdio transport) ---
if __name__ == "__main__":
    mcp.run()

mkdir my-mcp-server && cd my-mcp-server
python -m venv venv && source venv/bin/activate   # Windows: venv\Scripts\activate
pip install "mcp[cli]>=1.0.0"

# filesystem_server.py — MCP server with tools, resources, and prompts
# pip install "mcp[cli]>=1.0.0"
import os
import json
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("filesystem-server")

# Restrict to allowed paths for security
ALLOWED_ROOT = os.environ.get("ALLOWED_PATHS", os.getcwd())

def is_allowed(path: str) -> bool:
    """Check if path is within allowed directory (follows symlinks)."""
    resolved = Path(path).resolve()
    return str(resolved).startswith(str(Path(ALLOWED_ROOT).resolve()))

# ── TOOL: Read a file ────────────────────────────────────────
@mcp.tool()
def read_file(path: str) -> str:
    """Read the contents of a file at the given path."""
    try:
        if not is_allowed(path):
            return json.dumps({"error": "Path outside allowed directory"})
        with open(path, "r", encoding="utf-8") as f:
            content = f.read()
        return json.dumps({"path": path, "content": content})
    except FileNotFoundError:
        return json.dumps({"error": f"File not found: {path}"})
    except PermissionError:
        return json.dumps({"error": f"Permission denied: {path}"})
    except Exception as e:
        return json.dumps({"error": f"Unexpected error: {str(e)}"})

# ── TOOL: Write a file ───────────────────────────────────────
@mcp.tool()
def write_file(path: str, content: str) -> str:
    """Write content to a file at the given path."""
    try:
        if not is_allowed(path):
            return json.dumps({"error": "Path outside allowed directory"})
        os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            f.write(content)
        return json.dumps({"status": "ok", "path": path, "bytes": len(content)})
    except PermissionError:
        return json.dumps({"error": f"Permission denied: {path}"})
    except Exception as e:
        return json.dumps({"error": f"Unexpected error: {str(e)}"})

# ── TOOL: List directory ─────────────────────────────────────
@mcp.tool()
def list_directory(path: str) -> str:
    """List files and subdirectories at the given path."""
    try:
        if not is_allowed(path):
            return json.dumps({"error": "Path outside allowed directory"})
        entries = []
        for entry in os.scandir(path):
            entries.append({
                "name": entry.name,
                "type": "dir" if entry.is_dir() else "file",
                "size": entry.stat().st_size if entry.is_file() else None
            })
        return json.dumps({"path": path, "entries": entries})
    except FileNotFoundError:
        return json.dumps({"error": f"Directory not found: {path}"})
    except Exception as e:
        return json.dumps({"error": f"Unexpected error: {str(e)}"})

# ── RESOURCE: File tree of working directory ──────────────────
@mcp.resource("file-tree://cwd")
def get_file_tree() -> str:
    """Browse the current working directory file tree."""
    tree = []
    for root, dirs, files in os.walk(ALLOWED_ROOT):
        depth = root.replace(ALLOWED_ROOT, "").count(os.sep)
        if depth > 3:
            continue
        indent = "  " * depth
        tree.append(f"{indent}{os.path.basename(root)}/")
        for f in files[:20]:
            tree.append(f"{indent}  {f}")
    return "\n".join(tree)

# ── PROMPT: File analysis template ────────────────────────────
@mcp.prompt()
def file_summary(path: str) -> str:
    """Generate a structured file analysis prompt."""
    return (
        f"Analyze the file at '{path}'. Provide:\n"
        "1. A one-line summary of what the file does\n"
        "2. Key functions/classes and their purposes\n"
        "3. Dependencies and imports\n"
        "4. Potential issues or improvements\n"
        "5. A complexity rating (Low/Medium/High)"
    )

# ── Start server (stdio transport) ───────────────────────────
if __name__ == "__main__":
    mcp.run()

{
  "mcpServers": {
    "my-filesystem": {
      "command": "python",
      "args": ["/full/path/to/my-mcp-server/filesystem_server.py"],
      "env": {
        "ALLOWED_PATHS": "/home/user/projects"
      }
    }
  }
}