Module 8: Project — Real-World MCP Server
Core Implementation of the MCP Server
Core Implementation of the MCP Server
Capsule description
Design finished. Now you build. This capsule contains the complete implementation of the MCP server for Option A (SQLite + Python): database setup, resources, tools, prompts, and error handling. All the code is executable — you can copy each block, and by the end of the capsule you'll have a functional server that responds in MCP Inspector.
If you chose Option B or C, use this implementation as a reference for patterns. The structure is the same: data source setup → resources → tools → prompts → entry point. Only how you access the data changes.
Step 1: Database setup and connection
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:
"""Returns the path to the database. Creates the directory if it doesn't exist."""
DB_DIR.mkdir(parents=True, exist_ok=True)
return str(DB_PATH)
@contextmanager
def get_connection(db_path: str | None = None):
"""Context manager for SQLite connections.
Uses WAL mode for better concurrency and configures row_factory
to return dictionaries instead of tuples.
"""
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:
"""Creates the tables if they don't exist."""
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:
"""Inserts example data for demos and 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")Key points of the implementation:
get_connectionas a context manager guarantees that the connection always closes, even if there are errorsrow_factory = sqlite3.Rowlets you access columns by name (row["title"]) instead of by indexPRAGMA journal_mode=WALimproves concurrency — reads don't block writesPRAGMA foreign_keys=ONactivates the foreign key constraints (SQLite disables them by default)seed_sample_datais idempotent — if there's already data, it doesn't insert duplicates- The indexes on
status,priority,category_id, anddue_datespeed up the most common filters
Step 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="Task title")
description: Optional[str] = Field(default=None, max_length=2000, description="Detailed description")
status: TaskStatus = Field(default=TaskStatus.PENDING, description="Initial status")
priority: TaskPriority = Field(default=TaskPriority.MEDIUM, description="Priority: low, medium, high, critical")
category_id: Optional[int] = Field(default=None, ge=1, description="Category ID")
due_date: Optional[str] = Field(default=None, description="Due date YYYY-MM-DD")
tags: Optional[list[str]] = Field(default=None, description="Tags for the task")
class UpdateTaskInput(BaseModel):
task_id: int = Field(ge=1, description="ID of the task to update")
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="Filter by status")
priority: Optional[TaskPriority] = Field(default=None, description="Filter by priority")
category_id: Optional[int] = Field(default=None, ge=1, description="Filter by category")
limit: int = Field(default=20, ge=1, le=100, description="Maximum results")
class SearchTasksInput(BaseModel):
query: str = Field(min_length=1, description="Text to search in tasks")
search_in: str = Field(default="both", description="Where to search: title, description, both")
class CreateCategoryInput(BaseModel):
name: str = Field(min_length=1, max_length=50, description="Category name")
description: Optional[str] = Field(default=None, max_length=200)
color: str = Field(default="#6B7280", pattern=r"^#[0-9A-Fa-f]{6}$", description="Hex color")
class RunQueryInput(BaseModel):
sql: str = Field(min_length=1, description="SQL query — only SELECT allowed")
class TaskSummaryInput(BaseModel):
period: str = Field(description="Summary period: today, week, month")Step 3: Tools — Task CRUD operations
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:
"""Converts a SQLite Row to a serializable dictionary."""
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:
"""Creates a new task in the database.
Returns the created task with its assigned ID. If tags are included,
it associates them automatically (creates new tags if they don't exist).
"""
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"Category with ID {input.category_id} not found")
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 creating task: {e}")
async def list_tasks(input: ListTasksInput) -> str:
"""Lists tasks with optional filters by status, priority and category.
Without filters it returns the most recent tasks. Includes the category
name and the tags associated with each task.
"""
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 listing tasks: {e}")
async def update_task(input: UpdateTaskInput) -> str:
"""Updates fields of an existing task.
Only modifies the fields that are sent (the rest stay the same).
Returns the complete updated task.
"""
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"Task with ID {input.task_id} not found")
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 fields were provided to update")
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 updating task: {e}")
async def delete_task(task_id: int) -> str:
"""Deletes a task from the database.
It also deletes the associations with tags (CASCADE). Returns confirmation
with the ID of the deleted task.
"""
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"Task with ID {task_id} not found")
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 deleting task: {e}")
async def search_tasks(input: SearchTasksInput) -> str:
"""Searches tasks by text in the title, description, or both.
The search is case-insensitive. Returns tasks that contain
the searched text in the selected fields.
"""
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 searching tasks: {e}")Step 4: Tools — Categories and 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:
"""Creates a new category to organize tasks.
Categories group tasks by area (Backend, Frontend, DevOps, etc.).
Each category has a unique name, an optional description, and a hex color.
"""
try:
with get_connection() as conn:
existing = conn.execute("SELECT id FROM categories WHERE name = ?", (input.name,)).fetchone()
if existing:
return _error("duplicate", f"A category with name '{input.name}' already exists")
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 creating category: {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:
"""Runs a read-only SQL query against the database.
Only SELECT queries are allowed. Queries that modify data
(INSERT, UPDATE, DELETE, DROP) will be rejected. Useful for ad-hoc
queries that aren't covered by the other tools.
"""
if FORBIDDEN_KEYWORDS.search(input.sql):
return _error("permission", "Only SELECT queries are allowed. Queries that modify data cannot be executed.")
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 in the query: {e}")
async def get_task_summary(input: TaskSummaryInput) -> str:
"""Generates a summary of tasks for a specific period.
Available periods: 'today' (today), 'week' (last 7 days),
'month' (last 30 days). Includes counts by status and priority.
"""
period_map = {
"today": 0,
"week": 7,
"month": 30,
}
if input.period not in period_map:
return _error("validation", f"Period '{input.period}' not valid. Use: 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 generating summary: {e}")Step 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:
"""Lists all the tables in the database with their record count."""
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 listing tables: {e}"})
async def get_table_schema(table_name: str) -> str:
"""Returns a table's schema: columns, types, 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"Table '{table_name}' not found. Available tables: {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 getting schema: {e}"})
async def get_stats() -> str:
"""General statistics of the task database."""
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, 'Uncategorized') 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 getting statistics: {e}"})
async def get_overdue_tasks() -> str:
"""Overdue tasks: past due_date and status not completed/cancelled."""
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 getting overdue tasks: {e}"})
async def get_categories() -> str:
"""Lists all the categories with the task count per category."""
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 listing categories: {e}"})Step 6: Main server — everything connected
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 to manage tasks with SQLite. You can create, list, "
"update, search and delete tasks. Tasks are organized by categories "
"and tags, and have statuses (pending, in_progress, completed, cancelled) and "
"priorities (low, medium, high, critical). Use the resources to query "
"the database structure and general statistics."
),
)
# --- 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:
"""Analyzes the structure and data of a table, suggests improvements."""
return f"""Analyze the table '{table_name}' in the task manager database.
Please:
1. Read the resource taskdb://table/{table_name}/schema to see the structure
2. Use the run_query tool with: SELECT COUNT(*) FROM {table_name}
3. Use the run_query tool with: SELECT * FROM {table_name} LIMIT 5
4. Read the resource taskdb://stats for general context
With that information, generate an analysis that includes:
- Table structure (columns, types, constraints)
- Current data volume
- Distribution of values in key columns
- Possible improvements (indexes, additional constraints, normalization)
- 3 useful queries to explore this table"""
@mcp.prompt()
async def weekly_report(week_start: str = "") -> str:
"""Generates a weekly productivity report."""
context = f"for the week of {week_start}" if week_start else "for this week"
return f"""Generate a productivity report {context}.
Please:
1. Use the get_task_summary tool with period='week'
2. Use the list_tasks tool with status='completed' for finished tasks
3. Read the resource taskdb://tasks/overdue for overdue tasks
4. Read the resource taskdb://stats for general context
Generate a report that includes:
- Executive summary (completed vs pending vs overdue)
- Completed tasks (list with category and priority)
- Priority pending tasks (high and critical)
- Overdue tasks that require attention
- Completion rate for the period
- Recommendations for next week"""
@mcp.prompt()
async def optimize_query(sql_query: str) -> str:
"""Analyzes a SQL query and suggests optimizations."""
return f"""Analyze and optimize the following SQL query:
```sql
{sql_query}Please:
- Read the resources taskdb://tables and taskdb://table/tasks/schema to understand the structure
- Run the original query with the run_query tool to see the result
- If possible, run EXPLAIN QUERY PLAN with run_query
Generate an analysis that includes:
- What the query does
- Possible performance problems
- Optimized query (if applicable)
- Recommended indexes
- More efficient alternatives"""
if name == "main": mcp.run()
---
## Step 7: Verify in MCP Inspector
With all the code in place, verify that it works:
```bash
cd task-manager-mcp
source .venv/bin/activate
PYTHONPATH=. mcp dev src/server.py
MCP Inspector should show:
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
Quick verifications in Inspector
- Resource
taskdb://tables— You should see 4 tables with their counts - Resource
taskdb://stats— Statistics with 10 tasks (seed data) - Tool
list_taskswith{}(no filters) — The 10 tasks from the seed data - Tool
create_taskwith{"title": "Test from Inspector"}— Create a new task - Tool
search_taskswith{"query": "bug"}— Find the pagination bug task
If the 5 verifications pass, your server is ready. Capsule 04 adds automated tests and documentation.
Notes for Options B and C
Option B: File System + TypeScript
The patterns are the same. Instead of get_connection() you use fs.promises. Instead of SQL you use file system operations. The entry point registers tools with server.tool() and resources with server.resource().
The equivalent of database.py would be a module that configures the root directory of the project to analyze and validates that it exists.
Option C: External API
Instead of get_connection() you use httpx.AsyncClient(). Instead of SQL queries you make HTTP requests. The error handling includes timeouts, rate limits, and authentication.
The equivalent of seed_sample_data() doesn't apply — the data already exists in the external API.
Troubleshooting
"ModuleNotFoundError: No module named 'src'"
Run with PYTHONPATH=. so that Python finds the src package:
PYTHONPATH=. python src/server.py
PYTHONPATH=. mcp dev src/server.py"sqlite3.OperationalError: database is locked"
Another instance of the server has the database open. Close other processes that use tasks.db, or restart the server.
"The seed data duplicates every time I start the server"
seed_sample_data() checks whether there's already data before inserting. If you see duplicates, you're probably calling init_database() with a different path each time. Verify that DB_PATH is consistent.
"The tags don't get associated with the task"
Verify that PRAGMA foreign_keys=ON is active. Without this, SQLite ignores the foreign keys and the INSERTs on task_tags can fail silently.
"MCP Inspector doesn't show all the tools"
Verify that there are no import errors. Run PYTHONPATH=. python -c "from src.server import mcp" to verify that everything imports correctly.
"run_query allows destructive queries"
The FORBIDDEN_KEYWORDS regex should block INSERT, UPDATE, DELETE, DROP, etc. If a keyword isn't in the list, add it. Remember that the protection is basic — for a real production server, you'd need more robust sanitization.
Summary
- You implemented the complete database layer: connection, schema creation, and parameterized queries
- The Tools handle complete CRUD: create, update, complete, delete, list and search tasks
- The Resources expose read-only data: task://list, task://stats, category://list
- The Prompts provide reusable templates for analysis and planning
- All the logic uses parameterized queries to prevent SQL injection
- The server includes input validation with Pydantic and protection against destructive queries in
run_query - You verified each component with MCP Inspector before continuing
Resources
- SQLite Python Documentation — Official sqlite3 API
- FastMCP Documentation — MCP SDK for Python
- Pydantic Field Validators — Advanced validation
- SQLite WAL Mode — Write-Ahead Logging for concurrency
- Python Context Managers — contextlib and @contextmanager
- MCP Inspector — Visual debugging of MCP servers