The USB-C of AI: Model Context Protocol (MCP) Tutorial + Security Playbook (Python)

#!/usr/bin/env python3
# server.py
# Minimal, production-leaning Model Context Protocol (MCP) server over stdio.
# Tools: search (local), db_query (SQLite read-only), file_read (allow-listed).
# Security: allow-lists, read-only DB, timeouts, rate limits, audits, content firewall.

import sys, os, json, time, sqlite3, threading, queue, re
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
# -----------------------------
# Configuration (env overrides)
# -----------------------------
ALLOWED_DIRS = [Path(p).resolve() for p in os.getenv("MCP_ALLOWED_DIRS", ".").split(os.pathsep)]
SQLITE_PATH = os.getenv("MCP_SQLITE_PATH", "mcp_demo.db")
AUDIT_PATH  = os.getenv("MCP_AUDIT_PATH", "audit.log")
RATE_CALLS  = int(os.getenv("MCP_RATE_CALLS", "30"))      # calls per window
RATE_WINDOW = float(os.getenv("MCP_RATE_WINDOW", "60.0")) # seconds
# Demo local "search index" (swap with your own indexer later)
LOCAL_DOCS = [
    {"id":"guide",   "title":"Model Context Protocol (MCP) Guide", "text":"MCP connects AI clients to tools safely."},
    {"id":"security","title":"MCP Security Checklist",             "text":"Allow-lists, audits, and timeouts reduce risk."},
    {"id":"python",  "title":"Python MCP Server",                  "text":"A stdio JSON server exposes MCP tools."},
]
# -----------------------------
# Utilities
# -----------------------------
def to_json(obj: Any) -> str:
    return json.dumps(obj, ensure_ascii=False, separators=(",", ":"))
def now_ms() -> int:
    return int(time.time() * 1000)
def normalize_path(p: str) -> Path:
    return Path(p).expanduser().resolve()
def path_is_allowed(p: Path) -> bool:
    try:
        return any(str(p).startswith(str(base)) for base in ALLOWED_DIRS)
    except Exception:
        return False
# -----------------------------
# Content Firewall (very simple)
# -----------------------------
INJECTION_PATTERNS = [
    r"(?i)\bignore\s+all\s+previous\s+instructions\b",
    r"(?i)\bexfiltrate\b",
    r"(?i)\brm\s+-rf\b",
    r"(?i)\bshutdown\b",
    r"(?i)\bdisable\s+security\b",
    r"(?i)\bLOAD\s+DATA\b",
]
def content_firewall(payload: Dict[str, Any]) -> Optional[str]:
    """
    Return reason string if blocked, else None.
    Checks tool name, params strings for dangerous intents.
    """
    s = to_json(payload)
    for pat in INJECTION_PATTERNS:
        if re.search(pat, s):
            return f"Blocked by content firewall: pattern '{pat}'"
    return None
# -----------------------------
# Rate Limiter (token bucket)
# -----------------------------
class RateLimiter:
    def __init__(self, calls: int, window: float):
        self.calls = calls
        self.window = window
        self.lock = threading.Lock()
        self.events: List[float] = []
    def allow(self) -> bool:
        with self.lock:
            t = time.time()
            self.events = [e for e in self.events if t - e <= self.window]
            if len(self.events) < self.calls:
                self.events.append(t)
                return True
            return False
limiter = RateLimiter(RATE_CALLS, RATE_WINDOW)
# -----------------------------
# Audit Logger (JSON Lines)
# -----------------------------
audit_lock = threading.Lock()
def audit(event: Dict[str, Any]) -> None:
    event["ts"] = now_ms()
    with audit_lock, open(AUDIT_PATH, "a", encoding="utf-8") as f:
        f.write(to_json(event) + "\n")
# -----------------------------
# DB (SQLite read-only)
# -----------------------------
def connect_sqlite_ro(db_path: str) -> sqlite3.Connection:
    uri = f"file:{Path(db_path).resolve()}?mode=ro"
    return sqlite3.connect(uri, uri=True, check_same_thread=False)
def run_query_ro(conn: sqlite3.Connection, sql: str, params: List[Any]) -> List[Dict[str, Any]]:
    sql_stripped = sql.strip().upper()
    if not sql_stripped.startswith("SELECT"):
        raise ValueError("Only SELECT queries are allowed in read-only mode.")
    cur = conn.execute(sql, params or [])
    cols = [c[0] for c in cur.description]
    rows = [dict(zip(cols, r)) for r in cur.fetchall()]
    return rows
# -----------------------------
# Tool Implementations
# -----------------------------
def tool_search(params: Dict[str, Any]) -> Dict[str, Any]:
    q = (params.get("query") or "").strip().lower()
    if not q:
        return {"results": []}
    scored = []
    for doc in LOCAL_DOCS:
        text = f"{doc['title']} {doc['text']}".lower()
        score = sum(1 for w in q.split() if w in text)
        if score > 0:
            scored.append((score, doc))
    scored.sort(key=lambda x: x[0], reverse=True)
    return {"results": [d for _, d in scored[:10]]}
def tool_db_query(params: Dict[str, Any]) -> Dict[str, Any]:
    sql = params.get("sql") or ""
    query_params = params.get("params") or []
    conn = connect_sqlite_ro(SQLITE_PATH)
    try:
        rows = run_query_ro(conn, sql, query_params)
        return {"rows": rows, "count": len(rows)}
    finally:
        conn.close()
def tool_file_read(params: Dict[str, Any]) -> Dict[str, Any]:
    path = normalize_path(params.get("path") or "")
    if not path_is_allowed(path):
        raise PermissionError("Path not allowed by server policy.")
    if not path.exists() or not path.is_file():
        raise FileNotFoundError("File not found.")
    # size limit (1 MB)
    if path.stat().st_size > 1_000_000:
        raise ValueError("File too large; max 1MB.")
    with open(path, "r", encoding="utf-8", errors="ignore") as f:
        return {"path": str(path), "content": f.read()}
TOOLS = {
    "search": {
        "description": "Local keyword search over a small demo index.",
        "input_schema": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]},
        "fn": tool_search,
    },
    "db_query": {
        "description": "Read-only SQLite SELECT query.",
        "input_schema": {
            "type": "object",
            "properties": {
                "sql": {"type": "string"},
                "params": {"type": "array"}
            },
            "required": ["sql"]
        },
        "fn": tool_db_query,
    },
    "file_read": {
        "description": "Read text files from approved folders.",
        "input_schema": {
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"]
        },
        "fn": tool_file_read,
    }
}
# Allow-list which tools are callable (override via env)
ALLOWED_TOOLS = set((os.getenv("MCP_ALLOWED_TOOLS") or ",".join(TOOLS.keys())).split(","))
# -----------------------------
# JSON-RPC-ish Stdio Transport
# -----------------------------
INQ: "queue.Queue[str]" = queue.Queue()
def reader():
    for line in sys.stdin:
        line = line.strip()
        if line:
            INQ.put(line)
threading.Thread(target=reader, daemon=True).start()
def respond(id_: Any, result: Any=None, error: Any=None):
    msg = {"jsonrpc":"2.0","id":id_}
    if error is not None:
        msg["error"] = {"message": str(error)}
    else:
        msg["result"] = result
    sys.stdout.write(to_json(msg) + "\n")
    sys.stdout.flush()
def list_tools_payload():
    return {
        "protocol": "mcp-stdio-json",
        "tools": [
            {"name": name, "description": meta["description"], "input_schema": meta["input_schema"]}
            for name, meta in TOOLS.items() if name in ALLOWED_TOOLS
        ],
    }
def handle_call(req: Dict[str, Any]):
    req_id = req.get("id")
    method = req.get("method")
    params = req.get("params") or {}
    # Rate-limit check
    if not limiter.allow():
        audit({"event": "rate_limit", "method": method, "params": params})
        return respond(req_id, error="Rate limit exceeded. Try later.")
    # Content firewall
    blocked = content_firewall({"method": method, "params": params})
    if blocked:
        audit({"event":"blocked", "reason":blocked, "method":method, "params":params})
        return respond(req_id, error=blocked)
    try:
        if method == "ping":
            return respond(req_id, {"ok": True, "ts": now_ms()})
        if method == "list_tools":
            return respond(req_id, list_tools_payload())
        if method == "call_tool":
            name = params.get("name")
            args = params.get("arguments") or {}
            if name not in TOOLS or name not in ALLOWED_TOOLS:
                raise ValueError("Tool not allowed.")
            fn = TOOLS[name]["fn"]
            # Timeout guard
            result_box: List[Any] = []
            exc_box: List[BaseException] = []
            def runner():
                try:
                    result_box.append(fn(args))
                except BaseException as e:
                    exc_box.append(e)
            t = threading.Thread(target=runner, daemon=True)
            t.start(); t.join(timeout=8.0)
            if t.is_alive():
                audit({"event":"timeout", "tool":name, "args":args})
                return respond(req_id, error="Tool call timed out.")
            if exc_box:
                raise exc_box[0]
            audit({"event":"tool_call", "tool":name, "args":args, "ok":True})
            return respond(req_id, {"tool": name, "data": result_box[0]})
        # Unknown
        return respond(req_id, error="Unknown method.")
    except Exception as ex:
        audit({"event":"error", "method":method, "params":params, "error":str(ex)})
        return respond(req_id, error=str(ex))
def main():
    # Warm up: ensure DB exists (demo)
    dbf = Path(SQLITE_PATH)
    if not dbf.exists():
        conn = sqlite3.connect(dbf)
        try:
            conn.executescript("""
                CREATE TABLE IF NOT EXISTS employees(id INTEGER PRIMARY KEY, name TEXT, role TEXT);
                INSERT INTO employees(name, role) VALUES
                  ('Asha','Engineer'), ('Ben','Analyst'), ('Chen','Designer');
            """)
            conn.commit()
        finally:
            conn.close()
    # Announce readiness
    sys.stdout.write(to_json({"jsonrpc":"2.0","method":"ready","params":list_tools_payload()}) + "\n")
    sys.stdout.flush()
    # Event loop
    while True:
        try:
            line = INQ.get()
            req = json.loads(line)
            handle_call(req)
        except KeyboardInterrupt:
            break
        except Exception as e:
            # best-effort error
            sys.stdout.write(to_json({"jsonrpc":"2.0","method":"server_error","params":{"error":str(e)}}) + "\n")
            sys.stdout.flush()
if __name__ == "__main__":
    main()

python3 server.py

# allow only file_read and search
export MCP_ALLOWED_TOOLS="file_read,search"
# restrict file access to ./docs and ./snippets
export MCP_ALLOWED_DIRS="./docs:./snippets"
# use a different SQLite path (read-only mode is enforced at call time)
export MCP_SQLITE_PATH="example.db"

{
  "mcpServers": {
    "my-mcp-tools": {
      "command": "python3",
      "args": ["server.py"],
      "env": {
        "MCP_ALLOWED_DIRS": "./docs:./snippets"
      }
    }
  }
}

# list tools
printf '%s\n' \
'{"jsonrpc":"2.0","id":1,"method":"list_tools"}' | python3 server.py