The JSON failure library worked great — until my agents started running on multiple machines. Suddenly, crash fixes logged on one server were invisible to agents on another. Worse, concurrent writes from parallel agents corrupted the JSON file.
I needed a real database. Here's how I upgraded to SQLite without rewriting the whole library.
SQLite is the perfect upgrade path. It's a single file, supports concurrent readers, and handles writes with proper locking. Here's the migration:
import sqlite3
import time
import platform
from pathlib import Path
from typing import Optional
DB_PATH = Path.home() / ".mcp" / "failures.db"
def init_db():
"""Create the failures table if it doesn't exist."""
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(DB_PATH), timeout=30)
conn.execute("""
CREATE TABLE IF NOT EXISTS failures (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task TEXT NOT NULL,
error TEXT NOT NULL,
attempted_fix TEXT NOT NULL,
result TEXT NOT NULL,
timestamp REAL NOT NULL,
env TEXT
)
""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_task_result ON failures(task, result)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON failures(timestamp)")
conn.commit()
conn.close()
def log_failure(task: str, error: str, attempted_fix: str, result: str, env: str = None):
"""Log a failure to the SQLite database."""
conn = sqlite3.connect(str(DB_PATH), timeout=30)
conn.execute(
"INSERT INTO failures (task, error, attempted_fix, result, timestamp, env) VALUES (?, ?, ?, ?, ?, ?)",
(task, error, attempted_fix, result, time.time(), env)
)
conn.commit()
conn.close()
def check_failure_library(task: str, env: str = None) -> Optional[dict]:
"""Check if this task has a known failure with a verified fix."""
conn = sqlite3.connect(str(DB_PATH), timeout=30)
conn.row_factory = sqlite3.Row
query = "SELECT * FROM failures WHERE task = ? AND result = 'fixed'"
params = [task]
if env:
query += " AND (env = ? OR env IS NULL)"
params.append(env)
query += " ORDER BY timestamp DESC LIMIT 1"
row = conn.execute(query, params).fetchone()
conn.close()
return dict(row) if row else None
The key changes from the JSON version:
idx_task_result
makes the lookup fast even with thousands of entriestimeout=30
prevents "database is locked" errors on WindowsSQLite solves three problems the JSON library couldn't:
Concurrent writes — SQLite uses file-level locking. Multiple agents can read simultaneously, and writes are serialized automatically. No more corrupted JSON from race conditions. On Windows, SQLite uses msvcrt
internally for file locking; on Unix, it uses fcntl
. You don't need to implement platform-specific locking yourself.
Cross-machine sharing — Put the SQLite file on a shared network drive (NFS, SMB), and every agent on every machine reads the same failure library. No sync needed. This is the killer feature: a fix logged on your CI server is immediately available to agents running on your laptop.
Query flexibility — With SQL, you can filter by environment, date range, or result type. The JSON library required the entire file into memory and filtering in Python. With 10,000+ entries, that difference is measurable.
Moving from JSON to SQLite is a one-time migration. Here's how I did it without downtime:
def migrate_json_to_sqlite(json_path: Path, db_path: Path):
"""One-time migration from JSON failures to SQLite."""
if not json_path.exists():
init_db()
return
import json
init_db()
failures = json.loads(json_path.read_text())
conn = sqlite3.connect(str(db_path), timeout=30)
for f in failures:
conn.execute(
"INSERT INTO failures (task, error, attempted_fix, result, timestamp, env) VALUES (?, ?, ?, ?, ?, ?)",
(f["task"], f["error"], f["attempted_fix"], f["result"], f.get("timestamp", time.time()), f.get("env"))
)
conn.commit()
conn.close()
json_path.rename(json_path.with_suffix(".json.bak"))
Run this once at startup. After migration, the JSON file is backed up but no longer used.
Here's the updated MCPToolHandler with SQLite integration:
class MCPToolHandler:
def __init__(self):
init_db()
self.env = f"{platform.system()}|{platform.machine()}"
async def execute_tool(self, tool_name: str, args: dict):
known_failure = check_failure_library(tool_name, env=self.env)
if known_failure:
args = self._apply_fix(tool_name, args, known_failure)
else:
args = args # No known failure, use original args
try:
result = await self._call_tool(tool_name, args)
return result
except Exception as e:
attempted_fix = self._suggest_fix(tool_name, e)
log_failure(tool_name, str(e), attempted_fix, "pending", env=self.env)
try:
fixed_args = self._apply_fix(tool_name, args, known_failure)
result = await self._call_tool(tool_name, fixed_args)
log_failure(tool_name, str(e), attempted_fix, "fixed", env=self.env)
return result
except Exception as e2:
log_failure(tool_name, str(e), attempted_fix, "failed", env=self.env)
raise
The guard if known_failure:
prevents passing None
to _apply_fix
. The log_failure
call only marks "fixed"
after the retry succeeds — never before verification.
conn = sqlite3.connect(str(DB_PATH), timeout=30)
conn.execute("PRAGMA journal_mode=WAL")
Network drives: SQLite works on NFS/SMB but can be slow. For high-write scenarios, use a proper database server (PostgreSQL) instead. The timeout=30
parameter is critical on network filesystems where lock acquisition can be delayed.
File size growth: Failures accumulate. Add a cleanup job:
def prune_old_failures(days=30):
cutoff = time.time() - days * 86400
conn = sqlite3.connect(str(DB_PATH), timeout=30)
conn.execute("DELETE FROM failures WHERE timestamp < ?", (cutoff,))
conn.commit()
conn.close()
Cross-platform locking: SQLite handles file locking internally, but on Windows you may need to set timeout=30
to avoid "database is locked" errors under heavy concurrent writes. SQLite uses msvcrt.locking()
on Windows and fcntl.flock()
on Unix — you don't need to implement this yourself, but the timeout prevents indefinite hangs.
VACUUM for performance: After pruning old entries, reclaim disk space:
conn = sqlite3.connect(str(DB_PATH), timeout=30)
conn.execute("VACUUM")
conn.commit()
conn.close()
Have you migrated from file-based to database-based state in your agent infrastructure? What's your approach for cross-machine failure sharing — a shared database, a message queue, or something else entirely?
I'm curious whether SQLite is the right choice for teams with dozens of agents, or if a proper database server becomes necessary at that scale.
Drop a comment below — I read every response.