Módulo 8: Proyecto — MCP Server Real
Testing y Documentación
Testing y Documentación
Descripción de la cápsula
Tu server funciona en MCP Inspector. Puedes crear tareas, listar categorías, ejecutar queries. Pero "funciona cuando lo pruebo manualmente" no es lo mismo que "funciona." En esta cápsula escribes tests automatizados que verifican cada tool, resource, y prompt, y documentas el server para que cualquiera pueda instalarlo y usarlo.
Dos deliverables al terminar: un test suite que pasa con pytest, y un README.md completo.
Setup de testing
Dependencias
Si seguiste la cápsula 02, ya tienes pytest instalado. Verifica:
cd task-manager-mcp
source .venv/bin/activate
pytest --versionSi no lo tienes:
pip install pytest pytest-asyncioConfiguración de pytest
Crea pytest.ini en la raíz del proyecto:
[pytest]
asyncio_mode = auto
testpaths = tests
python_files = test_*.py
python_functions = test_*asyncio_mode = auto permite usar async def test_...() directamente, sin necesidad del decorador @pytest.mark.asyncio en cada test.
Estrategia de testing
Los tests usan una database SQLite en memoria — no tocan tu database real. Cada test empieza con datos limpios y conocidos. Esto te da:
- Velocidad: Sin I/O de disco, los tests corren en milisegundos
- Aislamiento: Un test que falla no corrompe datos de otro test
- Reproducibilidad: Los mismos datos producen los mismos resultados, siempre
Fixtures compartidos
tests/conftest.py
import pytest
import sqlite3
from unittest.mock import patch
from src.database import init_database, seed_sample_data
@pytest.fixture
def db_path(tmp_path):
"""Crea una database temporal para cada test."""
path = str(tmp_path / "test_tasks.db")
init_database(path)
seed_sample_data(path)
return path
@pytest.fixture
def empty_db_path(tmp_path):
"""Database temporal vacía (sin seed data)."""
path = str(tmp_path / "test_empty.db")
init_database(path)
return path
@pytest.fixture
def mock_db(db_path):
"""Patchea get_connection para usar la database de test."""
from contextlib import contextmanager
@contextmanager
def mock_get_connection(path=None):
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys=ON")
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
with patch("src.tools.tasks.get_connection", mock_get_connection), \
patch("src.tools.categories.get_connection", mock_get_connection), \
patch("src.tools.queries.get_connection", mock_get_connection), \
patch("src.resources.database.get_connection", mock_get_connection):
yield db_path
@pytest.fixture
def mock_empty_db(empty_db_path):
"""Patchea get_connection para usar la database vacía."""
from contextlib import contextmanager
@contextmanager
def mock_get_connection(path=None):
conn = sqlite3.connect(empty_db_path)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys=ON")
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
with patch("src.tools.tasks.get_connection", mock_get_connection), \
patch("src.tools.categories.get_connection", mock_get_connection), \
patch("src.tools.queries.get_connection", mock_get_connection), \
patch("src.resources.database.get_connection", mock_get_connection):
yield empty_db_pathPor qué patchear get_connection:
Cada módulo de tools y resources importa get_connection desde src.database. El patch redirige esas llamadas a la database de test. Así, los tests nunca tocan data/tasks.db.
Tests de tools: Tareas
tests/test_tools_tasks.py
import json
import pytest
from src.tools.tasks import create_task, list_tasks, update_task, delete_task, search_tasks
from src.models import (
CreateTaskInput, ListTasksInput, UpdateTaskInput, SearchTasksInput,
TaskStatus, TaskPriority,
)
class TestCreateTask:
async def test_create_basic_task(self, mock_db):
result = await create_task(CreateTaskInput(title="New test task"))
data = json.loads(result)
assert data["created"] is True
assert data["task"]["title"] == "New test task"
assert data["task"]["status"] == "pending"
assert data["task"]["priority"] == "medium"
assert data["task"]["id"] is not None
async def test_create_task_with_all_fields(self, mock_db):
result = await create_task(CreateTaskInput(
title="Complete task",
description="With all the fields",
status=TaskStatus.IN_PROGRESS,
priority=TaskPriority.HIGH,
category_id=1,
due_date="2026-04-01",
tags=["urgent", "feature"],
))
data = json.loads(result)
assert data["created"] is True
assert data["task"]["priority"] == "high"
assert data["task"]["status"] == "in_progress"
assert data["task"]["category_id"] == 1
assert "urgent" in data["task"]["tags"]
async def test_create_task_invalid_category(self, mock_db):
result = await create_task(CreateTaskInput(
title="Task with invalid category",
category_id=999,
))
data = json.loads(result)
assert data["error"] is True
assert data["error_type"] == "not_found"
assert "999" in data["message"]
async def test_create_task_with_new_tags(self, mock_db):
result = await create_task(CreateTaskInput(
title="Task with new tags",
tags=["new_tag", "another_tag"],
))
data = json.loads(result)
assert data["created"] is True
assert "new_tag" in data["task"]["tags"]
class TestListTasks:
async def test_list_all_tasks(self, mock_db):
result = await list_tasks(ListTasksInput())
data = json.loads(result)
assert data["total"] > 0
assert len(data["tasks"]) <= 20
async def test_list_tasks_filter_status(self, mock_db):
result = await list_tasks(ListTasksInput(status=TaskStatus.COMPLETED))
data = json.loads(result)
for task in data["tasks"]:
assert task["status"] == "completed"
async def test_list_tasks_filter_priority(self, mock_db):
result = await list_tasks(ListTasksInput(priority=TaskPriority.HIGH))
data = json.loads(result)
for task in data["tasks"]:
assert task["priority"] == "high"
async def test_list_tasks_with_limit(self, mock_db):
result = await list_tasks(ListTasksInput(limit=3))
data = json.loads(result)
assert len(data["tasks"]) <= 3
async def test_list_tasks_includes_category_name(self, mock_db):
result = await list_tasks(ListTasksInput())
data = json.loads(result)
task_with_category = next(
(t for t in data["tasks"] if t["category_id"] is not None), None
)
assert task_with_category is not None
assert task_with_category["category_name"] is not None
async def test_list_tasks_empty_db(self, mock_empty_db):
result = await list_tasks(ListTasksInput())
data = json.loads(result)
assert data["total"] == 0
assert data["tasks"] == []
class TestUpdateTask:
async def test_update_title(self, mock_db):
result = await update_task(UpdateTaskInput(task_id=1, title="Updated title"))
data = json.loads(result)
assert data["updated"] is True
assert data["task"]["title"] == "Updated title"
async def test_update_status(self, mock_db):
result = await update_task(UpdateTaskInput(task_id=1, status=TaskStatus.COMPLETED))
data = json.loads(result)
assert data["task"]["status"] == "completed"
async def test_update_nonexistent_task(self, mock_db):
result = await update_task(UpdateTaskInput(task_id=999, title="Does not exist"))
data = json.loads(result)
assert data["error"] is True
assert data["error_type"] == "not_found"
async def test_update_no_fields(self, mock_db):
result = await update_task(UpdateTaskInput(task_id=1))
data = json.loads(result)
assert data["error"] is True
assert data["error_type"] == "validation"
class TestDeleteTask:
async def test_delete_existing_task(self, mock_db):
result = await delete_task(task_id=1)
data = json.loads(result)
assert data["deleted"] is True
assert data["task_id"] == 1
verify = await list_tasks(ListTasksInput())
verify_data = json.loads(verify)
ids = [t["id"] for t in verify_data["tasks"]]
assert 1 not in ids
async def test_delete_nonexistent_task(self, mock_db):
result = await delete_task(task_id=999)
data = json.loads(result)
assert data["error"] is True
assert data["error_type"] == "not_found"
class TestSearchTasks:
async def test_search_by_title(self, mock_db):
result = await search_tasks(SearchTasksInput(query="JWT", search_in="title"))
data = json.loads(result)
assert data["results"] > 0
assert any("JWT" in t["title"] for t in data["tasks"])
async def test_search_no_results(self, mock_db):
result = await search_tasks(SearchTasksInput(query="xyznonexistent123"))
data = json.loads(result)
assert data["results"] == 0
assert data["tasks"] == []
async def test_search_both_fields(self, mock_db):
result = await search_tasks(SearchTasksInput(query="bug", search_in="both"))
data = json.loads(result)
assert data["results"] > 0Tests de tools: Categorías y queries
tests/test_tools_categories.py
import json
import pytest
from src.tools.categories import create_category
from src.models import CreateCategoryInput
class TestCreateCategory:
async def test_create_category(self, mock_db):
result = await create_category(CreateCategoryInput(
name="Testing",
description="Testing category",
color="#FF5733",
))
data = json.loads(result)
assert data["created"] is True
assert data["category"]["name"] == "Testing"
assert data["category"]["color"] == "#FF5733"
async def test_create_duplicate_category(self, mock_db):
result = await create_category(CreateCategoryInput(name="Backend"))
data = json.loads(result)
assert data["error"] is True
assert data["error_type"] == "duplicate"tests/test_tools_queries.py
import json
import pytest
from src.tools.queries import run_query, get_task_summary
from src.models import RunQueryInput, TaskSummaryInput
class TestRunQuery:
async def test_select_query(self, mock_db):
result = await run_query(RunQueryInput(sql="SELECT COUNT(*) as total FROM tasks"))
data = json.loads(result)
assert data["row_count"] == 1
assert data["results"][0]["total"] == 10
async def test_select_with_where(self, mock_db):
result = await run_query(RunQueryInput(
sql="SELECT title FROM tasks WHERE priority = 'high'"
))
data = json.loads(result)
assert data["row_count"] > 0
assert "title" in data["columns"]
async def test_reject_insert(self, mock_db):
result = await run_query(RunQueryInput(
sql="INSERT INTO tasks (title) VALUES ('hacked')"
))
data = json.loads(result)
assert data["error"] is True
assert data["error_type"] == "permission"
async def test_reject_delete(self, mock_db):
result = await run_query(RunQueryInput(sql="DELETE FROM tasks"))
data = json.loads(result)
assert data["error"] is True
assert data["error_type"] == "permission"
async def test_reject_drop(self, mock_db):
result = await run_query(RunQueryInput(sql="DROP TABLE tasks"))
data = json.loads(result)
assert data["error"] is True
async def test_invalid_sql(self, mock_db):
result = await run_query(RunQueryInput(sql="SELCT * FORM tasks"))
data = json.loads(result)
assert data["error"] is True
assert data["error_type"] == "query_error"
class TestTaskSummary:
async def test_summary_week(self, mock_db):
result = await get_task_summary(TaskSummaryInput(period="week"))
data = json.loads(result)
assert "by_status" in data
assert "by_priority" in data
assert "overdue_count" in data
assert data["period"] == "week"
async def test_summary_invalid_period(self, mock_db):
result = await get_task_summary(TaskSummaryInput(period="year"))
data = json.loads(result)
assert data["error"] is True
assert data["error_type"] == "validation"Tests de resources
tests/test_resources.py
import json
import pytest
from src.resources.database import (
get_tables, get_table_schema, get_stats,
get_overdue_tasks, get_categories,
)
class TestGetTables:
async def test_returns_all_tables(self, mock_db):
result = await get_tables()
data = json.loads(result)
table_names = [t["name"] for t in data["tables"]]
assert "tasks" in table_names
assert "categories" in table_names
assert "tags" in table_names
assert "task_tags" in table_names
async def test_includes_row_counts(self, mock_db):
result = await get_tables()
data = json.loads(result)
tasks_table = next(t for t in data["tables"] if t["name"] == "tasks")
assert tasks_table["row_count"] == 10
class TestGetTableSchema:
async def test_tasks_schema(self, mock_db):
result = await get_table_schema("tasks")
data = json.loads(result)
assert data["table"] == "tasks"
column_names = [c["name"] for c in data["columns"]]
assert "id" in column_names
assert "title" in column_names
assert "status" in column_names
assert "priority" in column_names
async def test_nonexistent_table(self, mock_db):
result = await get_table_schema("nonexistent")
data = json.loads(result)
assert "error" in data
assert "no encontrada" in data["error"]
async def test_schema_includes_foreign_keys(self, mock_db):
result = await get_table_schema("task_tags")
data = json.loads(result)
assert len(data["foreign_keys"]) > 0
class TestGetStats:
async def test_stats_with_seed_data(self, mock_db):
result = await get_stats()
data = json.loads(result)
assert data["total_tasks"] == 10
assert "by_status" in data
assert "by_priority" in data
assert "by_category" in data
assert data["total_categories"] == 4
async def test_stats_empty_db(self, mock_empty_db):
result = await get_stats()
data = json.loads(result)
assert data["total_tasks"] == 0
class TestGetCategories:
async def test_categories_list(self, mock_db):
result = await get_categories()
data = json.loads(result)
assert data["total"] == 4
names = [c["name"] for c in data["categories"]]
assert "Backend" in names
assert "Frontend" in names
async def test_categories_include_task_count(self, mock_db):
result = await get_categories()
data = json.loads(result)
for cat in data["categories"]:
assert "task_count" in cat
assert isinstance(cat["task_count"], int)
class TestGetOverdueTasks:
async def test_overdue_structure(self, mock_db):
result = await get_overdue_tasks()
data = json.loads(result)
assert "overdue_count" in data
assert "tasks" in data
assert isinstance(data["tasks"], list)Tests de integración
tests/test_integration.py
import json
import pytest
from src.tools.tasks import create_task, list_tasks, update_task, delete_task
from src.tools.categories import create_category
from src.resources.database import get_stats
from src.models import (
CreateTaskInput, ListTasksInput, UpdateTaskInput,
CreateCategoryInput, TaskStatus, TaskPriority,
)
class TestCRUDFlow:
"""Verifica el flujo completo: crear → leer → actualizar → eliminar."""
async def test_full_task_lifecycle(self, mock_empty_db):
cat_result = await create_category(CreateCategoryInput(name="Integration Test"))
cat_data = json.loads(cat_result)
category_id = cat_data["category"]["id"]
create_result = await create_task(CreateTaskInput(
title="Integration task",
description="Full lifecycle test",
priority=TaskPriority.HIGH,
category_id=category_id,
tags=["integration", "test"],
))
create_data = json.loads(create_result)
assert create_data["created"] is True
task_id = create_data["task"]["id"]
list_result = await list_tasks(ListTasksInput())
list_data = json.loads(list_result)
assert list_data["total"] == 1
assert list_data["tasks"][0]["title"] == "Integration task"
update_result = await update_task(UpdateTaskInput(
task_id=task_id,
status=TaskStatus.COMPLETED,
title="Integration task (completed)",
))
update_data = json.loads(update_result)
assert update_data["task"]["status"] == "completed"
assert update_data["task"]["title"] == "Integration task (completed)"
delete_result = await delete_task(task_id=task_id)
delete_data = json.loads(delete_result)
assert delete_data["deleted"] is True
final_list = await list_tasks(ListTasksInput())
final_data = json.loads(final_list)
assert final_data["total"] == 0
class TestStatsReflectChanges:
"""Verifica que los resources reflejan cambios hechos por tools."""
async def test_stats_update_after_create(self, mock_empty_db):
stats_before = json.loads(await get_stats())
assert stats_before["total_tasks"] == 0
await create_task(CreateTaskInput(title="First task"))
await create_task(CreateTaskInput(title="Second task"))
stats_after = json.loads(await get_stats())
assert stats_after["total_tasks"] == 2Ejecutar los tests
cd task-manager-mcp
source .venv/bin/activate
PYTHONPATH=. pytest -vDeberías ver output como:
tests/test_tools_tasks.py::TestCreateTask::test_create_basic_task PASSED
tests/test_tools_tasks.py::TestCreateTask::test_create_task_with_all_fields PASSED
tests/test_tools_tasks.py::TestCreateTask::test_create_task_invalid_category PASSED
tests/test_tools_tasks.py::TestCreateTask::test_create_task_with_new_tags PASSED
tests/test_tools_tasks.py::TestListTasks::test_list_all_tasks PASSED
...
tests/test_integration.py::TestCRUDFlow::test_full_task_lifecycle PASSED
tests/test_integration.py::TestStatsReflectChanges::test_stats_update_after_create PASSED
========================= 30+ passed =========================
Si algún test falla, lee el error. Los tests están diseñados para que el mensaje de error te diga exactamente qué salió mal.
Documentación: README.md
Template de README
Crea README.md en la raíz del proyecto:
# Task Manager MCP Server
MCP server que conecta Claude Code con una base de datos SQLite de gestión de tareas.
Permite crear, listar, buscar, actualizar y eliminar tareas organizadas por categorías
y tags, con reportes y análisis a través de prompts.
## Requisitos
- Python 3.11+
- pip
## Instalación
git clone <tu-repo>
cd task-manager-mcp
python -m venv .venv
source .venv/bin/activate # Linux/macOS
# .venv\Scripts\activate # Windows
pip install -r requirements.txt
## Ejecutar el server
PYTHONPATH=. python src/server.py
El server inicializa la base de datos y carga datos de ejemplo automáticamente
en la primera ejecución.
## Testing con MCP Inspector
PYTHONPATH=. mcp dev src/server.py
Abre MCP Inspector en el navegador para probar tools y resources interactivamente.
## Ejecutar tests
PYTHONPATH=. pytest -v
## Conectar a Claude Code
claude mcp add task-manager \
/ruta/completa/task-manager-mcp/.venv/bin/python \
-e PYTHONPATH=/ruta/completa/task-manager-mcp \
-- /ruta/completa/task-manager-mcp/src/server.py
Verificar:
claude
> /mcp
Deberías ver `task-manager` con estado "connected".
## Tools disponibles
| Tool | Descripción | Parámetros |
|------|-------------|------------|
| `create_task` | Crea una nueva tarea | title (req), description, status, priority, category_id, due_date, tags |
| `list_tasks` | Lista tareas con filtros | status, priority, category_id, limit |
| `update_task` | Actualiza una tarea | task_id (req), title, description, status, priority, category_id, due_date |
| `delete_task` | Elimina una tarea | task_id (req) |
| `search_tasks` | Busca tareas por texto | query (req), search_in (title/description/both) |
| `create_category` | Crea una categoría | name (req), description, color |
| `run_query` | Ejecuta SQL (solo SELECT) | sql (req) |
| `get_task_summary` | Resumen por periodo | period (req): today, week, month |
## Resources disponibles
| URI | Descripción |
|-----|-------------|
| `taskdb://tables` | Lista de tablas con conteo de registros |
| `taskdb://table/{name}/schema` | Schema de una tabla (columnas, tipos, constraints) |
| `taskdb://stats` | Estadísticas generales (tareas por status, prioridad, categoría) |
| `taskdb://tasks/overdue` | Tareas con fecha vencida |
| `taskdb://categories` | Categorías con conteo de tareas |
## Prompts disponibles
| Prompt | Descripción | Parámetros |
|--------|-------------|------------|
| `analyze_table` | Analiza estructura y datos de una tabla | table_name |
| `weekly_report` | Genera reporte semanal de productividad | week_start (opcional) |
| `optimize_query` | Analiza y optimiza una query SQL | sql_query |
## Ejemplos de uso con Claude Code
**Crear una tarea:**
> Crea una tarea "Implementar endpoint de login" con prioridad alta en la categoría Backend
**Buscar tareas:**
> Busca tareas que mencionen "bug" o "fix"
**Reporte semanal:**
> Genera un reporte de productividad de esta semana
**Análisis de tabla:**
> Analiza la estructura de la tabla tasks y sugiere mejoras
**Query custom:**
> Ejecuta: SELECT status, COUNT(*) FROM tasks GROUP BY status
## Estructura del proyecto
task-manager-mcp/
├── src/
│ ├── server.py # Entry point
│ ├── database.py # Conexión SQLite
│ ├── models.py # Pydantic models
│ ├── tools/
│ │ ├── tasks.py # CRUD de tareas
│ │ ├── categories.py # Gestión de categorías
│ │ └── queries.py # Queries y reportes
│ └── resources/
│ └── database.py # Resources de la DB
├── tests/ # Test suite
├── data/ # SQLite database
├── requirements.txt
└── README.md
## Licencia
MITVerificar documentación
Tu README debe pasar esta checklist:
- Instalación: Alguien que clona el repo puede seguir los pasos y tener el server corriendo
- Todos los tools listados: Con parámetros y descripción
- Todos los resources listados: Con URI y descripción
- Todos los prompts listados: Con parámetros
- Instrucciones de Claude Code: Cómo conectar y verificar
- Ejemplos de uso: Al menos 3 ejemplos concretos
- Estructura del proyecto: Árbol de archivos
- Tests: Cómo ejecutarlos
Milestone de esta cápsula
Al terminar, verifica:
-
PYTHONPATH=. pytest -vpasa todos los tests (30+) - Tests cubren happy paths: crear, listar, actualizar, borrar, buscar
- Tests cubren error cases: ID inexistente, categoría duplicada, query prohibida
- Tests de integración: flujo CRUD completo, stats reflejan cambios
- README.md completo con todas las secciones
- Puedes seguir el README desde cero y tener el server corriendo
Tu server ahora tiene dos garantías que no tenía antes: (1) si algo se rompe, un test lo detecta, y (2) alguien que nunca vio tu código puede instalarlo y usarlo siguiendo el README.
Cápsula 05 es donde conectas todo a Claude Code y haces la demo end-to-end.
Troubleshooting
"pytest no encuentra los tests"
Verifica que pytest.ini tiene testpaths = tests y que tus archivos se llaman test_*.py. Si usas un pyproject.toml, la configuración de pytest puede estar ahí en vez de en pytest.ini.
"async tests no se ejecutan"
Verifica que tienes asyncio_mode = auto en pytest.ini y que pytest-asyncio está instalado. Sin esto, pytest ignora las funciones async def test_....
"ModuleNotFoundError en tests"
Ejecuta con PYTHONPATH=. desde la raíz del proyecto:
cd task-manager-mcp
PYTHONPATH=. pytest -v"Los patches no funcionan — usa la database real"
Verifica que el path del patch coincide con donde se importa get_connection. Si src/tools/tasks.py tiene from src.database import get_connection, el patch debe ser src.tools.tasks.get_connection, no src.database.get_connection.
"Test falla con 'no such table: tasks'"
El fixture mock_db o mock_empty_db no se está aplicando. Verifica que tu test recibe el fixture como parámetro: async def test_something(self, mock_db):.
"Tests pasan localmente pero fallan en CI"
Revisa que el CI ejecuta con PYTHONPATH=. y que las dependencias incluyen pytest-asyncio. Verifica también que la versión de Python es 3.11+.
Resumen
- Construiste un test suite completo con pytest y pytest-asyncio para tu MCP server
- Los tests usan SQLite in-memory para aislamiento total entre tests
- Cubriste tests unitarios (database, tools, resources) y tests de integración (flujos end-to-end)
- La documentación incluye README profesional con instalación, configuración y uso
- Documentaste todas las capabilities (tools, resources, prompts) con ejemplos de uso
- El CHANGELOG sigue el formato Keep a Changelog para tracking de versiones
- Con tests y docs completos, tu server está listo para la demo final en la siguiente cápsula
Recursos
- pytest Documentation — Documentación oficial de pytest
- pytest-asyncio — Plugin para tests async
- Python unittest.mock — Patching y mocking
- SQLite In-Memory Databases — Databases en memoria para testing
- Write The Docs Guide — Guía de documentación técnica
- Keep a Changelog — Formato estándar de changelogs