Module 8: Project — Real-World MCP Server

Testing and Documentation

Testing and Documentation

Capsule description

Your server works in MCP Inspector. You can create tasks, list categories, run queries. But "it works when I test it manually" isn't the same as "it works." In this capsule you write automated tests that verify each tool, resource, and prompt, and you document the server so anyone can install and use it.

Two deliverables when you finish: a test suite that passes with pytest, and a complete README.md.


Testing setup

Dependencies

If you followed capsule 02, you already have pytest installed. Verify:

cd task-manager-mcp
source .venv/bin/activate
pytest --version

If you don't have it:

pip install pytest pytest-asyncio

pytest configuration

Create pytest.ini in the project's root:

[pytest]
asyncio_mode = auto
testpaths = tests
python_files = test_*.py
python_functions = test_*

asyncio_mode = auto lets you use async def test_...() directly, without needing the @pytest.mark.asyncio decorator on each test.

Testing strategy

The tests use an in-memory SQLite database — they don't touch your real database. Each test starts with clean, known data. This gives you:

  • Speed: Without disk I/O, the tests run in milliseconds
  • Isolation: A test that fails doesn't corrupt another test's data
  • Reproducibility: The same data produces the same results, always

Shared fixtures

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):
    """Creates a temporary database for each 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):
    """Empty temporary database (without seed data)."""
    path = str(tmp_path / "test_empty.db")
    init_database(path)
    return path


@pytest.fixture
def mock_db(db_path):
    """Patches get_connection to use the test database."""
    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):
    """Patches get_connection to use the empty database."""
    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_path

Why patch get_connection:

Each tool and resource module imports get_connection from src.database. The patch redirects those calls to the test database. That way, the tests never touch data/tasks.db.


Tool tests: Tasks

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"] > 0

Tool tests: Categories and 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"

Resource tests

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 "not found" 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)

Integration tests

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:
    """Verifies the complete flow: create → read → update → delete."""

    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:
    """Verifies that the resources reflect changes made by 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"] == 2

Run the tests

cd task-manager-mcp
source .venv/bin/activate
PYTHONPATH=. pytest -v

You should see output like:

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 =========================

If any test fails, read the error. The tests are designed so that the error message tells you exactly what went wrong.


Documentation: README.md

README template

Create README.md in the project's root:

# Task Manager MCP Server

MCP server that connects Claude Code with a SQLite task management database.
It lets you create, list, search, update and delete tasks organized by categories
and tags, with reports and analysis through prompts.

## Requirements

- Python 3.11+
- pip

## Installation

    git clone <your-repo>
    cd task-manager-mcp
    python -m venv .venv
    source .venv/bin/activate    # Linux/macOS
    # .venv\Scripts\activate     # Windows
    pip install -r requirements.txt

## Run the server

    PYTHONPATH=. python src/server.py

The server initializes the database and loads example data automatically
on the first run.

## Testing with MCP Inspector

    PYTHONPATH=. mcp dev src/server.py

Open MCP Inspector in the browser to test tools and resources interactively.

## Run tests

    PYTHONPATH=. pytest -v

## Connect to Claude Code

    claude mcp add task-manager \
      /full/path/task-manager-mcp/.venv/bin/python \
      -e PYTHONPATH=/full/path/task-manager-mcp \
      -- /full/path/task-manager-mcp/src/server.py

Verify:

    claude
    > /mcp

You should see `task-manager` with "connected" status.

## Available tools

| Tool | Description | Parameters |
|------|-------------|------------|
| `create_task` | Creates a new task | title (req), description, status, priority, category_id, due_date, tags |
| `list_tasks` | Lists tasks with filters | status, priority, category_id, limit |
| `update_task` | Updates a task | task_id (req), title, description, status, priority, category_id, due_date |
| `delete_task` | Deletes a task | task_id (req) |
| `search_tasks` | Searches tasks by text | query (req), search_in (title/description/both) |
| `create_category` | Creates a category | name (req), description, color |
| `run_query` | Runs SQL (SELECT only) | sql (req) |
| `get_task_summary` | Summary by period | period (req): today, week, month |

## Available resources

| URI | Description |
|-----|-------------|
| `taskdb://tables` | List of tables with record count |
| `taskdb://table/{name}/schema` | Schema of a table (columns, types, constraints) |
| `taskdb://stats` | General statistics (tasks by status, priority, category) |
| `taskdb://tasks/overdue` | Tasks with a past due date |
| `taskdb://categories` | Categories with task count |

## Available prompts

| Prompt | Description | Parameters |
|--------|-------------|------------|
| `analyze_table` | Analyzes a table's structure and data | table_name |
| `weekly_report` | Generates a weekly productivity report | week_start (optional) |
| `optimize_query` | Analyzes and optimizes a SQL query | sql_query |

## Usage examples with Claude Code

**Create a task:**

    > Create a task "Implement login endpoint" with high priority in the Backend category

**Search tasks:**

    > Search for tasks that mention "bug" or "fix"

**Weekly report:**

    > Generate a productivity report for this week

**Table analysis:**

    > Analyze the structure of the tasks table and suggest improvements

**Custom query:**

    > Run: SELECT status, COUNT(*) FROM tasks GROUP BY status

## Project structure

    task-manager-mcp/
    ├── src/
    │   ├── server.py          # Entry point
    │   ├── database.py        # SQLite connection
    │   ├── models.py          # Pydantic models
    │   ├── tools/
    │   │   ├── tasks.py       # Task CRUD
    │   │   ├── categories.py  # Category management
    │   │   └── queries.py     # Queries and reports
    │   └── resources/
    │       └── database.py    # DB resources
    ├── tests/                 # Test suite
    ├── data/                  # SQLite database
    ├── requirements.txt
    └── README.md

## License

MIT

Verify the documentation

Your README must pass this checklist:

  • Installation: Someone who clones the repo can follow the steps and have the server running
  • All tools listed: With parameters and description
  • All resources listed: With URI and description
  • All prompts listed: With parameters
  • Claude Code instructions: How to connect and verify
  • Usage examples: At least 3 concrete examples
  • Project structure: File tree
  • Tests: How to run them

Milestone of this capsule

When you finish, verify:

  • PYTHONPATH=. pytest -v passes all the tests (30+)
  • Tests cover happy paths: create, list, update, delete, search
  • Tests cover error cases: nonexistent ID, duplicate category, forbidden query
  • Integration tests: complete CRUD flow, stats reflect changes
  • Complete README.md with all the sections
  • You can follow the README from scratch and have the server running

Your server now has two guarantees it didn't have before: (1) if something breaks, a test detects it, and (2) someone who never saw your code can install and use it by following the README.

Capsule 05 is where you connect everything to Claude Code and do the end-to-end demo.


Troubleshooting

"pytest doesn't find the tests"

Verify that pytest.ini has testpaths = tests and that your files are named test_*.py. If you use a pyproject.toml, the pytest configuration may be there instead of in pytest.ini.

"async tests don't run"

Verify that you have asyncio_mode = auto in pytest.ini and that pytest-asyncio is installed. Without this, pytest ignores the async def test_... functions.

"ModuleNotFoundError in tests"

Run with PYTHONPATH=. from the project's root:

cd task-manager-mcp
PYTHONPATH=. pytest -v

"The patches don't work — it uses the real database"

Verify that the patch path matches where get_connection is imported. If src/tools/tasks.py has from src.database import get_connection, the patch must be src.tools.tasks.get_connection, not src.database.get_connection.

"Test fails with 'no such table: tasks'"

The mock_db or mock_empty_db fixture isn't being applied. Verify that your test receives the fixture as a parameter: async def test_something(self, mock_db):.

"Tests pass locally but fail in CI"

Check that CI runs with PYTHONPATH=. and that the dependencies include pytest-asyncio. Also verify that the Python version is 3.11+.


Summary

  • You built a complete test suite with pytest and pytest-asyncio for your MCP server
  • The tests use in-memory SQLite for total isolation between tests
  • You covered unit tests (database, tools, resources) and integration tests (end-to-end flows)
  • The documentation includes a professional README with installation, configuration and usage
  • You documented all the capabilities (tools, resources, prompts) with usage examples
  • The CHANGELOG follows the Keep a Changelog format for version tracking
  • With complete tests and docs, your server is ready for the final demo in the next capsule

Resources

  1. pytest Documentation — Official pytest documentation
  2. pytest-asyncio — Plugin for async tests
  3. Python unittest.mock — Patching and mocking
  4. SQLite In-Memory Databases — In-memory databases for testing
  5. Write The Docs Guide — Technical documentation guide
  6. Keep a Changelog — Standard changelog format