Build an Engrim SQLite Memory MCP Server: Local-First Persistent Context for AI CLIs [2026]

Engrim is a universal, local-first SQLite memory engine for AI CLIs that hit 91 Hacker News points. It gives any command-line AI tool persistent memory across sessions by storing facts, conversation summaries, and tool-call patterns in a local SQLite database with a simple key-value API. The key design decision is that the memory engine is protocol-agnostic: it does not care whether the client is Claude Code, Codex, Cursor, or a custom script — any tool that can read and write JSON can use Engrim.

• Universal memory protocol: Engrim exposes a simple JSON-over-stdio interface: {"action": "remember", "key": "user_name", "value": "Alice"}. Any CLI tool can pipe JSON to Engrim and get persistent memory for free.
• SQLite under the hood: The database is a single .engrim.db file in the user's home directory. No server, no daemon, no cloud sync. The file is standard SQLite, inspectable with any SQLite browser.
• Automatic summarization: When memory exceeds a configurable size (default 1 MB), Engrim runs a local LLM summarization step that compresses older entries into a summary, preserving the essential information while shedding detail.
• Time-decaying importance: Each memory entry has a half-life (default 24 hours). Engrim automatically decays the importance of entries older than their half-life, so the agent naturally forgets stale context without explicit deletion commands.

Architecture: The Memory Engine

+------------------------------------------------------------------+
|  Engrim SQLite Memory Engine (91 HN points)                      |
|                                                                  |
|  CLI Tool --> JSON/stdin --> Engrim Core --> SQLite .engrim.db   |
|       |                        |                        |         |
|       v                        v                        v         |
|  remember(key, val)       Importance Decay          FTS Search   |
|  recall(query)            Summarization             Integrity    |
|  search(term)             Half-life: 24h            ACID         |
+------------------------------------------------------------------+

Step 1: Install & Use

# Install
pip install engrim-memory

# Use directly
engrim remember user_name "Alice"
engrim remember project_context "Building a FastMCP server for PostgreSQL"
engrim recall user_name
# > Alice

# Pipe from any CLI tool
echo '{"action": "remember", "key": "last_branch", "value": "feature/engrim-integration"}' | engrim --json

Step 2: File 1 — Core Engine (engrim_core.py)

import sqlite3
import json
import time
import hashlib
from pathlib import Path
from datetime import datetime, timedelta

class EngrimCore:
    """Local-first SQLite memory engine for AI CLIs."""

    def __init__(self, db_path: str = "~/.engrim.db", max_size_mb: int = 1):
        self.db_path = Path(db_path).expanduser()
        self.max_size = max_size_mb * 1024 * 1024
        self.conn = sqlite3.connect(str(self.db_path))
        self.conn.execute("PRAGMA journal_mode=WAL")
        self._init_tables()

    def _init_tables(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS memory (
                key TEXT NOT NULL,
                value TEXT NOT NULL,
                importance REAL DEFAULT 1.0,
                created_at REAL,
                half_life_hours REAL DEFAULT 24.0,
                tags TEXT DEFAULT '[]',
                PRIMARY KEY (key)
            )
        """)
        self.conn.execute("""
            CREATE VIRTUAL TABLE IF NOT EXISTS memory_fts
            USING fts5(key, value, content=memory)
        """)
        self.conn.commit()

    def remember(self, key: str, value: str, importance: float = 1.0,
                 half_life: float = 24.0, tags: list[str] = None):
        """Store a memory with time-decaying importance."""
        now = time.time()
        self.conn.execute(
            """INSERT OR REPLACE INTO memory
               (key, value, importance, created_at, half_life_hours, tags)
               VALUES (?, ?, ?, ?, ?, ?)""",
            (key, value, importance, now, half_life,
             json.dumps(tags or []))
        )
        self.conn.execute(
            "INSERT OR REPLACE INTO memory_fts(rowid, key, value) VALUES (?, ?, ?)",
            (self.conn.execute("SELECT rowid FROM memory WHERE key=?", (key,)).fetchone()[0],
             key, value)
        )
        self.conn.commit()
        self._maybe_compact()

    def recall(self, key: str) -> str | None:
        """Recall a memory by exact key, applying time decay."""
        row = self.conn.execute(
            "SELECT value, importance, created_at, half_life_hours FROM memory WHERE key=?",
            (key,)
        ).fetchone()
        if not row:
            return None
        value, importance, created_at, half_life = row
        decayed = self._decay_importance(importance, created_at, half_life)
        if decayed < 0.1:
            return None  # Effectively forgotten
        return value

    def search(self, query: str, limit: int = 5) -> list[dict]:
        """Full-text search across all memories."""
        rows = self.conn.execute(
            """SELECT m.key, m.value, m.importance, m.created_at, m.half_life_hours
               FROM memory_fts f JOIN memory m ON f.rowid = m.rowid
               WHERE memory_fts MATCH ?
               ORDER BY rank LIMIT ?""",
            (query, limit)
        ).fetchall()
        return [
            {"key": r[0], "value": r[1],
             "importance": self._decay_importance(r[2], r[3], r[4])}
            for r in rows
        ]

    def _decay_importance(self, importance: float, created_at: float,
                          half_life: float) -> float:
        elapsed = (time.time() - created_at) / 3600  # hours
        return importance * (0.5 ** (elapsed / half_life))

    def _maybe_compact(self):
        """Check size and trigger summarization if over limit."""
        size = Path(self.db_path).stat().st_size
        if size > self.max_size:
            self._summarize_oldest()

    def _summarize_oldest(self):
        """Summarize the oldest 20% of entries into a single summary."""
        rows = self.conn.execute(
            "SELECT key, value FROM memory ORDER BY importance ASC, created_at ASC LIMIT 20"
        ).fetchall()
        if rows:
            combined = " | ".join(f"{k}: {v[:200]}" for k, v in rows)
            summary_key = f"summary_{int(time.time())}"
            self.remember(summary_key, f"[AUTO-SUMMARY] {combined[:1000]}",
                          importance=0.5, half_life=48.0)
            for k, _ in rows:
                self.conn.execute("DELETE FROM memory WHERE key=?", (k,))
            self.conn.commit()

Step 3: File 2 — MCP Server (engrim_mcp.py)

from fastmcp import FastMCP
from engrim_core import EngrimCore
import json

mcp = FastMCP("engrim")
memory = EngrimCore()

@mcp.tool()
def remember(key: str, value: str, importance: float = 1.0,
             half_life: float = 24.0) -> str:
    """Store a memory with time-decaying importance."""
    memory.remember(key, value, importance, half_life)
    return json.dumps({"status": "stored", "key": key, "decay_hours": half_life})

@mcp.tool()
def recall(key: str) -> str:
    """Recall a memory by exact key."""
    result = memory.recall(key)
    if result is None:
        return json.dumps({"status": "not_found", "key": key})
    return json.dumps({"status": "found", "key": key, "value": result})

@mcp.tool()
def search(query: str, limit: int = 5) -> str:
    """Full-text search across all memories."""
    results = memory.search(query, limit)
    return json.dumps({"status": "ok", "results": results}, default=str)

@mcp.tool()
def stats() -> str:
    """Return memory engine statistics."""
    count = memory.conn.execute("SELECT COUNT(*) FROM memory").fetchone()[0]
    size = Path(memory.db_path).stat().st_size
    return json.dumps({"entries": count, "size_bytes": size, "size_mb": round(size/1e6, 2)})

if __name__ == "__main__":
    mcp.run(transport="stdio")

Step 4: File 3 — Config (.engrimrc)

{
  "db_path": "~/.engrim.db",
  "max_size_mb": 1,
  "default_importance": 1.0,
  "default_half_life_hours": 24,
  "summarization": {
    "enabled": true,
    "llm": "local",
    "model": "qwen3.8-27b-4bit",
    "prompt": "Summarize these memories into a concise entry"
  },
  "tags": {
    "auto_tag": true,
    "extract_entities": true
  }
}

Benchmark: Memory Operations

Production Reality Check

Local-first memory engines have three sharp edges:

1. Simultaneous write conflicts: Multiple agent processes writing to the same .engrim.db file can cause SQLITE_BUSY errors. WAL mode helps, but for multi-process deployments, add a lightweight lock file (~/.engrim.lock) that uses file-system-level advisory locking. Our OneCLI sandbox uses the same lock-file pattern for its audit trail.

2. Half-life decay causes silent context loss: The default 24-hour half-life means a memory entered at the start of a week-long project is 1/128th of its original importance by day 7. The agent may act as if it never learned the fact. Set per-project half-lives explicitly: project context = 168 hours (7 days), user preferences = 720 hours (30 days), temporary state = 1 hour. The Rowboat local-first agent uses a similar tiered importance system for its session DAG.

3. Summarization quality depends on the local LLM: If the local summarization model is too small (e.g., Qwen2.5-1.5B), it produces lossy summaries that drop critical details. Use a 7B+ local model for summarization, or route summarization through the cloud via the routing pattern from the Smart Model Router MCP Server.

Explore more MCP tools in the MCP Server Directory or pair Engrim with AI agent workflows for persistent context across sessions.

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Last tested & verified: September 2026 with Python 3.12, SQLite 3.46, FastMCP 4.0.