MCP for Practical Builders

pip install mcp

uv add mcp

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Hello Server")

@mcp.tool()
def greet(name: str) -> str:
    """Say hello to someone by name."""
    return f"Hello, {name}! MCP is working."

if __name__ == "__main__":
    mcp.run()

python hello_mcp.py

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My First Server")


@mcp.tool()
def add_numbers(a: float, b: float) -> float:
    """Add two numbers and return the result."""
    return a + b


@mcp.tool()
def reverse_text(text: str) -> str:
    """Reverse a piece of text character by character."""
    return text[::-1]


@mcp.tool()
def word_count(text: str) -> dict:
    """
    Count words, characters, and lines in a piece of text.
    Returns a dict with keys: words, characters, lines.
    """
    return {
        "words": len(text.split()),
        "characters": len(text),
        "lines": len(text.splitlines()),
    }


if __name__ == "__main__":
    mcp.run()

{
  "mcpServers": {
    "my-first-server": {
      "command": "python",
      "args": ["/absolute/path/to/my_first_server.py"]
    }
  }
}

@mcp.tool()
def your_tool(
    param: str,          # type annotation is required
    optional: int = 10,  # defaults work
) -> dict:               # return type is used by the AI
    """
    This description is what Claude reads to decide when to use this tool.
    Be specific. Say what the inputs are and what it returns.

    Bad:  "Does something with text"
    Good: "Count words and characters in a string.
           Returns a dict with 'words' and 'characters' keys."
    """
    return {"words": len(param.split()), "characters": len(param)}

import os
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("File Server")

WORKSPACE = Path.home() / "ai-workspace"
WORKSPACE.mkdir(exist_ok=True)


def safe_path(relative: str) -> Path:
    """
    Resolve a path and verify it is inside WORKSPACE.
    Raises ValueError if the path tries to escape (e.g. using ../).
    """
    resolved = (WORKSPACE / relative).resolve()
    if not str(resolved).startswith(str(WORKSPACE.resolve())):
        raise ValueError(f"Access denied: path is outside the workspace")
    return resolved


@mcp.tool()
def read_file(path: str) -> str:
    """
    Read a file from the workspace. Path is relative to the workspace root.
    Example: read_file("notes/todo.txt")
    """
    try:
        return safe_path(path).read_text(encoding="utf-8")
    except ValueError as e:
        return f"Error: {e}"
    except FileNotFoundError:
        return f"Error: File not found — {path}"


@mcp.tool()
def write_file(path: str, content: str) -> str:
    """
    Write content to a file, creating the file and any missing directories.
    Path is relative to the workspace root.
    """
    try:
        p = safe_path(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content, encoding="utf-8")
        return f"Written {len(content)} characters to {path}"
    except ValueError as e:
        return f"Error: {e}"


@mcp.tool()
def list_files(directory: str = "") -> list[str]:
    """
    List files in a workspace directory. Leave empty for the workspace root.
    Returns relative paths from the workspace root.
    """
    try:
        target = safe_path(directory) if directory else WORKSPACE
        if not target.is_dir():
            return [f"Error: {directory} is not a directory"]
        return sorted(
            str(p.relative_to(WORKSPACE))
            for p in target.rglob("*")
            if p.is_file()
        )
    except ValueError as e:
        return [f"Error: {e}"]


@mcp.tool()
def delete_file(path: str) -> str:
    """Delete a file from the workspace. Cannot delete directories."""
    try:
        p = safe_path(path)
        if not p.exists():
            return f"Error: Not found — {path}"
        if p.is_dir():
            return "Error: Cannot delete a directory with this tool"
        p.unlink()
        return f"Deleted: {path}"
    except ValueError as e:
        return f"Error: {e}"


if __name__ == "__main__":
    mcp.run()

# NEVER DO THIS — SQL injection vulnerability
query = f"SELECT * FROM users WHERE name = '{name}'"

# ALWAYS DO THIS — safe
cursor.execute("SELECT * FROM users WHERE name = ?", (name,))

import sqlite3
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Database Server")

DB_PATH = Path.home() / "ai-workspace" / "data.db"


def get_conn():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # returns dicts instead of tuples
    return conn


@mcp.tool()
def list_tables() -> list[str]:
    """List all tables in the database."""
    with get_conn() as conn:
        rows = conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
        ).fetchall()
    return [row["name"] for row in rows]


@mcp.tool()
def describe_table(table_name: str) -> list[dict]:
    """Return column names and types for a table."""
    # Validate: only alphanumeric + underscore allowed in table names
    if not table_name.replace("_", "").isalnum():
        return [{"error": "Invalid table name"}]
    with get_conn() as conn:
        rows = conn.execute(f"PRAGMA table_info({table_name})").fetchall()
    return [{"column": r["name"], "type": r["type"]} for r in rows]


@mcp.tool()
def query_database(sql: str, limit: int = 50) -> list[dict]:
    """
    Run a SELECT query and return up to `limit` rows.
    Only SELECT statements are allowed — no INSERT, UPDATE, DELETE, or DROP.
    Example: query_database("SELECT * FROM notes WHERE title LIKE '%MCP%'")
    """
    clean = sql.strip().upper()
    if not clean.startswith("SELECT"):
        return [{"error": "Only SELECT queries are allowed"}]
    if ";" in sql:
        return [{"error": "Multiple statements not allowed"}]

    with get_conn() as conn:
        try:
            rows = conn.execute(sql).fetchmany(limit)
            return [dict(r) for r in rows]
        except sqlite3.Error as e:
            return [{"error": str(e)}]


@mcp.tool()
def insert_record(table_name: str, data: dict) -> str:
    """
    Insert a new record into a table.
    Example: insert_record("notes", {"title": "My note", "content": "Hello"})
    """
    if not table_name.replace("_", "").isalnum():
        return "Error: Invalid table name"

    cols = ", ".join(data.keys())
    placeholders = ", ".join(["?" for _ in data])
    with get_conn() as conn:
        try:
            conn.execute(
                f"INSERT INTO {table_name} ({cols}) VALUES ({placeholders})",
                list(data.values())
            )
            conn.commit()
            return f"Inserted 1 record into {table_name}"
        except sqlite3.Error as e:
            return f"Error: {e}"


# Create a demo table on first run
def seed():
    with get_conn() as conn:
        conn.execute("""CREATE TABLE IF NOT EXISTS notes (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            content TEXT,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )""")
        if conn.execute("SELECT COUNT(*) FROM notes").fetchone()[0] == 0:
            conn.executemany(
                "INSERT INTO notes (title, content) VALUES (?, ?)",
                [("MCP Idea", "Build a calendar MCP server"),
                 ("Blog post", "Write about AI observability"),
                 ("Project idea", "Multi-agent pipeline")]
            )
            conn.commit()

seed()

if __name__ == "__main__":
    mcp.run()

import httpx
from bs4 import BeautifulSoup
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Web Tools")


@mcp.tool()
async def fetch_webpage(url: str) -> str:
    """
    Fetch the readable text content of a webpage.
    Returns cleaned text (no HTML tags, scripts, or styles).
    Works best on public pages. Does not work behind login walls.
    """
    if not url.startswith(("http://", "https://")):
        return "Error: URL must start with http:// or https://"

    try:
        async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
            r = await client.get(url, headers={"User-Agent": "Mozilla/5.0"})
            r.raise_for_status()

        soup = BeautifulSoup(r.text, "html.parser")
        for tag in soup(["script", "style", "nav", "footer", "header"]):
            tag.decompose()
        text = soup.get_text(separator="\n", strip=True)
        return text[:8000]  # prevent context overflow

    except httpx.HTTPStatusError as e:
        return f"HTTP {e.response.status_code}: {url}"
    except httpx.RequestError as e:
        return f"Network error: {e}"


@mcp.tool()
async def call_api(url: str, method: str = "GET", headers: dict | None = None, body: dict | None = None) -> dict:
    """
    Make an HTTP API request. Returns status code and response body.
    method: GET, POST, PUT, PATCH, DELETE
    headers: optional dict (for auth tokens, content-type, etc.)
    body: optional dict sent as JSON (for POST/PUT/PATCH)
    """
    allowed = {"GET", "POST", "PUT", "PATCH", "DELETE"}
    if method.upper() not in allowed:
        return {"error": f"method must be one of {allowed}"}
    if not url.startswith(("http://", "https://")):
        return {"error": "URL must start with http:// or https://"}

    try:
        async with httpx.AsyncClient(timeout=15.0) as client:
            r = await client.request(method.upper(), url, headers=headers or {}, json=body)
        try:
            data = r.json()
        except Exception:
            data = r.text
        return {"status_code": r.status_code, "ok": r.is_success, "data": data}
    except httpx.RequestError as e:
        return {"error": str(e)}


if __name__ == "__main__":
    mcp.run()

pip install httpx beautifulsoup4

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Resources Demo")

notes = {
    "project-ideas": "1. Calendar MCP server\n2. AI blog writer\n3. RAG over notes",
    "daily-log": "Finished Chapter 7. Starting Chapter 8.",
}


@mcp.resource("notes://list")
def list_notes() -> str:
    """List all note IDs available."""
    return "\n".join(f"notes://{k}" for k in notes)


@mcp.resource("notes://{note_id}")
def get_note(note_id: str) -> str:
    """Get a specific note by its ID. URI: notes://project-ideas"""
    return notes.get(note_id, f"Note not found: {note_id}")

from mcp.types import PromptMessage, TextContent

@mcp.prompt()
def summarise(document: str, style: str = "bullet") -> list[PromptMessage]:
    """
    Template for summarising a document.
    style: 'bullet' (list of points) or 'paragraph' (prose)
    """
    instruction = (
        "as a bulleted list of key points" if style == "bullet"
        else "as two clear paragraphs"
    )
    return [
        PromptMessage(
            role="user",
            content=TextContent(
                type="text",
                text=f"Summarise the following document {instruction}:\n\n{document}"
            )
        )
    ]

import logging
from functools import wraps

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("mcp-server.log"),
        logging.StreamHandler(),
    ],
)
logger = logging.getLogger(__name__)


def log_call(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f"→ {func.__name__} | args={args} kwargs={kwargs}")
        try:
            result = func(*args, **kwargs)
            logger.info(f"✓ {func.__name__} | result_size={len(str(result))}")
            return result
        except Exception as e:
            logger.error(f"✗ {func.__name__} | error={e}")
            raise
    return wrapper


# Usage:
# @mcp.tool()
# @log_call
# def read_file(path: str) -> str:
#     ...

"""
personal_assistant.py — Complete Personal Content Assistant MCP server.
Gives Claude access to your notes, content database, and the web.

Install:  pip install mcp httpx beautifulsoup4
Run:      python personal_assistant.py
Connect:  add to claude_desktop_config.json
"""

import logging
import sqlite3
from pathlib import Path

import httpx
from bs4 import BeautifulSoup
from mcp.server.fastmcp import FastMCP

# ─── Config ───────────────────────────────────────────────────────────────────

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.FileHandler("assistant.log"), logging.StreamHandler()],
)
logger = logging.getLogger(__name__)

WORKSPACE = Path.home() / "ai-workspace"
WORKSPACE.mkdir(exist_ok=True)
DB_PATH = WORKSPACE / "content.db"

mcp = FastMCP("Personal Content Assistant")


# ─── Safety ───────────────────────────────────────────────────────────────────

def safe_path(rel: str) -> Path:
    resolved = (WORKSPACE / rel).resolve()
    if not str(resolved).startswith(str(WORKSPACE.resolve())):
        raise ValueError("Path is outside the workspace")
    return resolved


# ─── Database ─────────────────────────────────────────────────────────────────

def db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    with db() as conn:
        conn.execute("""CREATE TABLE IF NOT EXISTS drafts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL, content TEXT,
            status TEXT DEFAULT 'draft',
            updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )""")
        conn.commit()

init_db()


# ─── File tools ───────────────────────────────────────────────────────────────

@mcp.tool()
def read_note(path: str) -> str:
    """Read a note file. Path is relative to the workspace."""
    logger.info(f"read_note: {path}")
    try:
        return safe_path(path).read_text(encoding="utf-8")
    except (FileNotFoundError, ValueError) as e:
        return f"Error: {e}"


@mcp.tool()
def write_note(path: str, content: str) -> str:
    """Write or overwrite a note file in the workspace."""
    logger.info(f"write_note: {path}")
    try:
        p = safe_path(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content, encoding="utf-8")
        return f"Saved {len(content)} chars to {path}"
    except ValueError as e:
        return f"Error: {e}"


@mcp.tool()
def list_notes() -> list[str]:
    """List all note files (*.md and *.txt) in the workspace."""
    return sorted(
        str(p.relative_to(WORKSPACE))
        for p in WORKSPACE.rglob("*")
        if p.is_file() and p.suffix in (".md", ".txt")
    )


# ─── Database tools ───────────────────────────────────────────────────────────

@mcp.tool()
def list_drafts(status: str = "all") -> list[dict]:
    """List blog drafts. Filter by status: draft, ready, published, or all."""
    logger.info(f"list_drafts: status={status}")
    with db() as conn:
        if status == "all":
            rows = conn.execute("SELECT id, title, status, updated_at FROM drafts ORDER BY updated_at DESC").fetchall()
        else:
            rows = conn.execute("SELECT id, title, status, updated_at FROM drafts WHERE status=? ORDER BY updated_at DESC", (status,)).fetchall()
    return [dict(r) for r in rows]


@mcp.tool()
def save_draft(title: str, content: str, status: str = "draft") -> str:
    """Save a new blog draft. status: draft, ready, or published."""
    logger.info(f"save_draft: {title}")
    with db() as conn:
        conn.execute("INSERT INTO drafts (title, content, status) VALUES (?, ?, ?)", (title, content, status))
        conn.commit()
    return f"Saved draft: '{title}' ({status})"


@mcp.tool()
def get_draft(draft_id: int) -> dict:
    """Get the full content of a draft by its ID."""
    with db() as conn:
        row = conn.execute("SELECT * FROM drafts WHERE id=?", (draft_id,)).fetchone()
    return dict(row) if row else {"error": f"Draft {draft_id} not found"}


# ─── Web tools ────────────────────────────────────────────────────────────────

@mcp.tool()
async def research_url(url: str) -> str:
    """Fetch and extract readable text from a webpage. Use this to research topics."""
    logger.info(f"research_url: {url}")
    if not url.startswith(("http://", "https://")):
        return "Error: URL must start with http:// or https://"
    try:
        async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
            r = await client.get(url, headers={"User-Agent": "Mozilla/5.0"})
            r.raise_for_status()
        soup = BeautifulSoup(r.text, "html.parser")
        for t in soup(["script", "style", "nav", "footer"]):
            t.decompose()
        return soup.get_text(separator="\n", strip=True)[:8000]
    except Exception as e:
        return f"Error: {e}"


if __name__ == "__main__":
    mcp.run()

{
  "mcpServers": {
    "personal-assistant": {
      "command": "python",
      "args": ["/absolute/path/to/personal_assistant.py"]
    }
  }
}