Módulo 8: Proyecto — MCP Server Real

Implementación Core del MCP Server

Siguiente →

Implementación Core del MCP Server

Descripción de la cápsula

Diseño terminado. Ahora construyes. Esta cápsula contiene la implementación completa del MCP server para la Opción A (SQLite + Python): database setup, resources, tools, prompts, y error handling. Todo el código es ejecutable — puedes copiar cada bloque, y al final de la cápsula tendrás un server funcional que responde en MCP Inspector.

Si elegiste Opción B o C, usa esta implementación como referencia de patrones. La estructura es la misma: setup de data source → resources → tools → prompts → entry point. Solo cambia cómo accedes a los datos.

Paso 1: Database setup y conexión

src/database.py

import sqlite3
import os
import logging
from pathlib import Path
from contextlib import contextmanager

logger = logging.getLogger(__name__)

DB_DIR = Path(__file__).parent.parent / "data"
DB_PATH = DB_DIR / "tasks.db"


def get_db_path() -> str:
    """Retorna la ruta a la database. Crea el directorio si no existe."""
    DB_DIR.mkdir(parents=True, exist_ok=True)
    return str(DB_PATH)


@contextmanager
def get_connection(db_path: str | None = None):
    """Context manager para conexiones SQLite.

    Usa WAL mode para mejor concurrencia y configura row_factory
    para retornar diccionarios en vez de tuplas.
    """
    path = db_path or get_db_path()
    conn = sqlite3.connect(path)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA foreign_keys=ON")
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def init_database(db_path: str | None = None) -> None:
    """Crea las tablas si no existen."""
    with get_connection(db_path) as conn:
        conn.executescript("""
            CREATE TABLE IF NOT EXISTS categories (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL UNIQUE,
                description TEXT,
                color TEXT DEFAULT '#6B7280',
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            );

            CREATE TABLE IF NOT EXISTS tasks (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                description TEXT,
                status TEXT DEFAULT 'pending'
                    CHECK(status IN ('pending', 'in_progress', 'completed', 'cancelled')),
                priority TEXT DEFAULT 'medium'
                    CHECK(priority IN ('low', 'medium', 'high', 'critical')),
                category_id INTEGER REFERENCES categories(id) ON DELETE SET NULL,
                due_date TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            );

            CREATE TABLE IF NOT EXISTS tags (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL UNIQUE
            );

            CREATE TABLE IF NOT EXISTS task_tags (
                task_id INTEGER REFERENCES tasks(id) ON DELETE CASCADE,
                tag_id INTEGER REFERENCES tags(id) ON DELETE CASCADE,
                PRIMARY KEY (task_id, tag_id)
            );

            CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status);
            CREATE INDEX IF NOT EXISTS idx_tasks_priority ON tasks(priority);
            CREATE INDEX IF NOT EXISTS idx_tasks_category ON tasks(category_id);
            CREATE INDEX IF NOT EXISTS idx_tasks_due_date ON tasks(due_date);
        """)
        logger.info("Database initialized at %s", db_path or DB_PATH)


def seed_sample_data(db_path: str | None = None) -> None:
    """Inserta datos de ejemplo para demos y testing."""
    with get_connection(db_path) as conn:
        count = conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0]
        if count > 0:
            logger.info("Database already has data, skipping seed")
            return

        conn.executescript("""
            INSERT INTO categories (name, description, color) VALUES
                ('Backend', 'Backend and API tasks', '#3B82F6'),
                ('Frontend', 'UI and UX tasks', '#10B981'),
                ('DevOps', 'Infrastructure and deployment', '#F59E0B'),
                ('Docs', 'Documentation and guides', '#8B5CF6');

            INSERT INTO tags (name) VALUES
                ('bug'), ('feature'), ('refactor'), ('urgent'), ('research');

            INSERT INTO tasks (title, description, status, priority, category_id, due_date) VALUES
                ('Implement JWT authentication', 'Add login/register with JWT tokens', 'in_progress', 'high', 1, '2026-03-20'),
                ('Design landing page', 'Create mockups for the landing page', 'pending', 'medium', 2, '2026-03-25'),
                ('Set up CI/CD pipeline', 'GitHub Actions for tests and deploy', 'pending', 'high', 3, '2026-03-18'),
                ('Write API docs', 'Document all the REST endpoints', 'pending', 'medium', 4, '2026-03-22'),
                ('Fix pagination bug', 'Results repeat on page 3', 'completed', 'critical', 1, '2026-03-10'),
                ('Migrate to PostgreSQL', 'Switch from SQLite to PostgreSQL for production', 'pending', 'low', 3, '2026-04-01'),
                ('Add dark mode', 'Implement dark theme across the app', 'in_progress', 'low', 2, '2026-03-28'),
                ('Optimize N+1 queries', 'Detect and fix N+1 queries in the ORM', 'pending', 'high', 1, '2026-03-15'),
                ('Create onboarding flow', 'Interactive guide for new users', 'pending', 'medium', 2, '2026-03-30'),
                ('Set up monitoring', 'Configure alerts and dashboards', 'cancelled', 'medium', 3, NULL);

            INSERT INTO task_tags (task_id, tag_id) VALUES
                (1, 2), (1, 4),
                (2, 2),
                (3, 2),
                (4, 2),
                (5, 1), (5, 4),
                (6, 3),
                (7, 2),
                (8, 1), (8, 3),
                (9, 2),
                (10, 2);
        """)
        logger.info("Sample data seeded: 4 categories, 5 tags, 10 tasks")

Puntos clave de la implementación:

  • get_connection como context manager garantiza que la conexión se cierre siempre, incluso si hay errores
  • row_factory = sqlite3.Row permite acceder a columnas por nombre (row["title"]) en vez de por índice
  • PRAGMA journal_mode=WAL mejora la concurrencia — lecturas no bloquean escrituras
  • PRAGMA foreign_keys=ON activa los foreign key constraints (SQLite los desactiva por defecto)
  • seed_sample_data es idempotente — si ya hay datos, no inserta duplicados
  • Los índices en status, priority, category_id, y due_date aceleran los filtros más comunes

Paso 2: Pydantic Models

src/models.py

from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum


class TaskStatus(str, Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    CANCELLED = "cancelled"


class TaskPriority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


class CreateTaskInput(BaseModel):
    title: str = Field(min_length=1, max_length=200, description="Título de la tarea")
    description: Optional[str] = Field(default=None, max_length=2000, description="Descripción detallada")
    status: TaskStatus = Field(default=TaskStatus.PENDING, description="Estado inicial")
    priority: TaskPriority = Field(default=TaskPriority.MEDIUM, description="Prioridad: low, medium, high, critical")
    category_id: Optional[int] = Field(default=None, ge=1, description="ID de la categoría")
    due_date: Optional[str] = Field(default=None, description="Fecha límite YYYY-MM-DD")
    tags: Optional[list[str]] = Field(default=None, description="Tags para la tarea")


class UpdateTaskInput(BaseModel):
    task_id: int = Field(ge=1, description="ID de la tarea a actualizar")
    title: Optional[str] = Field(default=None, min_length=1, max_length=200)
    description: Optional[str] = Field(default=None, max_length=2000)
    status: Optional[TaskStatus] = Field(default=None)
    priority: Optional[TaskPriority] = Field(default=None)
    category_id: Optional[int] = Field(default=None, ge=1)
    due_date: Optional[str] = Field(default=None)


class ListTasksInput(BaseModel):
    status: Optional[TaskStatus] = Field(default=None, description="Filtrar por estado")
    priority: Optional[TaskPriority] = Field(default=None, description="Filtrar por prioridad")
    category_id: Optional[int] = Field(default=None, ge=1, description="Filtrar por categoría")
    limit: int = Field(default=20, ge=1, le=100, description="Máximo de resultados")


class SearchTasksInput(BaseModel):
    query: str = Field(min_length=1, description="Texto a buscar en tareas")
    search_in: str = Field(default="both", description="Dónde buscar: title, description, both")


class CreateCategoryInput(BaseModel):
    name: str = Field(min_length=1, max_length=50, description="Nombre de la categoría")
    description: Optional[str] = Field(default=None, max_length=200)
    color: str = Field(default="#6B7280", pattern=r"^#[0-9A-Fa-f]{6}$", description="Color hex")


class RunQueryInput(BaseModel):
    sql: str = Field(min_length=1, description="Query SQL — solo SELECT permitido")


class TaskSummaryInput(BaseModel):
    period: str = Field(description="Periodo del resumen: today, week, month")

Paso 3: Tools — Operaciones CRUD de tareas

src/tools/tasks.py

import json
import logging
from datetime import datetime
from src.database import get_connection
from src.models import CreateTaskInput, UpdateTaskInput, ListTasksInput, SearchTasksInput

logger = logging.getLogger(__name__)


def _task_to_dict(row) -> dict:
    """Convierte un Row de SQLite a diccionario serializable."""
    return {
        "id": row["id"],
        "title": row["title"],
        "description": row["description"],
        "status": row["status"],
        "priority": row["priority"],
        "category_id": row["category_id"],
        "due_date": row["due_date"],
        "created_at": row["created_at"],
        "updated_at": row["updated_at"],
    }


def _error(error_type: str, message: str) -> str:
    return json.dumps({"error": True, "error_type": error_type, "message": message}, ensure_ascii=False)


async def create_task(input: CreateTaskInput) -> str:
    """Crea una nueva tarea en la base de datos.

    Retorna la tarea creada con su ID asignado. Si se incluyen tags,
    los asocia automáticamente (crea tags nuevos si no existen).
    """
    try:
        with get_connection() as conn:
            if input.category_id:
                cat = conn.execute("SELECT id FROM categories WHERE id = ?", (input.category_id,)).fetchone()
                if not cat:
                    return _error("not_found", f"Categoría con ID {input.category_id} no encontrada")

            cursor = conn.execute(
                """INSERT INTO tasks (title, description, status, priority, category_id, due_date)
                   VALUES (?, ?, ?, ?, ?, ?)""",
                (input.title, input.description, input.status.value, input.priority.value,
                 input.category_id, input.due_date),
            )
            task_id = cursor.lastrowid

            if input.tags:
                for tag_name in input.tags:
                    conn.execute("INSERT OR IGNORE INTO tags (name) VALUES (?)", (tag_name.lower().strip(),))
                    tag = conn.execute("SELECT id FROM tags WHERE name = ?", (tag_name.lower().strip(),)).fetchone()
                    conn.execute("INSERT OR IGNORE INTO task_tags (task_id, tag_id) VALUES (?, ?)", (task_id, tag["id"]))

            task = conn.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
            result = _task_to_dict(task)
            result["tags"] = input.tags or []

        logger.info("Task created: id=%d title='%s'", task_id, input.title)
        return json.dumps({"created": True, "task": result}, indent=2, ensure_ascii=False)

    except Exception as e:
        logger.error("Error creating task: %s", e)
        return _error("database_error", f"Error al crear tarea: {e}")


async def list_tasks(input: ListTasksInput) -> str:
    """Lista tareas con filtros opcionales por estado, prioridad y categoría.

    Sin filtros retorna las tareas más recientes. Incluye el nombre de la
    categoría y los tags asociados a cada tarea.
    """
    try:
        with get_connection() as conn:
            query = "SELECT t.*, c.name as category_name FROM tasks t LEFT JOIN categories c ON t.category_id = c.id"
            conditions = []
            params = []

            if input.status:
                conditions.append("t.status = ?")
                params.append(input.status.value)
            if input.priority:
                conditions.append("t.priority = ?")
                params.append(input.priority.value)
            if input.category_id:
                conditions.append("t.category_id = ?")
                params.append(input.category_id)

            if conditions:
                query += " WHERE " + " AND ".join(conditions)

            query += " ORDER BY t.created_at DESC LIMIT ?"
            params.append(input.limit)

            rows = conn.execute(query, params).fetchall()

            tasks = []
            for row in rows:
                task = _task_to_dict(row)
                task["category_name"] = row["category_name"]

                tag_rows = conn.execute(
                    "SELECT t.name FROM tags t JOIN task_tags tt ON t.id = tt.tag_id WHERE tt.task_id = ?",
                    (row["id"],),
                ).fetchall()
                task["tags"] = [t["name"] for t in tag_rows]
                tasks.append(task)

        return json.dumps({"total": len(tasks), "tasks": tasks}, indent=2, ensure_ascii=False)

    except Exception as e:
        logger.error("Error listing tasks: %s", e)
        return _error("database_error", f"Error al listar tareas: {e}")


async def update_task(input: UpdateTaskInput) -> str:
    """Actualiza campos de una tarea existente.

    Solo modifica los campos que se envían (los demás quedan igual).
    Retorna la tarea actualizada completa.
    """
    try:
        with get_connection() as conn:
            existing = conn.execute("SELECT * FROM tasks WHERE id = ?", (input.task_id,)).fetchone()
            if not existing:
                return _error("not_found", f"Tarea con ID {input.task_id} no encontrada")

            updates = []
            params = []

            if input.title is not None:
                updates.append("title = ?")
                params.append(input.title)
            if input.description is not None:
                updates.append("description = ?")
                params.append(input.description)
            if input.status is not None:
                updates.append("status = ?")
                params.append(input.status.value)
            if input.priority is not None:
                updates.append("priority = ?")
                params.append(input.priority.value)
            if input.category_id is not None:
                updates.append("category_id = ?")
                params.append(input.category_id)
            if input.due_date is not None:
                updates.append("due_date = ?")
                params.append(input.due_date)

            if not updates:
                return _error("validation", "No se proporcionaron campos para actualizar")

            updates.append("updated_at = ?")
            params.append(datetime.now().isoformat())
            params.append(input.task_id)

            conn.execute(f"UPDATE tasks SET {', '.join(updates)} WHERE id = ?", params)

            updated = conn.execute("SELECT * FROM tasks WHERE id = ?", (input.task_id,)).fetchone()

        logger.info("Task updated: id=%d", input.task_id)
        return json.dumps({"updated": True, "task": _task_to_dict(updated)}, indent=2, ensure_ascii=False)

    except Exception as e:
        logger.error("Error updating task: %s", e)
        return _error("database_error", f"Error al actualizar tarea: {e}")


async def delete_task(task_id: int) -> str:
    """Elimina una tarea de la base de datos.

    También elimina las asociaciones con tags (CASCADE). Retorna confirmación
    con el ID de la tarea eliminada.
    """
    try:
        with get_connection() as conn:
            existing = conn.execute("SELECT id, title FROM tasks WHERE id = ?", (task_id,)).fetchone()
            if not existing:
                return _error("not_found", f"Tarea con ID {task_id} no encontrada")

            conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,))

        logger.info("Task deleted: id=%d title='%s'", task_id, existing["title"])
        return json.dumps({"deleted": True, "task_id": task_id, "title": existing["title"]}, ensure_ascii=False)

    except Exception as e:
        logger.error("Error deleting task: %s", e)
        return _error("database_error", f"Error al eliminar tarea: {e}")


async def search_tasks(input: SearchTasksInput) -> str:
    """Busca tareas por texto en título, descripción o ambos.

    La búsqueda no distingue mayúsculas/minúsculas. Retorna tareas
    que contengan el texto buscado en los campos seleccionados.
    """
    try:
        with get_connection() as conn:
            search_term = f"%{input.query}%"

            if input.search_in == "title":
                where = "t.title LIKE ?"
                params = [search_term]
            elif input.search_in == "description":
                where = "t.description LIKE ?"
                params = [search_term]
            else:
                where = "(t.title LIKE ? OR t.description LIKE ?)"
                params = [search_term, search_term]

            rows = conn.execute(
                f"""SELECT t.*, c.name as category_name
                    FROM tasks t LEFT JOIN categories c ON t.category_id = c.id
                    WHERE {where} ORDER BY t.created_at DESC LIMIT 20""",
                params,
            ).fetchall()

            tasks = []
            for row in rows:
                task = _task_to_dict(row)
                task["category_name"] = row["category_name"]
                tasks.append(task)

        return json.dumps({
            "query": input.query,
            "search_in": input.search_in,
            "results": len(tasks),
            "tasks": tasks,
        }, indent=2, ensure_ascii=False)

    except Exception as e:
        logger.error("Error searching tasks: %s", e)
        return _error("database_error", f"Error al buscar tareas: {e}")

Paso 4: Tools — Categorías y queries

src/tools/categories.py

import json
import logging
from src.database import get_connection
from src.models import CreateCategoryInput

logger = logging.getLogger(__name__)


def _error(error_type: str, message: str) -> str:
    return json.dumps({"error": True, "error_type": error_type, "message": message}, ensure_ascii=False)


async def create_category(input: CreateCategoryInput) -> str:
    """Crea una nueva categoría para organizar tareas.

    Las categorías agrupan tareas por área (Backend, Frontend, DevOps, etc.).
    Cada categoría tiene un nombre único, descripción opcional, y color hex.
    """
    try:
        with get_connection() as conn:
            existing = conn.execute("SELECT id FROM categories WHERE name = ?", (input.name,)).fetchone()
            if existing:
                return _error("duplicate", f"Ya existe una categoría con nombre '{input.name}'")

            cursor = conn.execute(
                "INSERT INTO categories (name, description, color) VALUES (?, ?, ?)",
                (input.name, input.description, input.color),
            )

            category = conn.execute("SELECT * FROM categories WHERE id = ?", (cursor.lastrowid,)).fetchone()

        logger.info("Category created: id=%d name='%s'", cursor.lastrowid, input.name)
        return json.dumps({
            "created": True,
            "category": {
                "id": category["id"],
                "name": category["name"],
                "description": category["description"],
                "color": category["color"],
                "created_at": category["created_at"],
            },
        }, indent=2, ensure_ascii=False)

    except Exception as e:
        logger.error("Error creating category: %s", e)
        return _error("database_error", f"Error al crear categoría: {e}")

src/tools/queries.py

import json
import logging
import re
from datetime import datetime, timedelta
from src.database import get_connection
from src.models import RunQueryInput, TaskSummaryInput

logger = logging.getLogger(__name__)


def _error(error_type: str, message: str) -> str:
    return json.dumps({"error": True, "error_type": error_type, "message": message}, ensure_ascii=False)


FORBIDDEN_KEYWORDS = re.compile(
    r"\b(INSERT|UPDATE|DELETE|DROP|ALTER|CREATE|TRUNCATE|REPLACE|ATTACH|DETACH)\b",
    re.IGNORECASE,
)


async def run_query(input: RunQueryInput) -> str:
    """Ejecuta una query SQL de solo lectura contra la base de datos.

    Solo queries SELECT están permitidas. Queries que modifiquen datos
    (INSERT, UPDATE, DELETE, DROP) serán rechazadas. Útil para consultas
    ad-hoc que no están cubiertas por los otros tools.
    """
    if FORBIDDEN_KEYWORDS.search(input.sql):
        return _error("permission", "Solo queries SELECT están permitidas. No se pueden ejecutar queries que modifiquen datos.")

    try:
        with get_connection() as conn:
            cursor = conn.execute(input.sql)
            columns = [desc[0] for desc in cursor.description] if cursor.description else []
            rows = cursor.fetchall()

            results = [dict(zip(columns, row)) for row in rows]

        return json.dumps({
            "query": input.sql,
            "columns": columns,
            "row_count": len(results),
            "results": results,
        }, indent=2, ensure_ascii=False)

    except Exception as e:
        logger.error("Error executing query: %s", e)
        return _error("query_error", f"Error en la query: {e}")


async def get_task_summary(input: TaskSummaryInput) -> str:
    """Genera un resumen de tareas para un periodo específico.

    Periodos disponibles: 'today' (hoy), 'week' (últimos 7 días),
    'month' (últimos 30 días). Incluye conteos por estado y prioridad.
    """
    period_map = {
        "today": 0,
        "week": 7,
        "month": 30,
    }

    if input.period not in period_map:
        return _error("validation", f"Periodo '{input.period}' no válido. Usa: today, week, month")

    days_back = period_map[input.period]
    cutoff = (datetime.now() - timedelta(days=days_back)).isoformat()

    try:
        with get_connection() as conn:
            if days_back == 0:
                date_filter = "DATE(t.created_at) = DATE('now')"
            else:
                date_filter = f"t.created_at >= '{cutoff}'"

            total = conn.execute(f"SELECT COUNT(*) FROM tasks t WHERE {date_filter}").fetchone()[0]

            by_status = {}
            for row in conn.execute(f"SELECT status, COUNT(*) as count FROM tasks t WHERE {date_filter} GROUP BY status"):
                by_status[row["status"]] = row["count"]

            by_priority = {}
            for row in conn.execute(f"SELECT priority, COUNT(*) as count FROM tasks t WHERE {date_filter} GROUP BY priority"):
                by_priority[row["priority"]] = row["count"]

            completed = conn.execute(
                f"""SELECT title, updated_at FROM tasks t
                    WHERE status = 'completed' AND {date_filter}
                    ORDER BY updated_at DESC LIMIT 10"""
            ).fetchall()

            overdue = conn.execute(
                "SELECT COUNT(*) FROM tasks WHERE due_date < DATE('now') AND status NOT IN ('completed', 'cancelled')"
            ).fetchone()[0]

        return json.dumps({
            "period": input.period,
            "total_tasks": total,
            "by_status": by_status,
            "by_priority": by_priority,
            "completed_recently": [{"title": r["title"], "completed_at": r["updated_at"]} for r in completed],
            "overdue_count": overdue,
        }, indent=2, ensure_ascii=False)

    except Exception as e:
        logger.error("Error generating summary: %s", e)
        return _error("database_error", f"Error al generar resumen: {e}")

Paso 5: Resources

src/resources/database.py

import json
import logging
from src.database import get_connection

logger = logging.getLogger(__name__)


async def get_tables() -> str:
    """Lista todas las tablas de la base de datos con su conteo de registros."""
    try:
        with get_connection() as conn:
            tables = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
            ).fetchall()

            result = []
            for table in tables:
                count = conn.execute(f"SELECT COUNT(*) FROM {table['name']}").fetchone()[0]
                result.append({"name": table["name"], "row_count": count})

        return json.dumps({"tables": result, "total_tables": len(result)}, indent=2)

    except Exception as e:
        logger.error("Error listing tables: %s", e)
        return json.dumps({"error": f"Error al listar tablas: {e}"})


async def get_table_schema(table_name: str) -> str:
    """Retorna el schema de una tabla: columnas, tipos, constraints."""
    try:
        with get_connection() as conn:
            safe_tables = [r["name"] for r in conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table'"
            ).fetchall()]

            if table_name not in safe_tables:
                return json.dumps({"error": f"Tabla '{table_name}' no encontrada. Tablas disponibles: {safe_tables}"})

            columns = conn.execute(f"PRAGMA table_info({table_name})").fetchall()
            fkeys = conn.execute(f"PRAGMA foreign_key_list({table_name})").fetchall()
            indexes = conn.execute(f"PRAGMA index_list({table_name})").fetchall()

            schema = {
                "table": table_name,
                "columns": [
                    {
                        "name": col["name"],
                        "type": col["type"],
                        "nullable": not col["notnull"],
                        "default": col["dflt_value"],
                        "primary_key": bool(col["pk"]),
                    }
                    for col in columns
                ],
                "foreign_keys": [
                    {"column": fk["from"], "references": f"{fk['table']}({fk['to']})"}
                    for fk in fkeys
                ],
                "indexes": [
                    {"name": idx["name"], "unique": bool(idx["unique"])}
                    for idx in indexes
                ],
            }

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

    except Exception as e:
        logger.error("Error getting schema for %s: %s", table_name, e)
        return json.dumps({"error": f"Error al obtener schema: {e}"})


async def get_stats() -> str:
    """Estadísticas generales de la base de datos de tareas."""
    try:
        with get_connection() as conn:
            total_tasks = conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0]

            by_status = {}
            for row in conn.execute("SELECT status, COUNT(*) as c FROM tasks GROUP BY status"):
                by_status[row["status"]] = row["c"]

            by_priority = {}
            for row in conn.execute("SELECT priority, COUNT(*) as c FROM tasks GROUP BY priority"):
                by_priority[row["priority"]] = row["c"]

            by_category = {}
            for row in conn.execute(
                "SELECT COALESCE(c.name, 'Sin categoría') as name, COUNT(*) as c "
                "FROM tasks t LEFT JOIN categories c ON t.category_id = c.id GROUP BY c.name"
            ):
                by_category[row["name"]] = row["c"]

            overdue = conn.execute(
                "SELECT COUNT(*) FROM tasks WHERE due_date < DATE('now') AND status NOT IN ('completed', 'cancelled')"
            ).fetchone()[0]

            categories_count = conn.execute("SELECT COUNT(*) FROM categories").fetchone()[0]
            tags_count = conn.execute("SELECT COUNT(*) FROM tags").fetchone()[0]

        return json.dumps({
            "total_tasks": total_tasks,
            "by_status": by_status,
            "by_priority": by_priority,
            "by_category": by_category,
            "overdue_tasks": overdue,
            "total_categories": categories_count,
            "total_tags": tags_count,
        }, indent=2, ensure_ascii=False)

    except Exception as e:
        logger.error("Error getting stats: %s", e)
        return json.dumps({"error": f"Error al obtener estadísticas: {e}"})


async def get_overdue_tasks() -> str:
    """Tareas vencidas: due_date pasado y estado no completado/cancelado."""
    try:
        with get_connection() as conn:
            rows = conn.execute(
                """SELECT t.*, c.name as category_name
                   FROM tasks t LEFT JOIN categories c ON t.category_id = c.id
                   WHERE t.due_date < DATE('now') AND t.status NOT IN ('completed', 'cancelled')
                   ORDER BY t.due_date ASC"""
            ).fetchall()

            tasks = []
            for row in rows:
                tasks.append({
                    "id": row["id"],
                    "title": row["title"],
                    "status": row["status"],
                    "priority": row["priority"],
                    "category": row["category_name"],
                    "due_date": row["due_date"],
                    "days_overdue": (
                        __import__("datetime").datetime.now()
                        - __import__("datetime").datetime.fromisoformat(row["due_date"])
                    ).days,
                })

        return json.dumps({"overdue_count": len(tasks), "tasks": tasks}, indent=2, ensure_ascii=False)

    except Exception as e:
        logger.error("Error getting overdue tasks: %s", e)
        return json.dumps({"error": f"Error al obtener tareas vencidas: {e}"})


async def get_categories() -> str:
    """Lista todas las categorías con conteo de tareas por categoría."""
    try:
        with get_connection() as conn:
            rows = conn.execute(
                """SELECT c.*, COUNT(t.id) as task_count
                   FROM categories c LEFT JOIN tasks t ON c.id = t.category_id
                   GROUP BY c.id ORDER BY c.name"""
            ).fetchall()

            categories = [
                {
                    "id": row["id"],
                    "name": row["name"],
                    "description": row["description"],
                    "color": row["color"],
                    "task_count": row["task_count"],
                }
                for row in rows
            ]

        return json.dumps({"categories": categories, "total": len(categories)}, indent=2, ensure_ascii=False)

    except Exception as e:
        logger.error("Error listing categories: %s", e)
        return json.dumps({"error": f"Error al listar categorías: {e}"})

Paso 6: Server principal — todo conectado

src/server.py

import logging
from mcp.server.fastmcp import FastMCP
from src.database import init_database, seed_sample_data
from src.tools.tasks import create_task, list_tasks, update_task, delete_task, search_tasks
from src.tools.categories import create_category
from src.tools.queries import run_query, get_task_summary
from src.resources.database import (
    get_tables, get_table_schema, get_stats,
    get_overdue_tasks, get_categories,
)

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(levelname)s: %(message)s")
logger = logging.getLogger(__name__)

init_database()
seed_sample_data()

mcp = FastMCP(
    "task-manager",
    version="1.0.0",
    instructions=(
        "MCP server para gestionar tareas con SQLite. Puedes crear, listar, "
        "actualizar, buscar y eliminar tareas. Las tareas se organizan por categorías "
        "y tags, y tienen estados (pending, in_progress, completed, cancelled) y "
        "prioridades (low, medium, high, critical). Usa los resources para consultar "
        "la estructura de la base de datos y estadísticas generales."
    ),
)

# --- Tools ---
mcp.tool()(create_task)
mcp.tool()(list_tasks)
mcp.tool()(update_task)
mcp.tool()(delete_task)
mcp.tool()(search_tasks)
mcp.tool()(create_category)
mcp.tool()(run_query)
mcp.tool()(get_task_summary)

# --- Resources ---
mcp.resource("taskdb://tables")(get_tables)
mcp.resource("taskdb://table/{table_name}/schema")(get_table_schema)
mcp.resource("taskdb://stats")(get_stats)
mcp.resource("taskdb://tasks/overdue")(get_overdue_tasks)
mcp.resource("taskdb://categories")(get_categories)


# --- Prompts ---
@mcp.prompt()
async def analyze_table(table_name: str) -> str:
    """Analiza la estructura y datos de una tabla, sugiere mejoras."""
    return f"""Analiza la tabla '{table_name}' en la base de datos del task manager.

Por favor:
1. Lee el resource taskdb://table/{table_name}/schema para ver la estructura
2. Usa el tool run_query con: SELECT COUNT(*) FROM {table_name}
3. Usa el tool run_query con: SELECT * FROM {table_name} LIMIT 5
4. Lee el resource taskdb://stats para contexto general

Con esa información, genera un análisis que incluya:
- Estructura de la tabla (columnas, tipos, constraints)
- Volumen de datos actual
- Distribución de valores en columnas clave
- Posibles mejoras (índices, constraints adicionales, normalización)
- 3 queries útiles para explorar esta tabla"""


@mcp.prompt()
async def weekly_report(week_start: str = "") -> str:
    """Genera un reporte semanal de productividad."""
    context = f"para la semana del {week_start}" if week_start else "para esta semana"
    return f"""Genera un reporte de productividad {context}.

Por favor:
1. Usa el tool get_task_summary con period='week'
2. Usa el tool list_tasks con status='completed' para tareas terminadas
3. Lee el resource taskdb://tasks/overdue para tareas vencidas
4. Lee el resource taskdb://stats para contexto general

Genera un reporte que incluya:
- Resumen ejecutivo (completadas vs pendientes vs vencidas)
- Tareas completadas (listar con categoría y prioridad)
- Tareas pendientes prioritarias (high y critical)
- Tareas vencidas que requieren atención
- Tasa de completación del periodo
- Recomendaciones para la próxima semana"""


@mcp.prompt()
async def optimize_query(sql_query: str) -> str:
    """Analiza una query SQL y sugiere optimizaciones."""
    return f"""Analiza y optimiza la siguiente query SQL:

```sql
{sql_query}

Por favor:

  1. Lee los resources taskdb://tables y taskdb://table/tasks/schema para entender la estructura
  2. Ejecuta la query original con el tool run_query para ver el resultado
  3. Si es posible, ejecuta EXPLAIN QUERY PLAN con run_query

Genera un análisis que incluya:

  • Qué hace la query
  • Posibles problemas de rendimiento
  • Query optimizada (si aplica)
  • Índices recomendados
  • Alternativas más eficientes"""

if name == "main": mcp.run()


---

## Paso 7: Verificar en MCP Inspector

Con todo el código en su lugar, verifica que funciona:

```bash
cd task-manager-mcp
source .venv/bin/activate
PYTHONPATH=. mcp dev src/server.py

MCP Inspector debería mostrar:

Tools (8): create_task, list_tasks, update_task, delete_task, search_tasks, create_category, run_query, get_task_summary

Resources (5): taskdb://tables, taskdb://table/{table_name}/schema, taskdb://stats, taskdb://tasks/overdue, taskdb://categories

Prompts (3): analyze_table, weekly_report, optimize_query

Verificaciones rápidas en Inspector

  1. Resource taskdb://tables — Deberías ver 4 tablas con sus conteos
  2. Resource taskdb://stats — Estadísticas con 10 tareas (seed data)
  3. Tool list_tasks con {} (sin filtros) — Las 10 tareas del seed data
  4. Tool create_task con {"title": "Test from Inspector"} — Crear una tarea nueva
  5. Tool search_tasks con {"query": "bug"} — Encontrar la tarea del bug en paginación

Si las 5 verificaciones pasan, tu server está listo. Cápsula 04 agrega tests automatizados y documentación.

Notas para Opciones B y C

Opción B: File System + TypeScript

Los patrones son los mismos. En vez de get_connection() usas fs.promises. En vez de SQL usas operaciones de file system. El entry point registra tools con server.tool() y resources con server.resource().

El equivalente de database.py sería un módulo que configura el directorio raíz del proyecto a analizar y valida que existe.

Opción C: API Externa

En vez de get_connection() usas httpx.AsyncClient(). En vez de queries SQL haces requests HTTP. El error handling incluye timeouts, rate limits, y autenticación.

El equivalente de seed_sample_data() no aplica — los datos ya existen en la API externa.

Troubleshooting

"ModuleNotFoundError: No module named 'src'"

Ejecuta con PYTHONPATH=. para que Python encuentre el paquete src:

PYTHONPATH=. python src/server.py
PYTHONPATH=. mcp dev src/server.py

"sqlite3.OperationalError: database is locked"

Otra instancia del server tiene la database abierta. Cierra otros procesos que usen tasks.db, o reinicia el server.

"El seed data se duplica cada vez que inicio el server"

seed_sample_data() verifica si ya hay datos antes de insertar. Si ves duplicados, probablemente estás llamando init_database() con una ruta diferente cada vez. Verifica que DB_PATH es consistente.

"Los tags no se asocian a la tarea"

Verifica que PRAGMA foreign_keys=ON está activo. Sin esto, SQLite ignora las foreign keys y los INSERT en task_tags pueden fallar silenciosamente.

"MCP Inspector no muestra todos los tools"

Verifica que no hay errores de import. Ejecuta PYTHONPATH=. python -c "from src.server import mcp" para verificar que todo importa correctamente.

"run_query permite queries destructivas"

El regex FORBIDDEN_KEYWORDS debería bloquear INSERT, UPDATE, DELETE, DROP, etc. Si un keyword no está en la lista, agrégalo. Recuerda que la protección es básica — para un server en producción real, necesitarías sanitización más robusta.

Resumen

  • Implementaste el database layer completo: conexión, schema creation, y queries parametrizadas
  • Los Tools manejan CRUD completo: create, update, complete, delete, list y search de tareas
  • Los Resources exponen datos de solo lectura: task://list, task://stats, category://list
  • Los Prompts proporcionan templates reutilizables para análisis y planificación
  • Toda la lógica usa queries parametrizadas para prevenir SQL injection
  • El server incluye validación de entrada con Pydantic y protección contra queries destructivas en run_query
  • Verificaste cada componente con MCP Inspector antes de continuar

Recursos

  1. SQLite Python Documentation — API oficial de sqlite3
  2. FastMCP Documentation — SDK de MCP para Python
  3. Pydantic Field Validators — Validación avanzada
  4. SQLite WAL Mode — Write-Ahead Logging para concurrencia
  5. Python Context Managers — contextlib y @contextmanager
  6. MCP Inspector — Debugging visual de MCP servers
← AnteriorDiseño y Arquitectura del MCP ServerSiguiente →Testing y Documentación