The FastMCP tutorials so far returned pure, stateless results. Real tools usually need state that persists — a record that is still there after the process restarts. In this tutorial you will build a FastMCP task-manager server whose tools read and write a SQLite database, so tasks survive restarts, redeploys, and reboots.

You will separate the persistence layer (db.py) from the MCP tools (server.py), expose a live-stats resource, cover it with pytest (including a test that proves data survives a restart), and deploy it as a launchd service. The stack is Mac-native: uv, make, and Python’s built-in sqlite3 — no database server to install.

What you will build

  • A db.py persistence module over sqlite3: schema init and CRUD for tasks.
  • Tools: add_task, list_tasks, complete_task, delete_task, with input validation surfaced as ToolError.
  • A tasks://stats resource returning live counts.
  • A pytest suite that isolates the database per test and verifies persistence.
  • A launchd deployment whose database path is stable across redeploys.

Prerequisites

  • macOS 13+ with Homebrew (brew.sh).
  • uv 0.5+ — brew install uv; verify uv --version.
  • Xcode Command Line Tools (xcode-select --install) for make.
  • Basic SQL familiarity. sqlite3 ships with Python, so there is nothing else to install.

Everything runs through uv and make.

Step 1: Scaffold and lock down hygiene

Create the workspace and a .gitignore first. It ignores the database files — they are runtime state, not source.

Create the files

mkdir -p sqlite-fastmcp-server-macos
cd sqlite-fastmcp-server-macos
touch .gitignore

Add the code: .gitignore

# Python
__pycache__/
*.py[cod]
.venv/
.uv/
.pytest_cache/
.ruff_cache/
.mypy_cache/
# Runtime / deploy artifacts
logs/
*.log
*.pid
# SQLite database files (runtime state, not source)
*.db
*.db-journal
*.db-wal
*.db-shm
# OS / editor noise
.DS_Store

Detailed breakdown

  • *.db and the -journal/-wal/-shm variants are SQLite’s data and its write-ahead/journal sidecar files. They hold live state and must never be committed — the schema in code recreates an empty database on first run.
  • The rest is standard uv/macOS hygiene.

Step 2: Initialize the project with uv

Create the uv project and add FastMCP plus the test tools. sqlite3 is part of the standard library, so it is not a dependency.

Create the files

uv init --name macmcp --no-workspace
rm -f main.py hello.py
uv add fastmcp
uv add --dev pytest pytest-asyncio

Add the code: pyproject.toml

[project]
name = "macmcp"
version = "0.1.0"
description = "A stateful FastMCP server backed by SQLite"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "fastmcp>=3.4.3",
]

[dependency-groups]
dev = [
    "pytest>=9.1.1",
    "pytest-asyncio>=1.4.0",
]

Detailed breakdown

  • fastmcp is the only runtime dependency. Persistence uses the standard library, which keeps the deployment tiny and dependency-free.

Step 3: Write the persistence layer

Keep all database access in one module. Every operation opens a short-lived connection and commits — SQLite opens in microseconds, and this keeps the code thread-safe under an async HTTP server without a shared, cross-thread connection. The database path comes from an environment variable so tests and deployments can isolate it.

Create the file

touch db.py

Add the code: db.py

"""SQLite persistence for the tasks server.

State lives in a real database file, so tasks survive restarts and redeploys.
Every operation opens a short-lived connection (SQLite is fast to open and this
keeps the code thread-safe under an async HTTP server). The database path comes
from the DB_PATH environment variable so tests and deployments can point
somewhere isolated.
"""

import os
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path

DEFAULT_DB = "tasks.db"

SCHEMA = """
CREATE TABLE IF NOT EXISTS tasks (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    title      TEXT NOT NULL,
    notes      TEXT NOT NULL DEFAULT '',
    done       INTEGER NOT NULL DEFAULT 0,
    created_at TEXT NOT NULL
);
"""


def _path() -> Path:
    return Path(os.environ.get("DB_PATH", DEFAULT_DB))


@contextmanager
def _connect():
    conn = sqlite3.connect(_path())
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    try:
        yield conn
        conn.commit()
    finally:
        conn.close()


def init() -> None:
    """Create the schema if it does not exist (safe to call repeatedly)."""
    with _connect() as conn:
        conn.executescript(SCHEMA)


def _row_to_task(row: sqlite3.Row) -> dict:
    return {
        "id": row["id"],
        "title": row["title"],
        "notes": row["notes"],
        "done": bool(row["done"]),
        "created_at": row["created_at"],
    }


def add_task(title: str, notes: str = "") -> dict:
    now = datetime.now(timezone.utc).isoformat()
    with _connect() as conn:
        cur = conn.execute(
            "INSERT INTO tasks (title, notes, created_at) VALUES (?, ?, ?)",
            (title, notes, now),
        )
        return get_task(cur.lastrowid, conn)


def get_task(task_id: int, conn: sqlite3.Connection | None = None) -> dict | None:
    if conn is not None:
        row = conn.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
        return _row_to_task(row) if row else None
    with _connect() as own:
        row = own.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
        return _row_to_task(row) if row else None


def list_tasks(status: str = "all") -> list[dict]:
    query = "SELECT * FROM tasks"
    params: tuple = ()
    if status == "open":
        query += " WHERE done = 0"
    elif status == "done":
        query += " WHERE done = 1"
    query += " ORDER BY id"
    with _connect() as conn:
        return [_row_to_task(r) for r in conn.execute(query, params).fetchall()]


def complete_task(task_id: int) -> dict | None:
    with _connect() as conn:
        conn.execute("UPDATE tasks SET done = 1 WHERE id = ?", (task_id,))
        return get_task(task_id, conn)


def delete_task(task_id: int) -> bool:
    with _connect() as conn:
        cur = conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
        return cur.rowcount > 0


def stats() -> dict:
    with _connect() as conn:
        row = conn.execute(
            "SELECT COUNT(*) AS total, "
            "COALESCE(SUM(done), 0) AS done FROM tasks"
        ).fetchone()
    total, done = row["total"], row["done"]
    return {"total": total, "done": done, "open": total - done}

Detailed breakdown

  • _connect is a context manager that opens a connection, sets row_factory = sqlite3.Row (so rows behave like dicts), commits on success, and always closes. Opening per operation avoids sharing a connection across async request threads — the simplest correct model.
  • init runs the schema with CREATE TABLE IF NOT EXISTS, so it is safe to call on every startup. This is your lightweight migration story; for evolving schemas you would add versioned migration scripts here.
  • Every query is parameterized (? placeholders), which prevents SQL injection — never format user input into SQL strings.
  • get_task accepts an optional connection so add_task/complete_task can read back the row inside the same transaction, while callers can also use it standalone.
  • _row_to_task converts a row into a plain, JSON-serializable dict with done as a real bool — the shape the tools return to clients.

Step 4: Write the server

The tools are a thin, validated layer over db.py. db.init() runs at import so the schema exists before the first request. A resource exposes live stats.

Create the file

touch server.py

Add the code: server.py

"""A stateful FastMCP server backed by SQLite.

The tools are a thin layer over `db.py`; all state lives in a SQLite database
file, so tasks persist across restarts and redeploys. A resource exposes live
stats. Errors (e.g. an unknown task id) are surfaced as ToolError.
"""

import os

from fastmcp import FastMCP
from fastmcp.exceptions import ToolError

import db

mcp = FastMCP("macmcp")

db.init()  # ensure the schema exists before serving


@mcp.tool
def add_task(title: str, notes: str = "") -> dict:
    """Create a task and return it (including its new id)."""
    if not title.strip():
        raise ToolError("title must not be empty")
    return db.add_task(title.strip(), notes)


@mcp.tool
def list_tasks(status: str = "all") -> list[dict]:
    """List tasks. status is one of: all, open, done."""
    if status not in ("all", "open", "done"):
        raise ToolError("status must be one of: all, open, done")
    return db.list_tasks(status)


@mcp.tool
def complete_task(task_id: int) -> dict:
    """Mark a task as done and return it."""
    task = db.complete_task(task_id)
    if task is None:
        raise ToolError(f"no task with id {task_id}")
    return task


@mcp.tool
def delete_task(task_id: int) -> dict:
    """Delete a task by id."""
    if not db.delete_task(task_id):
        raise ToolError(f"no task with id {task_id}")
    return {"deleted": task_id}


@mcp.resource("tasks://stats")
def task_stats() -> dict:
    """Live counts of total, open, and done tasks."""
    return db.stats()


def main() -> None:
    if os.environ.get("MCP_TRANSPORT", "stdio").lower() == "http":
        mcp.run(
            transport="http",
            host=os.environ.get("MCP_HOST", "127.0.0.1"),
            port=int(os.environ.get("MCP_PORT", "8000")),
        )
    else:
        mcp.run()  # stdio


if __name__ == "__main__":
    main()

Detailed breakdown

  • db.init() at import guarantees the table exists before any tool runs, including the first request after a fresh deploy.
  • The tools validate then delegate. add_task rejects an empty title, list_tasks rejects an unknown status, and complete_task/delete_task raise ToolError for an unknown id. ToolError is how FastMCP returns a clean, typed error to the client instead of a stack trace.
  • tasks://stats is a resource, not a tool. Clients read it to see live counts without mutating anything — the natural fit for derived, read-only state.
  • The transport switch matches the other articles: stdio for local dev, HTTP for the deployed service.

Step 5: Test the layer, the tools, and persistence

Two test files. test_db.py drives the persistence layer directly; test_tools.py drives the tools through the in-memory client. Both isolate the database under pytest’s tmp_path, and one test proves state survives a “restart.”

Create the files

mkdir -p tests
touch tests/__init__.py tests/test_db.py tests/test_tools.py pytest.ini

Add the code: pytest.ini

[pytest]
asyncio_mode = auto

Add the code: tests/test_db.py

"""Test the SQLite persistence layer against an isolated database file."""

import importlib

import pytest


@pytest.fixture(autouse=True)
def isolated_db(tmp_path, monkeypatch):
    monkeypatch.setenv("DB_PATH", str(tmp_path / "tasks.db"))
    import db

    importlib.reload(db)  # pick up the new DB_PATH cleanly
    db.init()
    return db


def test_add_and_get(isolated_db):
    db = isolated_db
    task = db.add_task("write article", "about sqlite")
    assert task["id"] == 1
    assert task["title"] == "write article"
    assert task["done"] is False
    assert db.get_task(1) == task


def test_list_filters_by_status(isolated_db):
    db = isolated_db
    db.add_task("a")
    db.add_task("b")
    db.complete_task(1)
    assert [t["id"] for t in db.list_tasks("all")] == [1, 2]
    assert [t["id"] for t in db.list_tasks("open")] == [2]
    assert [t["id"] for t in db.list_tasks("done")] == [1]


def test_complete_and_delete(isolated_db):
    db = isolated_db
    db.add_task("a")
    assert db.complete_task(1)["done"] is True
    assert db.delete_task(1) is True
    assert db.get_task(1) is None
    assert db.delete_task(999) is False


def test_stats(isolated_db):
    db = isolated_db
    db.add_task("a")
    db.add_task("b")
    db.complete_task(1)
    assert db.stats() == {"total": 2, "done": 1, "open": 1}


def test_state_persists_across_connections(isolated_db, tmp_path):
    """Data written by one connection is visible to a brand-new one."""
    db = isolated_db
    db.add_task("survives")
    # A fresh reload simulates a server restart pointing at the same file.
    import importlib

    importlib.reload(db)
    assert db.get_task(1)["title"] == "survives"

Add the code: tests/test_tools.py

"""Test the tools in-memory against an isolated database."""

import importlib

import pytest
from fastmcp import Client
from fastmcp.exceptions import ToolError


@pytest.fixture()
def mcp(tmp_path, monkeypatch):
    monkeypatch.setenv("DB_PATH", str(tmp_path / "tasks.db"))
    import db
    import server

    importlib.reload(db)
    importlib.reload(server)  # rebinds the tools to the reloaded db + fresh schema
    return server.mcp


async def test_add_then_list(mcp):
    async with Client(mcp) as client:
        created = (await client.call_tool("add_task", {"title": "ship it"})).data
        assert created["id"] == 1
        listed = (await client.call_tool("list_tasks", {})).data
        assert [t["title"] for t in listed] == ["ship it"]


async def test_complete_updates_stats_resource(mcp):
    async with Client(mcp) as client:
        await client.call_tool("add_task", {"title": "a"})
        await client.call_tool("add_task", {"title": "b"})
        await client.call_tool("complete_task", {"task_id": 1})
        stats = (await client.read_resource("tasks://stats"))[0].text
        assert '"done": 1' in stats and '"open": 1' in stats


async def test_delete(mcp):
    async with Client(mcp) as client:
        await client.call_tool("add_task", {"title": "temp"})
        result = (await client.call_tool("delete_task", {"task_id": 1})).data
        assert result == {"deleted": 1}


async def test_unknown_id_raises(mcp):
    async with Client(mcp) as client:
        with pytest.raises(ToolError):
            await client.call_tool("complete_task", {"task_id": 999})


async def test_empty_title_rejected(mcp):
    async with Client(mcp) as client:
        with pytest.raises(ToolError):
            await client.call_tool("add_task", {"title": "   "})

Detailed breakdown

  • isolated_db / mcp fixtures set DB_PATH to a temp file and importlib.reload the modules so each test starts from a fresh, empty schema and never touches your real tasks.db. Reloading server rebinds its tools to the reloaded db.
  • test_state_persists_across_connections is the point of the whole article: it writes a task, reloads the module (a stand-in for a process restart pointing at the same file), and confirms the task is still there — persistence, proven in a unit test.
  • test_tools.py covers the client-visible contract: CRUD round-trips, the stats resource updating after a mutation, and ToolError for bad input and unknown ids.

Run them:

uv run pytest -v

You should see all tests pass.

Step 6: The Makefile

Wrap development, the database, and the launchd deployment. Plain make prints help.

Create the file

touch Makefile

Add the code: Makefile

.DEFAULT_GOAL := help

# --- Deploy configuration (launchd user agent) --------------------------------
LABEL   := com.macmcp.tasks
PLIST   := $(HOME)/Library/LaunchAgents/$(LABEL).plist
UV      := $(shell command -v uv)
DIR     := $(shell pwd)
DOMAIN  := gui/$(shell id -u)
DB      := tasks.db

.PHONY: help install serve test demo reset-db deploy undeploy status logs clean

help:  ## Show this help screen
	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \
		| awk 'BEGIN {FS = ":.*?## "}; {printf "  \033[36m%-10s\033[0m %s\n", $$1, $$2}'

install:  ## Sync runtime and dev dependencies
	uv sync

serve:  ## Run the HTTP server in the foreground (Ctrl-C to stop)
	MCP_TRANSPORT=http uv run python server.py

test:  ## Run the test suite
	uv run pytest -v

demo:  ## Add sample tasks against a running server and show stats
	uv run python scripts/demo.py

reset-db:  ## Delete the database file (wipes all tasks)
	rm -f $(DB) $(DB)-journal $(DB)-wal $(DB)-shm
	@echo "Removed $(DB)."

deploy:  ## Render the launchd plist and start the service
	@mkdir -p logs
	@sed -e 's#@UV@#$(UV)#g' \
	     -e 's#@DIR@#$(DIR)#g' \
	     -e 's#@PATH@#$(dir $(UV)):/usr/bin:/bin#g' \
	     deploy/$(LABEL).plist.template > $(PLIST)
	launchctl bootout $(DOMAIN) $(PLIST) 2>/dev/null || true
	launchctl bootstrap $(DOMAIN) $(PLIST)
	@echo "Deployed $(LABEL). Check: make status"

undeploy:  ## Stop the service and remove the launchd plist
	launchctl bootout $(DOMAIN) $(PLIST) 2>/dev/null || true
	rm -f $(PLIST)
	@echo "Removed $(LABEL)."

status:  ## Show whether the service is registered and running
	@launchctl print $(DOMAIN)/$(LABEL) 2>/dev/null \
		| grep -E 'state =|pid =' || echo "$(LABEL) is not loaded."

logs:  ## Tail the service log files
	@tail -n 40 -f logs/server.out.log logs/server.err.log

clean:  ## Remove caches and logs (keeps the database)
	rm -rf .pytest_cache logs __pycache__ tests/__pycache__

Detailed breakdown

  • reset-db deletes the database (and its journal/WAL sidecars) when you want a clean slate. clean deliberately keeps the database — cleaning caches should never wipe your data.
  • deploy bakes an absolute DB_PATH into the plist (Step 7), so the service always reads the same file regardless of its working directory. This is what makes state stable across redeploys.
  • demo exercises the tools against a running server. The launchd lifecycle matches the companion deploy article.

Step 7: Deploy as a launchd service

The plist runs the server over HTTP with an absolute DB_PATH.

Create the files

mkdir -p deploy scripts
touch deploy/com.macmcp.tasks.plist.template scripts/demo.py

Add the code: deploy/com.macmcp.tasks.plist.template

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.macmcp.tasks</string>

    <key>ProgramArguments</key>
    <array>
        <string>@UV@</string>
        <string>run</string>
        <string>python</string>
        <string>server.py</string>
    </array>

    <key>WorkingDirectory</key>
    <string>@DIR@</string>

    <key>EnvironmentVariables</key>
    <dict>
        <key>MCP_TRANSPORT</key>
        <string>http</string>
        <key>MCP_HOST</key>
        <string>127.0.0.1</string>
        <key>MCP_PORT</key>
        <string>8000</string>
        <key>DB_PATH</key>
        <string>@DIR@/tasks.db</string>
        <key>PATH</key>
        <string>@PATH@</string>
    </dict>

    <key>RunAtLoad</key>
    <true/>

    <key>KeepAlive</key>
    <true/>

    <key>StandardOutPath</key>
    <string>@DIR@/logs/server.out.log</string>

    <key>StandardErrorPath</key>
    <string>@DIR@/logs/server.err.log</string>
</dict>
</plist>

Add the code: scripts/demo.py

"""Live demo: add a few tasks, complete one, and print the stats resource."""

import asyncio
import os

from fastmcp import Client

URL = os.environ.get("MCP_URL", "http://127.0.0.1:8000/mcp/")


async def main() -> None:
    async with Client(URL) as client:
        for title in ("write article", "review PR", "ship release"):
            created = (await client.call_tool("add_task", {"title": title})).data
            print("added:", created["id"], created["title"])
        await client.call_tool("complete_task", {"task_id": 1})
        stats = (await client.read_resource("tasks://stats"))[0].text
        print("stats:", stats)
        open_tasks = (await client.call_tool("list_tasks", {"status": "open"})).data
        print("open :", [t["title"] for t in open_tasks])


if __name__ == "__main__":
    asyncio.run(main())

Detailed breakdown

  • DB_PATH=@DIR@/tasks.db is the key line: an absolute path means the supervised service reads and writes one stable database, so tasks added today are still there after a redeploy or a reboot (RunAtLoad + KeepAlive).
  • demo.py adds a few tasks, completes one, and reads tasks://stats — a quick way to see the server working over HTTP.

Deploy and try it:

make deploy
make status         # state = running, a live pid
make demo
# added: 1 write article
# added: 2 review PR
# added: 3 ship release
# stats: {"total": 3, "done": 1, "open": 2}
# open : ['review PR', 'ship release']

Now prove persistence: redeploy (which restarts the process) and the tasks are still there.

make deploy
# ...then list tasks via a client — all three are still present, id 1 still done.
make undeploy

Troubleshooting

  • Tasks vanish after a restart. The service is not reading the file you think. Confirm the plist’s DB_PATH is an absolute path (it is, via make deploy), and that you did not run make reset-db.
  • database is locked under load. SQLite serializes writers. For a write-heavy server, enable WAL mode (PRAGMA journal_mode=WAL) in init(), or move to a client/server database. For typical MCP tool traffic the default is fine.
  • A tool raises no task with id N. That id does not exist (or was deleted). Call list_tasks to see valid ids.
  • Tests interfere with each other or with tasks.db. Each test sets DB_PATH to a temp file and reloads the modules; if you add tests, keep the isolated_db/mcp fixture so nothing writes the real database.
  • Schema changes don’t take effect. CREATE TABLE IF NOT EXISTS won’t alter an existing table. Add a migration (an ALTER TABLE, versioned) in init(), or make reset-db in development.

Recap

You built a stateful FastMCP server whose data lives in SQLite:

  • db.py isolates persistence: schema init plus parameterized CRUD, with a short-lived connection per operation.
  • server.py exposes validated tools and a live tasks://stats resource, raising ToolError on bad input.
  • The tests isolate the database per test and prove state survives a restart.
  • launchd + an absolute DB_PATH keep the data stable across redeploys and reboots, all driven by uv and make.

Next improvements

  • Enable WAL mode and add indexes as the table grows.
  • Add versioned migrations in init() for schema evolution.
  • Add a search_tasks(query) tool using SQLite LIKE or FTS5 full-text search.
  • Swap SQLite for Postgres (via psycopg) if you need concurrent writers or multiple server instances sharing state.