Your First MCP Server — Tools

# Python SDK (includes FastMCP helper class)
pip install mcp

# MCP Inspector — browser UI for calling tools interactively
# Requires Node.js >= 18
npm install -g @modelcontextprotocol/inspector

# Verify
python -c "import mcp; print(mcp.__version__)"
mcp --version

# hello_server.py — minimal MCP server (Python)
from mcp.server.fastmcp import FastMCP

# 1. Instantiate FastMCP with a server name.
#    This name appears in capability negotiation and tool listings.
mcp = FastMCP("hello-server")


@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers and return their sum.

    Use this when the user wants to perform addition or
    compute a total from two numeric values.
    """
    # Type hints above become the JSON Schema automatically.
    # FastMCP validates incoming arguments before calling this function.
    return a + b


if __name__ == "__main__":
    # mcp.run() starts the stdio transport loop.
    # The process reads JSON-RPC from stdin and writes to stdout.
    # Your shell will appear to hang — that's normal, it's listening.
    mcp.run()

# document_server.py — 3-tool MCP server with full error handling
from __future__ import annotations

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

mcp = FastMCP("document-server")

# ── In-memory document store (replace with real DB in production) ──
DOCUMENTS: dict[str, str] = {
    "mcp-overview": "MCP is an open protocol for connecting AI to tools.",
    "fastmcp-guide": "FastMCP reduces MCP server boilerplate with decorators.",
    "transport-docs": "stdio is local; HTTP/SSE is for shared, remote servers.",
}
NOTES: dict[str, str] = {}


# ─────────────────────────────────────────────
# Tool 1: search_documents
# Pattern: read/query — no side effects
# ─────────────────────────────────────────────
@mcp.tool()
def search_documents(query: str, limit: int = 5) -> list[dict]:
    """Search the document store for documents matching a query string.

    Performs case-insensitive substring matching across document IDs
    and content. Returns a ranked list of matching documents.

    Args:
        query: The search string. Case-insensitive.
        limit: Maximum number of results to return. Default is 5.

    Returns:
        List of dicts with 'id', 'preview', and 'score' fields.
        Empty list if no matches found.
    """
    if not query or not query.strip():
        raise ValueError("query must be a non-empty string")
    if limit < 1 or limit > 50:
        raise ValueError("limit must be between 1 and 50")

    query_lower = query.lower()
    results = []

    for doc_id, content in DOCUMENTS.items():
        score = 0
        if query_lower in doc_id.lower():
            score += 2
        if query_lower in content.lower():
            score += 1
        if score > 0:
            results.append({
                "id": doc_id,
                "preview": content[:120] + ("..." if len(content) > 120 else ""),
                "score": score,
            })

    results.sort(key=lambda r: r["score"], reverse=True)
    return results[:limit]


# ─────────────────────────────────────────────
# Tool 2: calculate_similarity
# Pattern: pure computation — deterministic, no I/O
# ─────────────────────────────────────────────
@mcp.tool()
def calculate_similarity(text_a: str, text_b: str) -> dict:
    """Calculate cosine similarity between two text strings using word overlap.

    Uses a bag-of-words approach: splits both texts into word sets,
    then computes cosine similarity from word frequency vectors.
    Not a deep embedding model — use for quick relevance checks.

    Args:
        text_a: First text to compare.
        text_b: Second text to compare.

    Returns:
        Dict with 'similarity' (0.0–1.0), 'common_words', and
        'interpretation' fields.
    """
    if not text_a or not text_b:
        raise ValueError("Both text_a and text_b must be non-empty strings")

    def word_freq(text: str) -> dict[str, int]:
        words = text.lower().split()
        freq: dict[str, int] = {}
        for w in words:
            w = w.strip(".,!?;:")
            if w:
                freq[w] = freq.get(w, 0) + 1
        return freq

    freq_a = word_freq(text_a)
    freq_b = word_freq(text_b)
    all_words = set(freq_a) | set(freq_b)

    if not all_words:
        return {"similarity": 0.0, "common_words": [], "interpretation": "no words"}

    dot = sum(freq_a.get(w, 0) * freq_b.get(w, 0) for w in all_words)
    mag_a = math.sqrt(sum(v ** 2 for v in freq_a.values()))
    mag_b = math.sqrt(sum(v ** 2 for v in freq_b.values()))

    similarity = dot / (mag_a * mag_b) if mag_a and mag_b else 0.0
    common = sorted(set(freq_a) & set(freq_b))

    if similarity > 0.7:
        interpretation = "highly similar"
    elif similarity > 0.3:
        interpretation = "somewhat related"
    else:
        interpretation = "largely different"

    return {
        "similarity": round(similarity, 4),
        "common_words": common[:10],  # cap at 10 for readability
        "interpretation": interpretation,
    }


# ─────────────────────────────────────────────
# Tool 3: save_note
# Pattern: write/side-effect — mutates state
# ─────────────────────────────────────────────
@mcp.tool()
def save_note(title: str, content: str) -> dict:
    """Save a note with a title and content.

    Notes persist for the lifetime of the server process. Use this
    to store information the LLM wants to remember across turns.

    Args:
        title: Note identifier. Alphanumeric + hyphens only. Max 100 chars.
        content: Note body text. Max 5000 chars.

    Returns:
        Dict with 'success', 'title', and 'total_notes' fields.
    """
    # Validate inputs before mutating state
    if not title or not title.strip():
        raise ValueError("title must be a non-empty string")
    if len(title) > 100:
        raise ValueError("title must be 100 characters or fewer")
    if not all(c.isalnum() or c in "-_ " for c in title):
        raise ValueError("title may only contain letters, numbers, hyphens, underscores, and spaces")
    if not content:
        raise ValueError("content must be a non-empty string")
    if len(content) > 5000:
        raise ValueError(f"content must be 5000 chars or fewer (got {len(content)})")

    NOTES[title.strip()] = content
    return {
        "success": True,
        "title": title.strip(),
        "total_notes": len(NOTES),
    }


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

{
  "mcpServers": {
    "document-server": {
      "command": "/usr/bin/python3",
      "args": ["/absolute/path/to/document_server.py"],
      "env": {
        "PYTHONPATH": "/absolute/path/to/your/project"
      }
    }
  }
}

// Windows example:
// "command": "C:\\Python311\\python.exe",
// "args": ["C:\\Users\\YourName\\projects\\document_server.py"]

// Using a venv (recommended for dependency isolation):
// "command": "/absolute/path/to/venv/bin/python",
// "args": ["/absolute/path/to/document_server.py"]

{
  "mcpServers": {
    "document-server": {
      "command": "python",
      "args": ["document_server.py"],
      "cwd": "/absolute/path/to/your/project"
    }
  }
}

// After saving, restart Claude Code or run:
// /mcp  — to view currently connected servers and their tools

from mcp.server.fastmcp import FastMCP
import httpx

mcp = FastMCP("error-demo-server")


# PATTERN 1: Input validation — raise ValueError with descriptive message
@mcp.tool()
def get_document(doc_id: str) -> dict:
    """Retrieve a document by its ID."""
    if not doc_id or not doc_id.strip():
        raise ValueError("doc_id must be a non-empty string")   # LLM sees this message
    if doc_id not in DOCUMENTS:
        raise KeyError(f"Document '{doc_id}' not found. Available: {list(DOCUMENTS.keys())}")
    return {"id": doc_id, "content": DOCUMENTS[doc_id]}


# PATTERN 2: External I/O — always wrap in try/except
@mcp.tool()
async def fetch_url(url: str) -> dict:
    """Fetch the content of a URL and return the status code and body."""
    if not url.startswith(("http://", "https://")):
        raise ValueError("url must start with http:// or https://")

    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.get(url)
            response.raise_for_status()  # raises httpx.HTTPStatusError on 4xx/5xx
            return {
                "status_code": response.status_code,
                "body_preview": response.text[:500],
            }
    except httpx.TimeoutException:
        raise RuntimeError(f"Request to {url} timed out after 10 seconds")
    except httpx.HTTPStatusError as e:
        raise RuntimeError(f"HTTP {e.response.status_code} error from {url}: {e.response.text[:200]}")
    except httpx.RequestError as e:
        raise RuntimeError(f"Network error fetching {url}: {str(e)}")


# What happens when save_note raises ValueError:
# The client receives:
# {
#   "jsonrpc": "2.0",
#   "id": 42,
#   "result": {
#     "isError": true,
#     "content": [{"type": "text", "text": "ValueError: title must be non-empty"}]
#   }
# }
# The LLM reads "ValueError: title must be non-empty" and can fix the call.

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