Tools y Resources en Python: Decoradores, Pydantic, y Diferencias con TypeScript

Descripción de la cápsula

En la cápsula anterior creaste tu primer MCP server Python y viste la magia de los decoradores: @mcp.tool() extrae nombre, descripción, y schema de tu función automáticamente. Ahora es momento de ir más profundo.

Esta cápsula cubre la implementación completa de tools y resources en Python: validación avanzada con Pydantic, manejo de tipos complejos, patrones de diseño para tools robustos, resources dinámicos con templates de URI, y las diferencias idiomáticas que hacen que el SDK de Python se sienta diferente al de TypeScript.

El objetivo es que al terminar esta cápsula, puedas implementar cualquier tool o resource que necesites para tu MCP server Python, con validación sólida y error handling adecuado.

Tools avanzados con @mcp.tool()

Más allá del "Hello World"

Los tools del setup eran simples: una función, un string de retorno. Los tools reales necesitan:

• Validación de inputs complejos
• Múltiples parámetros con tipos variados
• Manejo de errores robusto
• Retorno de contenido estructurado
• Interacción con servicios externos

Tool con múltiples tipos de parámetros

from mcp.server.fastmcp import FastMCP
import json

mcp = FastMCP("advanced-tools")


@mcp.tool()
async def search_items(
    query: str,
    category: str = "all",
    max_results: int = 10,
    include_metadata: bool = False,
    tags: list[str] | None = None,
) -> str:
    """Busca items en el catálogo.

    Permite filtrar por categoría, limitar resultados,
    incluir metadata, y filtrar por tags.
    """
    results = []
    for i in range(min(max_results, 5)):
        item = {
            "id": i + 1,
            "name": f"Item que coincide con '{query}' #{i + 1}",
            "category": category,
        }
        if include_metadata:
            item["metadata"] = {"relevance": 0.95 - (i * 0.1), "source": "catalog"}
        if tags:
            item["matched_tags"] = tags
        results.append(item)

    return json.dumps(results, indent=2, ensure_ascii=False)

El schema JSON generado automáticamente:

{
  "type": "object",
  "properties": {
    "query": { "type": "string" },
    "category": { "type": "string", "default": "all" },
    "max_results": { "type": "integer", "default": 10 },
    "include_metadata": { "type": "boolean", "default": false },
    "tags": {
      "anyOf": [
        { "type": "array", "items": { "type": "string" } },
        { "type": "null" }
      ],
      "default": null
    }
  },
  "required": ["query"]
}

El SDK convierte los type hints de Python a JSON Schema automáticamente. list[str] | None se convierte en un schema con anyOf.

Tool con validación Pydantic

Para inputs complejos, usa Pydantic models como parámetro:

from pydantic import BaseModel, Field


class CreateTaskInput(BaseModel):
    title: str = Field(min_length=1, max_length=200, description="Título de la tarea")
    description: str = Field(default="", max_length=2000, description="Descripción detallada")
    priority: int = Field(default=3, ge=1, le=5, description="Prioridad del 1 (máxima) al 5 (mínima)")
    assignee: str | None = Field(default=None, description="Persona asignada")
    tags: list[str] = Field(default_factory=list, description="Etiquetas para organizar")


tasks: list[dict] = []


@mcp.tool()
async def create_task(input: CreateTaskInput) -> str:
    """Crea una nueva tarea con validación completa."""
    task = {
        "id": len(tasks) + 1,
        "title": input.title,
        "description": input.description,
        "priority": input.priority,
        "assignee": input.assignee,
        "tags": input.tags,
        "status": "pending",
    }
    tasks.append(task)
    return json.dumps({"message": "Tarea creada", "task": task}, indent=2, ensure_ascii=False)

¿Por qué Pydantic en lugar de type hints simples?

Los type hints simples validan tipos pero no valores. Pydantic agrega:

Comparación: Pydantic vs Zod

Zod (TypeScript):

const CreateTaskSchema = z.object({
  title: z.string().min(1).max(200).describe("Título de la tarea"),
  description: z.string().max(2000).default("").describe("Descripción detallada"),
  priority: z.number().int().min(1).max(5).default(3).describe("Prioridad 1-5"),
  assignee: z.string().optional().describe("Persona asignada"),
  tags: z.array(z.string()).default([]).describe("Etiquetas"),
});

server.tool("create_task", "Crea una nueva tarea", CreateTaskSchema.shape, async (input) => {
  // ...
});

Pydantic (Python):

class CreateTaskInput(BaseModel):
    title: str = Field(min_length=1, max_length=200, description="Título de la tarea")
    description: str = Field(default="", max_length=2000, description="Descripción detallada")
    priority: int = Field(default=3, ge=1, le=5, description="Prioridad 1-5")
    assignee: str | None = Field(default=None, description="Persona asignada")
    tags: list[str] = Field(default_factory=list, description="Etiquetas")

@mcp.tool()
async def create_task(input: CreateTaskInput) -> str:
    """Crea una nueva tarea."""
    # ...

La diferencia clave: en Zod defines el schema separado de la función. En Pydantic, defines un modelo que es a la vez el schema y la estructura de datos. El modelo Pydantic es una clase Python que puedes reutilizar en toda tu aplicación — no solo para MCP.

Tool con error handling

Los tools fallan. APIs caen, archivos no existen, inputs son inválidos. Tu tool debe manejar eso:

import httpx


@mcp.tool()
async def fetch_url(url: str, timeout: int = 10) -> str:
    """Obtiene el contenido de una URL.

    Retorna el contenido de texto de la respuesta HTTP.
    """
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(url, timeout=timeout)
            response.raise_for_status()
            content_type = response.headers.get("content-type", "")
            if "text" in content_type or "json" in content_type:
                return response.text[:5000]
            return f"Respuesta recibida ({response.status_code}), pero el contenido no es texto: {content_type}"
    except httpx.TimeoutException:
        return f"Error: timeout después de {timeout} segundos conectando a {url}"
    except httpx.HTTPStatusError as e:
        return f"Error HTTP {e.response.status_code}: {e.response.text[:500]}"
    except httpx.RequestError as e:
        return f"Error de conexión: {e}"
    except Exception as e:
        return f"Error inesperado: {type(e).__name__}: {e}"

Patrón recomendado: No lances excepciones desde tools. Retorna mensajes de error como texto. El modelo puede leer el error e intentar corregir su approach. Si lanzas una excepción, el SDK la captura pero el mensaje puede ser menos útil.

Tool con patrón dry-run

import os


@mcp.tool()
async def delete_files(
    directory: str,
    pattern: str,
    dry_run: bool = True,
) -> str:
    """Elimina archivos que coincidan con un patrón.

    Por defecto ejecuta en modo dry_run (solo muestra qué se eliminaría).
    Usa dry_run=false para ejecutar la eliminación real.
    """
    import fnmatch

    if not os.path.isdir(directory):
        return f"Error: '{directory}' no es un directorio válido."

    matches = []
    for filename in os.listdir(directory):
        if fnmatch.fnmatch(filename, pattern):
            matches.append(filename)

    if not matches:
        return f"No se encontraron archivos que coincidan con '{pattern}' en {directory}."

    if dry_run:
        file_list = "\n".join(f"  - {f}" for f in matches)
        return f"[DRY RUN] Se eliminarían {len(matches)} archivos:\n{file_list}\n\nUsa dry_run=false para ejecutar."

    deleted = []
    errors = []
    for filename in matches:
        try:
            filepath = os.path.join(directory, filename)
            os.remove(filepath)
            deleted.append(filename)
        except OSError as e:
            errors.append(f"{filename}: {e}")

    result = f"Eliminados: {len(deleted)} archivos."
    if errors:
        result += f"\nErrores: {len(errors)}\n" + "\n".join(f"  - {e}" for e in errors)
    return result

El patrón dry-run es fundamental para tools destructivos. El modelo primero ejecuta con dry_run=True, el usuario ve el preview, y luego puede confirmar con dry_run=False.

Resources avanzados con @mcp.resource()

Resources estáticos vs dinámicos

Resource estático — el URI es fijo:

@mcp.resource("config://app/version")
async def app_version() -> str:
    """Versión actual de la aplicación."""
    return "2.5.1"

Resource dinámico con template — el URI tiene parámetros:

@mcp.resource("users://{user_id}/profile")
async def user_profile(user_id: str) -> str:
    """Perfil de un usuario específico."""
    users_db = {
        "1": {"name": "Ann", "role": "admin", "email": "ann@example.com"},
        "2": {"name": "Carl", "role": "developer", "email": "carl@example.com"},
    }
    user = users_db.get(user_id)
    if not user:
        return json.dumps({"error": f"Usuario {user_id} no encontrado"})
    return json.dumps(user, indent=2)

El {user_id} en el URI es un template. Cuando un cliente pide users://42/profile, el SDK extrae user_id="42" y lo pasa a tu función.

Resource que retorna JSON estructurado

@mcp.resource("metrics://server/health")
async def server_health() -> str:
    """Métricas de salud del servidor."""
    import psutil

    health = {
        "status": "healthy",
        "cpu_percent": psutil.cpu_percent(),
        "memory": {
            "total_gb": round(psutil.virtual_memory().total / (1024**3), 2),
            "used_percent": psutil.virtual_memory().percent,
        },
        "disk": {
            "total_gb": round(psutil.disk_usage("/").total / (1024**3), 2),
            "used_percent": psutil.disk_usage("/").percent,
        },
    }
    return json.dumps(health, indent=2)

Comparación: Resources en Python vs TypeScript

TypeScript:

server.resource(
  "user-profile",
  "users://{user_id}/profile",
  { description: "Perfil de un usuario", mimeType: "application/json" },
  async (uri) => {
    const userId = uri.pathname.split("/")[1];
    const user = await getUserById(userId);
    return {
      contents: [{
        uri: uri.href,
        mimeType: "application/json",
        text: JSON.stringify(user),
      }],
    };
  }
);

Python:

@mcp.resource("users://{user_id}/profile")
async def user_profile(user_id: str) -> str:
    """Perfil de un usuario."""
    user = await get_user_by_id(user_id)
    return json.dumps(user, indent=2)

Diferencias clave:

• Python extrae los parámetros del URI template automáticamente como argumentos de la función
• No necesitas parsear el URI manualmente
• El retorno es un string simple — el SDK lo envuelve en el formato MCP
• La descripción viene del docstring

Patrones de diseño para tools

Patrón 1: CRUD completo

from pydantic import BaseModel, Field
from datetime import datetime

mcp = FastMCP("todo-crud")

todos: list[dict] = []
next_id = 1


class TodoCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    description: str = Field(default="")
    priority: int = Field(default=3, ge=1, le=5)


class TodoUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    description: str | None = None
    priority: int | None = Field(default=None, ge=1, le=5)
    completed: bool | None = None


@mcp.tool()
async def create_todo(input: TodoCreate) -> str:
    """Crea un nuevo TODO."""
    global next_id
    todo = {
        "id": next_id,
        "title": input.title,
        "description": input.description,
        "priority": input.priority,
        "completed": False,
        "created_at": datetime.now().isoformat(),
    }
    next_id += 1
    todos.append(todo)
    return json.dumps({"message": "TODO creado", "todo": todo}, indent=2)


@mcp.tool()
async def get_todo(todo_id: int) -> str:
    """Obtiene un TODO por su ID."""
    for todo in todos:
        if todo["id"] == todo_id:
            return json.dumps(todo, indent=2)
    return json.dumps({"error": f"TODO con ID {todo_id} no encontrado"})


@mcp.tool()
async def update_todo(todo_id: int, updates: TodoUpdate) -> str:
    """Actualiza un TODO existente."""
    for todo in todos:
        if todo["id"] == todo_id:
            update_data = updates.model_dump(exclude_none=True)
            todo.update(update_data)
            todo["updated_at"] = datetime.now().isoformat()
            return json.dumps({"message": "TODO actualizado", "todo": todo}, indent=2)
    return json.dumps({"error": f"TODO con ID {todo_id} no encontrado"})


@mcp.tool()
async def delete_todo(todo_id: int) -> str:
    """Elimina un TODO por su ID."""
    global todos
    original_len = len(todos)
    todos = [t for t in todos if t["id"] != todo_id]
    if len(todos) < original_len:
        return f"TODO con ID {todo_id} eliminado."
    return f"Error: TODO con ID {todo_id} no encontrado."


@mcp.tool()
async def list_todos(
    status: str = "all",
    sort_by: str = "created_at",
) -> str:
    """Lista TODOs con filtros opcionales.

    status: 'all', 'completed', o 'pending'.
    sort_by: 'created_at', 'priority', o 'title'.
    """
    filtered = todos
    if status == "completed":
        filtered = [t for t in todos if t["completed"]]
    elif status == "pending":
        filtered = [t for t in todos if not t["completed"]]

    if sort_by in ("priority", "title", "created_at"):
        filtered = sorted(filtered, key=lambda t: t.get(sort_by, ""))

    return json.dumps({"total": len(filtered), "todos": filtered}, indent=2)

Patrón 2: Tool que envuelve una API externa

import httpx


@mcp.tool()
async def get_weather(city: str, units: str = "metric") -> str:
    """Obtiene el clima actual de una ciudad.

    units: 'metric' (Celsius) o 'imperial' (Fahrenheit).
    """
    api_key = os.environ.get("OPENWEATHER_API_KEY")
    if not api_key:
        return "Error: OPENWEATHER_API_KEY no configurada. Establece la variable de entorno."

    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                "https://api.openweathermap.org/data/2.5/weather",
                params={"q": city, "appid": api_key, "units": units, "lang": "es"},
                timeout=10,
            )
            response.raise_for_status()
            data = response.json()

        temp = data["main"]["temp"]
        feels_like = data["main"]["feels_like"]
        description = data["weather"][0]["description"]
        humidity = data["main"]["humidity"]
        unit_symbol = "°C" if units == "metric" else "°F"

        return (
            f"Clima en {city}:\n"
            f"  Temperatura: {temp}{unit_symbol} (sensación: {feels_like}{unit_symbol})\n"
            f"  Condición: {description}\n"
            f"  Humedad: {humidity}%"
        )
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            return f"Error: ciudad '{city}' no encontrada."
        return f"Error de API: {e.response.status_code}"
    except httpx.RequestError as e:
        return f"Error de conexión: {e}"

Patrón 3: Tool + Resource complementarios

import json

project_files: dict[str, str] = {}


@mcp.resource("project://files/list")
async def list_project_files() -> str:
    """Lista de archivos en el proyecto."""
    return json.dumps(list(project_files.keys()), indent=2)


@mcp.resource("project://files/{filename}")
async def read_project_file(filename: str) -> str:
    """Contenido de un archivo del proyecto."""
    content = project_files.get(filename)
    if content is None:
        return f"Error: archivo '{filename}' no encontrado."
    return content


@mcp.tool()
async def write_project_file(filename: str, content: str) -> str:
    """Escribe un archivo en el proyecto."""
    project_files[filename] = content
    return f"Archivo '{filename}' escrito ({len(content)} caracteres)."


@mcp.tool()
async def analyze_project() -> str:
    """Analiza todos los archivos del proyecto.

    Retorna estadísticas de cada archivo.
    """
    if not project_files:
        return "No hay archivos en el proyecto."

    stats = []
    for name, content in project_files.items():
        lines = content.count("\n") + 1
        words = len(content.split())
        stats.append({"file": name, "lines": lines, "words": words, "chars": len(content)})

    return json.dumps({"total_files": len(stats), "files": stats}, indent=2)

El resource project://files/list permite al modelo ver qué archivos hay. El resource project://files/{filename} permite leer un archivo específico. El tool write_project_file permite crear archivos. Y analyze_project actúa sobre los datos. Resource para leer, tool para actuar.

Decoradores vs Clases: la diferencia fundamental

El approach de TypeScript: explícito y verboso

server.tool(
  "analyze_text",                    // nombre: explícito
  "Analiza un texto y retorna estadísticas",  // descripción: explícita
  {                                   // schema: explícito con Zod
    text: z.string().min(1),
    include_sentiment: z.boolean().default(false),
  },
  async ({ text, include_sentiment }) => {  // handler: separado
    const words = text.split(/\s+/).length;
    const sentences = text.split(/[.!?]+/).filter(Boolean).length;

    let result = `Palabras: ${words}, Oraciones: ${sentences}`;
    if (include_sentiment) {
      result += `, Sentimiento: neutral`;
    }

    return { content: [{ type: "text", text: result }] };
  }
);

Cuatro piezas separadas: nombre, descripción, schema, handler.

El approach de Python: inferido y conciso

@mcp.tool()
async def analyze_text(text: str, include_sentiment: bool = False) -> str:
    """Analiza un texto y retorna estadísticas."""
    words = len(text.split())
    sentences = len([s for s in text.split(".") if s.strip()])

    result = f"Palabras: {words}, Oraciones: {sentences}"
    if include_sentiment:
        result += ", Sentimiento: neutral"
    return result

Una sola pieza: la función. Todo lo demás se infiere.

Ninguno es "mejor" — son filosofías diferentes. TypeScript es explícito: ves exactamente qué se registra. Python es convencional: si sigues las convenciones (docstrings, type hints), el framework hace el trabajo.

Ejercicios

Ejercicio 1: Tool con Pydantic model anidado (Medio)

Crea un tool create_event que reciba un modelo Pydantic con:

• title (str, requerido, 1-100 chars)
• date (str, requerido, formato ISO)
• location con name (str) y address (str, opcional)
• attendees (lista de strings, máximo 50)

from pydantic import BaseModel, Field
from mcp.server.fastmcp import FastMCP
import json

mcp = FastMCP("event-manager")

events: list[dict] = []


class Location(BaseModel):
    name: str = Field(description="Nombre del lugar")
    address: str | None = Field(default=None, description="Dirección completa")


class CreateEventInput(BaseModel):
    title: str = Field(min_length=1, max_length=100, description="Título del evento")
    date: str = Field(description="Fecha en formato ISO (YYYY-MM-DD)")
    location: Location = Field(description="Ubicación del evento")
    attendees: list[str] = Field(
        default_factory=list, max_length=50, description="Lista de asistentes"
    )


@mcp.tool()
async def create_event(input: CreateEventInput) -> str:
    """Crea un nuevo evento con ubicación y asistentes."""
    event = {
        "id": len(events) + 1,
        "title": input.title,
        "date": input.date,
        "location": input.location.model_dump(),
        "attendees": input.attendees,
        "attendee_count": len(input.attendees),
    }
    events.append(event)
    return json.dumps({"message": "Evento creado", "event": event}, indent=2, ensure_ascii=False)


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

Ejercicio 2: Resource con template URI dinámico (Medio)

Crea un MCP server con:

• Resource inventory://products/{category} que retorne productos filtrados por categoría
• Resource inventory://products/{category}/{product_id} que retorne un producto específico
• Tool add_product para agregar productos al inventario

from mcp.server.fastmcp import FastMCP
import json

mcp = FastMCP("inventory")

products: list[dict] = [
    {"id": 1, "name": "Laptop Pro", "category": "electronics", "price": 1299.99},
    {"id": 2, "name": "Mechanical Keyboard", "category": "electronics", "price": 89.99},
    {"id": 3, "name": "Python Cookbook", "category": "books", "price": 45.00},
    {"id": 4, "name": "Standing Desk", "category": "furniture", "price": 599.99},
]


@mcp.resource("inventory://products/{category}")
async def products_by_category(category: str) -> str:
    """Productos filtrados por categoría."""
    filtered = [p for p in products if p["category"] == category]
    return json.dumps(
        {"category": category, "count": len(filtered), "products": filtered},
        indent=2,
    )


@mcp.resource("inventory://products/{category}/{product_id}")
async def product_detail(category: str, product_id: str) -> str:
    """Detalle de un producto específico."""
    for p in products:
        if p["category"] == category and str(p["id"]) == product_id:
            return json.dumps(p, indent=2)
    return json.dumps({"error": f"Producto {product_id} no encontrado en '{category}'"})


@mcp.tool()
async def add_product(name: str, category: str, price: float) -> str:
    """Agrega un producto al inventario."""
    product = {
        "id": max((p["id"] for p in products), default=0) + 1,
        "name": name,
        "category": category,
        "price": price,
    }
    products.append(product)
    return json.dumps({"message": "Producto agregado", "product": product}, indent=2)


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

Ejercicio 3: Convertir tools de TypeScript a Python (Medio)

Convierte el siguiente MCP server de TypeScript a Python idiomático:

const server = new McpServer({ name: "file-stats", version: "1.0.0" });

server.tool("count_lines", "Cuenta líneas de un archivo", {
  filepath: z.string().describe("Ruta al archivo"),
  skip_empty: z.boolean().default(false).describe("Ignorar líneas vacías"),
}, async ({ filepath, skip_empty }) => {
  const content = await fs.readFile(filepath, "utf-8");
  let lines = content.split("\n");
  if (skip_empty) {
    lines = lines.filter(line => line.trim().length > 0);
  }
  return {
    content: [{
      type: "text",
      text: JSON.stringify({ filepath, total_lines: lines.length, skip_empty }),
    }],
  };
});

server.tool("file_info", "Información de un archivo", {
  filepath: z.string(),
}, async ({ filepath }) => {
  const stats = await fs.stat(filepath);
  return {
    content: [{
      type: "text",
      text: JSON.stringify({
        size_bytes: stats.size,
        created: stats.birthtime.toISOString(),
        modified: stats.mtime.toISOString(),
        is_directory: stats.isDirectory(),
      }),
    }],
  };
});

import os
import json
from datetime import datetime
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("file-stats")


@mcp.tool()
async def count_lines(filepath: str, skip_empty: bool = False) -> str:
    """Cuenta líneas de un archivo.

    Opcionalmente ignora líneas vacías.
    """
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            lines = f.readlines()

        if skip_empty:
            lines = [line for line in lines if line.strip()]

        return json.dumps(
            {"filepath": filepath, "total_lines": len(lines), "skip_empty": skip_empty},
            indent=2,
        )
    except FileNotFoundError:
        return json.dumps({"error": f"Archivo no encontrado: {filepath}"})
    except PermissionError:
        return json.dumps({"error": f"Sin permisos para leer: {filepath}"})


@mcp.tool()
async def file_info(filepath: str) -> str:
    """Información de un archivo: tamaño, fechas, tipo."""
    try:
        stats = os.stat(filepath)
        return json.dumps(
            {
                "size_bytes": stats.st_size,
                "created": datetime.fromtimestamp(stats.st_ctime).isoformat(),
                "modified": datetime.fromtimestamp(stats.st_mtime).isoformat(),
                "is_directory": os.path.isdir(filepath),
            },
            indent=2,
        )
    except FileNotFoundError:
        return json.dumps({"error": f"Archivo no encontrado: {filepath}"})


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

Ejercicio 4: Tool con enum y validación estricta (Difícil)

Crea un tool process_data que:

• Reciba un DataProcessInput con un campo format que sea un Enum (csv, json, xml)
• Reciba data como string
• Reciba options como un modelo Pydantic con skip_header (bool), delimiter (str), y encoding (str)
• Valide que delimiter solo se use cuando format es csv

from enum import Enum
from pydantic import BaseModel, Field, model_validator
from mcp.server.fastmcp import FastMCP
import json

mcp = FastMCP("data-processor")


class DataFormat(str, Enum):
    CSV = "csv"
    JSON = "json"
    XML = "xml"


class ProcessOptions(BaseModel):
    skip_header: bool = Field(default=False, description="Saltar primera línea (solo CSV)")
    delimiter: str = Field(default=",", description="Delimitador (solo CSV)")
    encoding: str = Field(default="utf-8", description="Codificación del texto")


class DataProcessInput(BaseModel):
    format: DataFormat = Field(description="Formato de los datos")
    data: str = Field(min_length=1, description="Datos a procesar")
    options: ProcessOptions = Field(default_factory=ProcessOptions)

    @model_validator(mode="after")
    def validate_options(self):
        if self.format != DataFormat.CSV and self.options.delimiter != ",":
            raise ValueError("El delimiter solo aplica para formato CSV")
        if self.format != DataFormat.CSV and self.options.skip_header:
            raise ValueError("skip_header solo aplica para formato CSV")
        return self


@mcp.tool()
async def process_data(input: DataProcessInput) -> str:
    """Procesa datos en formato CSV, JSON o XML."""
    if input.format == DataFormat.CSV:
        lines = input.data.strip().split("\n")
        if input.options.skip_header and len(lines) > 1:
            lines = lines[1:]
        rows = [line.split(input.options.delimiter) for line in lines]
        return json.dumps(
            {"format": "csv", "rows": len(rows), "columns": len(rows[0]) if rows else 0, "data": rows},
            indent=2,
        )

    elif input.format == DataFormat.JSON:
        try:
            parsed = json.loads(input.data)
            item_count = len(parsed) if isinstance(parsed, list) else 1
            return json.dumps({"format": "json", "items": item_count, "valid": True}, indent=2)
        except json.JSONDecodeError as e:
            return json.dumps({"format": "json", "valid": False, "error": str(e)})

    elif input.format == DataFormat.XML:
        tag_count = input.data.count("<") // 2
        return json.dumps({"format": "xml", "estimated_tags": tag_count}, indent=2)

    return json.dumps({"error": "Formato no soportado"})


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

Ejercicio 5: Server completo con las 3 primitivas (Difícil)

Crea un MCP server "bookmark-manager" con:

• Resources: bookmarks://all, bookmarks://category/{cat}
• Tools: add_bookmark, delete_bookmark, search_bookmarks
• Prompt: organize_bookmarks que pida al modelo organizar los bookmarks por categoría

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
from datetime import datetime
import json

mcp = FastMCP("bookmark-manager")

bookmarks: list[dict] = []
next_id = 1


class BookmarkInput(BaseModel):
    url: str = Field(description="URL del bookmark")
    title: str = Field(min_length=1, max_length=200, description="Título descriptivo")
    category: str = Field(default="uncategorized", description="Categoría")
    tags: list[str] = Field(default_factory=list, description="Tags para búsqueda")


@mcp.resource("bookmarks://all")
async def all_bookmarks() -> str:
    """Todos los bookmarks almacenados."""
    return json.dumps(
        {"total": len(bookmarks), "bookmarks": bookmarks}, indent=2, ensure_ascii=False
    )


@mcp.resource("bookmarks://category/{category}")
async def bookmarks_by_category(category: str) -> str:
    """Bookmarks filtrados por categoría."""
    filtered = [b for b in bookmarks if b["category"] == category]
    return json.dumps(
        {"category": category, "count": len(filtered), "bookmarks": filtered},
        indent=2,
        ensure_ascii=False,
    )


@mcp.tool()
async def add_bookmark(input: BookmarkInput) -> str:
    """Agrega un nuevo bookmark."""
    global next_id
    bookmark = {
        "id": next_id,
        "url": input.url,
        "title": input.title,
        "category": input.category,
        "tags": input.tags,
        "created_at": datetime.now().isoformat(),
    }
    next_id += 1
    bookmarks.append(bookmark)
    return json.dumps({"message": "Bookmark agregado", "bookmark": bookmark}, indent=2, ensure_ascii=False)


@mcp.tool()
async def delete_bookmark(bookmark_id: int) -> str:
    """Elimina un bookmark por ID."""
    global bookmarks
    before = len(bookmarks)
    bookmarks = [b for b in bookmarks if b["id"] != bookmark_id]
    if len(bookmarks) < before:
        return f"Bookmark {bookmark_id} eliminado."
    return f"Error: bookmark {bookmark_id} no encontrado."


@mcp.tool()
async def search_bookmarks(query: str) -> str:
    """Busca bookmarks por título, URL o tags."""
    query_lower = query.lower()
    results = [
        b
        for b in bookmarks
        if query_lower in b["title"].lower()
        or query_lower in b["url"].lower()
        or any(query_lower in tag.lower() for tag in b["tags"])
    ]
    return json.dumps({"query": query, "results": len(results), "bookmarks": results}, indent=2, ensure_ascii=False)


@mcp.prompt()
async def organize_bookmarks() -> str:
    """Pide al modelo organizar los bookmarks por categoría."""
    data = json.dumps(bookmarks, indent=2, ensure_ascii=False) if bookmarks else "No hay bookmarks."
    return f"""Analiza los siguientes bookmarks y sugiere una reorganización por categorías:

{data}

Por favor:
1. Agrupa los bookmarks en categorías lógicas
2. Sugiere nuevas categorías si las actuales no son descriptivas
3. Identifica bookmarks duplicados o similares
4. Recomienda tags útiles para cada bookmark
5. Usa el tool delete_bookmark y add_bookmark para implementar los cambios"""


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

Troubleshooting

"El tool no recibe los parámetros correctos"

Causa: El modelo envía parámetros que no coinciden con los type hints de tu función.

Solución:

@mcp.tool()
async def my_tool(
    name: str,           # ✅ Tipo explícito
    count: int = 5,      # ✅ Default explícito
    tags: list[str] | None = None,  # ✅ Optional con default
) -> str:
    """Descripción que le dice al modelo exactamente qué parámetros usar."""
    ...

"ValidationError de Pydantic al invocar el tool"

Causa: Los datos enviados no pasan la validación de Pydantic.

Solución:

class MyInput(BaseModel):
    value: int = Field(ge=0, le=100)

@mcp.tool()
async def my_tool(input: MyInput) -> str:
    """El rango válido es 0-100. Valores fuera de rango serán rechazados."""
    return str(input.value)

"El resource template no extrae los parámetros del URI"

Causa: Los nombres de los parámetros en el URI template no coinciden con los de la función.

Solución:

@mcp.resource("data://{item_type}/{item_id}")
async def get_item(item_type: str, item_id: str) -> str:
    ...

"TypeError: object str can't be used in 'await' expression"

Causa: Estás usando await en una función que no es async, o estás llamando una función sync con await.

Solución:

@mcp.tool()
async def my_tool(path: str) -> str:
    content = open(path).read()  # sync, sin await
    return content

Resumen

En esta cápsula aprendiste:

• Tools avanzados con múltiples parámetros, tipos complejos, y Pydantic models
• Pydantic vs Zod: ambos validan, pero Pydantic se integra con type hints de Python
• Resources dinámicos con templates de URI que extraen parámetros automáticamente
• Patrones de diseño: CRUD completo, API wrapper, dry-run, tool + resource complementarios
• Decoradores vs Clases: Python infiere nombre, descripción, y schema; TypeScript los declara explícitamente
• Error handling como texto de retorno (no excepciones) para que el modelo pueda interpretar errores

Próxima cápsula: Patrones async en MCP Python — asyncio, context managers async, generators, y por qué async importa cuando tu MCP server conecta con APIs y databases.

Recursos adicionales

1. Pydantic Field Validators — Validación avanzada con Pydantic
2. MCP Python SDK — Tools — Implementación oficial de tools
3. Python Type Hints Cheat Sheet — Referencia rápida de type hints
4. Zod Documentation — Para comparar con Pydantic
5. MCP Specification — Resources — Spec oficial de resources
6. MCP Specification — Tools — Spec oficial de tools

Siguiente cápsula: Patrones Async en MCP — asyncio en el contexto de MCP servers, conexiones a APIs, y error handling async.