Память AI-агента: сравниваем файлы, локальное хранилище и MCP-сервис

Память AI-агента: сравниваем файлы, локальное хранилище и MCP-сервис
Temporary fallback cover; replace in editorial pass.

Level: intermediate · Reading time: 40 minutes · Updated: 20 October 2026

Every AI agent eventually hits the same wall: the session ends and everything it learned about your project disappears. The obvious fix — "just save notes" — creates a new problem within weeks: a pile of half-true, outdated notes that the agent dutifully retrieves and trusts. This article builds three minimal agent memory backends — plain files, a local SQLite store and an MCP service — and gives you one test harness to compare them on retrieval accuracy, latency, portability and data control. You run the measurements; we do not hand you numbers we did not produce.

What this article is and is not

Everything below is a recipe. The code is short enough to read in full and was written to be run, but the editor did not run the benchmark for this article, so there is no "winner" table with figures. The results table at the end is a blank template. That is deliberate: memory performance depends heavily on your data, your hardware and how your agent phrases queries, and borrowed numbers would mislead you more than they would help.

What you will have after working through it:

The concrete case

The running example is a coding assistant that works on one repository over many sessions. Between sessions it should keep things like:

The third item is the interesting one. If the agent stored the old host in August and the new host in September, a naive memory returns both, and the agent may pick the wrong one. A memory that does not handle supersession is not just noisy — it is actively harmful. So our test set explicitly checks for it.

Why not put everything in the context window each session? Because it does not scale: a project accumulates hundreds of facts, most irrelevant to any given task, and stuffing them all in costs tokens and dilutes attention. The point of memory is selective recall.

Design rules that apply to all three backends

Before writing any storage code, fix the data model. Most memory rot comes from a bad model, not a bad database.

  1. One fact per record. A record about "test commands" should not also contain "PR style". Mixed records match too many queries and are hard to update.
  2. Every record has a stable key. Writing a record with an existing key replaces it. This is the whole staleness strategy in one line: new facts overwrite old ones instead of piling up next to them.
  3. Every record has updated. An ISO date. It lets you audit, expire and tie-break.
  4. Optional expires. For facts you know are temporary ("release freeze until Friday"). Expired records are filtered at read time.
  5. Records are written by deliberate tool calls, not by logging whole transcripts. Transcript dumps are the fastest way to an unusable archive.

The shared record shape, as JSON Lines:

{"key": "test-command", "topic": "testing", "body": "Run tests with `make test`; the Makefile sets DJANGO_SETTINGS_MODULE.", "updated": "2026-08-14", "expires": null}

Step 0: project layout and the test dataset

mkdir agent-memory-bench && cd agent-memory-bench
python3 -m venv .venv && . .venv/bin/activate
pip install "mcp[cli]"
mkdir data

Python 3.10 or newer is assumed. The file and SQLite backends use only the standard library; the mcp package is needed only for the third backend.

Create data/memories.jsonl. The records below are invented examples for the walkthrough; replace them with real (non-secret) facts from your own project before drawing conclusions. Note that staging-db appears twice — that is the staleness trap.

{"key": "test-command", "topic": "testing", "body": "Run tests with make test; the Makefile sets DJANGO_SETTINGS_MODULE.", "updated": "2026-08-14", "expires": null}
{"key": "pr-style", "topic": "workflow", "body": "User prefers small pull requests with one concern each.", "updated": "2026-08-02", "expires": null}
{"key": "staging-db", "topic": "infrastructure", "body": "Staging database host is db-a.internal.example.", "updated": "2026-08-10", "expires": null}
{"key": "lint", "topic": "testing", "body": "Linting uses ruff; run make lint before committing.", "updated": "2026-08-20", "expires": null}
{"key": "migrations", "topic": "database", "body": "Never edit an applied migration; create a new one instead.", "updated": "2026-08-21", "expires": null}
{"key": "freeze", "topic": "release", "body": "Release freeze: no merges to main until the freeze is lifted.", "updated": "2026-09-05", "expires": "2026-09-12"}
{"key": "logging", "topic": "code-style", "body": "Use the structlog logger, not print, in application code.", "updated": "2026-08-25", "expires": null}
{"key": "ci", "topic": "workflow", "body": "CI runs on every push; the slow integration suite runs nightly only.", "updated": "2026-08-30", "expires": null}
{"key": "staging-db", "topic": "infrastructure", "body": "Staging database host moved to db-b.internal.example.", "updated": "2026-09-01", "expires": null}
{"key": "secrets", "topic": "security", "body": "Secrets live in the vault; never write them into memory or code.", "updated": "2026-09-03", "expires": null}

Create data/queries.jsonl. Each query lists the key that should come back. Queries with "must_contain" also check the body text, which is how we detect a stale answer: the right key with the wrong content is a staleness error. Queries are phrased the way an agent would phrase them, not copied from the records — otherwise every backend looks perfect.

{"q": "how do I run the test suite", "expected": "test-command"}
{"q": "what size should pull requests be", "expected": "pr-style"}
{"q": "staging database hostname", "expected": "staging-db", "must_contain": "db-b"}
{"q": "which linter do we use", "expected": "lint"}
{"q": "can I modify an old migration", "expected": "migrations"}
{"q": "is there a release freeze right now", "expected": null}
{"q": "how should I log errors", "expected": "logging"}
{"q": "when does the integration suite run", "expected": "ci"}
{"q": "where are API keys stored", "expected": "secrets"}

The freeze query expects null: assuming you run the benchmark after 2026-09-12, the freeze record has expired and returning it counts as a staleness error. Adjust dates if you run it earlier.

Step 1: the shared interface and scoring

All three backends implement the same two methods. Create memory_api.py:

import datetime as dt
import re
from typing import Protocol

TOKEN = re.compile(r"[a-z0-9]+")
STOP = {"the", "a", "an", "do", "i", "we", "is", "are", "how", "what",
        "which", "where", "when", "should", "can", "to", "of", "in", "on",
        "there", "right", "now", "use", "my", "our"}


def tokens(text: str) -> list[str]:
    return [t for t in TOKEN.findall(text.lower()) if t not in STOP]


def is_live(record: dict, today: dt.date | None = None) -> bool:
    exp = record.get("expires")
    if not exp:
        return True
    today = today or dt.date.today()
    return dt.date.fromisoformat(exp) >= today


class Memory(Protocol):
    def remember(self, record: dict) -> None: ...
    def recall(self, query: str, k: int = 3) -> list[dict]: ...

The stop-word list is crude on purpose. It is the same for every backend, so it does not bias the comparison, and you can see exactly what it does.

Step 2: backend A — plain Markdown files

The simplest memory: one Markdown file per key in a directory. You can open it in any editor, commit it to git and review it in a pull request. Retrieval is a keyword-overlap score over all files.

Create backend_files.py:

import json
import pathlib
import re

from memory_api import is_live, tokens

SAFE_KEY = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$")


class FileMemory:
    def __init__(self, root: str = "store_files"):
        self.root = pathlib.Path(root)
        self.root.mkdir(exist_ok=True)

    def _path(self, key: str) -> pathlib.Path:
        if not SAFE_KEY.match(key):
            raise ValueError(f"unsafe key: {key!r}")
        return self.root / f"{key}.md"

    def remember(self, record: dict) -> None:
        meta = {k: record.get(k) for k in ("key", "topic", "updated", "expires")}
        text = "---\n" + json.dumps(meta) + "\n---\n" + record["body"] + "\n"
        self._path(record["key"]).write_text(text, encoding="utf-8")

    def _load(self, path: pathlib.Path) -> dict:
        _, meta, body = path.read_text(encoding="utf-8").split("---\n", 2)
        rec = json.loads(meta)
        rec["body"] = body.strip()
        return rec

    def recall(self, query: str, k: int = 3) -> list[dict]:
        q = set(tokens(query))
        scored = []
        for path in self.root.glob("*.md"):
            rec = self._load(path)
            if not is_live(rec):
                continue
            doc = set(tokens(rec["topic"] + " " + rec["body"] + " " + rec["key"]))
            score = len(q & doc)
            if score:
                scored.append((score, rec["updated"], rec))
        scored.sort(key=lambda s: (s[0], s[1]), reverse=True)
        return [rec for _, _, rec in scored[:k]]

Things to notice:

Step 3: backend B — local SQLite with FTS5

The second backend keeps everything in one SQLite file and uses the FTS5 extension for full-text search with BM25 ranking. It is still local, still a single file you can copy, but search no longer scans everything in Python.

Check that your Python's SQLite build includes FTS5:

python3 -c "import sqlite3; c=sqlite3.connect(':memory:'); c.execute('create virtual table t using fts5(x)'); print('fts5 ok')"

If that raises no such module: fts5, your SQLite was built without it; install a Python build that bundles a recent SQLite before continuing.

Create backend_sqlite.py:

import sqlite3

from memory_api import is_live, tokens


class SqliteMemory:
    def __init__(self, path: str = "store.db"):
        self.con = sqlite3.connect(path)
        self.con.execute(
            "CREATE TABLE IF NOT EXISTS records ("
            " key TEXT PRIMARY KEY, topic TEXT, body TEXT,"
            " updated TEXT, expires TEXT)"
        )
        self.con.execute(
            "CREATE VIRTUAL TABLE IF NOT EXISTS records_fts USING fts5("
            " key, topic, body, content='records', content_rowid='rowid')"
        )
        self.con.commit()

    def remember(self, record: dict) -> None:
        cur = self.con.cursor()
        old = cur.execute(
            "SELECT rowid, key, topic, body FROM records WHERE key = ?",
            (record["key"],),
        ).fetchone()
        if old:
            cur.execute(
                "INSERT INTO records_fts(records_fts, rowid, key, topic, body)"
                " VALUES ('delete', ?, ?, ?, ?)", old,
            )
            cur.execute("DELETE FROM records WHERE key = ?", (record["key"],))
        cur.execute(
            "INSERT INTO records(key, topic, body, updated, expires)"
            " VALUES (?, ?, ?, ?, ?)",
            (record["key"], record["topic"], record["body"],
             record["updated"], record.get("expires")),
        )
        cur.execute(
            "INSERT INTO records_fts(rowid, key, topic, body)"
            " VALUES (?, ?, ?, ?)",
            (cur.lastrowid, record["key"], record["topic"], record["body"]),
        )
        self.con.commit()

    def recall(self, query: str, k: int = 3) -> list[dict]:
        toks = tokens(query)
        if not toks:
            return []
        match = " OR ".join(f'"{t}"' for t in toks)
        rows = self.con.execute(
            "SELECT r.key, r.topic, r.body, r.updated, r.expires"
            " FROM records_fts f JOIN records r ON r.rowid = f.rowid"
            " WHERE records_fts MATCH ?"
            " ORDER BY bm25(records_fts) LIMIT ?",
            (match, k * 3),
        ).fetchall()
        cols = ("key", "topic", "body", "updated", "expires")
        recs = [dict(zip(cols, row)) for row in rows]
        return [r for r in recs if is_live(r)][:k]

Notes on the choices:

Step 4: backend C — an MCP memory service

The third option exposes memory through the Model Context Protocol (MCP). The agent no longer imports your Python class; it calls remember and recall as tools on a separate server. That separation is the whole point: any MCP-capable client — a different agent framework, a desktop app, a teammate's setup — can use the same memory.

To keep the comparison honest, the MCP server below wraps the same SQLite backend. Any difference you measure between B and C is therefore the cost (or benefit) of the protocol and process boundary, not of a different search algorithm.

Create mcp_memory_server.py:

from mcp.server.fastmcp import FastMCP

from backend_sqlite import SqliteMemory

mcp = FastMCP("agent-memory")
store = SqliteMemory("store_mcp.db")


@mcp.tool()
def remember(key: str, topic: str, body: str, updated: str,
             expires: str | None = None) -> str:
    """Store one fact. Writing an existing key replaces the old fact."""
    store.remember({"key": key, "topic": topic, "body": body,
                    "updated": updated, "expires": expires})
    return f"stored {key}"


@mcp.tool()
def recall(query: str, k: int = 3) -> list[dict]:
    """Return up to k live facts relevant to the query."""
    return store.recall(query, k)


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

mcp.run() defaults to the stdio transport: the client starts the server as a subprocess and talks to it over standard input and output. Nothing listens on the network. You can inspect the tools interactively with the MCP inspector shipped in the mcp[cli] extra:

mcp dev mcp_memory_server.py

For the harness we need a programmatic client. Create backend_mcp.py, which adapts the MCP tools to our Memory interface:

import asyncio
import json

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


class McpMemory:
    """Synchronous adapter over an MCP memory server, for benchmarking."""

    def __init__(self, script: str = "mcp_memory_server.py"):
        self.params = StdioServerParameters(command="python", args=[script])
        self.loop = asyncio.new_event_loop()
        self._ctx = None
        self.session = None
        self.loop.run_until_complete(self._open())

    async def _open(self):
        self._ctx = stdio_client(self.params)
        read, write = await self._ctx.__aenter__()
        self._sess_ctx = ClientSession(read, write)
        self.session = await self._sess_ctx.__aenter__()
        await self.session.initialize()

    def close(self):
        async def _close():
            await self._sess_ctx.__aexit__(None, None, None)
            await self._ctx.__aexit__(None, None, None)
        self.loop.run_until_complete(_close())
        self.loop.close()

    def _call(self, name: str, args: dict):
        return self.loop.run_until_complete(self.session.call_tool(name, args))

    def remember(self, record: dict) -> None:
        self._call("remember", record)

    def recall(self, query: str, k: int = 3) -> list[dict]:
        result = self._call("recall", {"query": query, "k": k})
        if result.structuredContent and "result" in result.structuredContent:
            return result.structuredContent["result"]
        out = []
        for item in result.content:
            if getattr(item, "type", None) == "text":
                parsed = json.loads(item.text)
                out.extend(parsed if isinstance(parsed, list) else [parsed])
        return out

Two honest caveats about this adapter. First, how a list return value is serialised (one structured result versus several text items) has varied across versions of the Python MCP SDK, which is why recall handles both shapes; if your version returns something else, print result once and adjust. Second, the session is opened once and reused. If you open a new session per call, you will be measuring process start-up, not memory lookup — sometimes that is the number you care about, but label it as such.

Step 5: the harness

Create bench.py. It loads the dataset into a backend, runs every query several times, and reports accuracy and latency.

import argparse
import json
import pathlib
import shutil
import statistics
import time

QUERIES = [json.loads(l) for l in open("data/queries.jsonl", encoding="utf-8")]
MEMORIES = [json.loads(l) for l in open("data/memories.jsonl", encoding="utf-8")]


def fresh(name: str):
    if name == "files":
        shutil.rmtree("store_files", ignore_errors=True)
        from backend_files import FileMemory
        return FileMemory()
    if name == "sqlite":
        pathlib.Path("store.db").unlink(missing_ok=True)
        from backend_sqlite import SqliteMemory
        return SqliteMemory()
    if name == "mcp":
        pathlib.Path("store_mcp.db").unlink(missing_ok=True)
        from backend_mcp import McpMemory
        return McpMemory()
    raise SystemExit(f"unknown backend {name}")


def pct(values, p):
    values = sorted(values)
    idx = min(len(values) - 1, round(p / 100 * (len(values) - 1)))
    return values[idx]


def run(name: str, repeats: int, k: int):
    mem = fresh(name)
    for rec in MEMORIES:          # file order = write order, so later wins
        mem.remember(rec)

    hits = 0
    rr = []
    stale = 0
    latencies = []
    graded = [q for q in QUERIES if q["expected"]]

    for q in QUERIES:
        for i in range(repeats):
            t0 = time.perf_counter()
            res = mem.recall(q["q"], k)
            latencies.append((time.perf_counter() - t0) * 1000)
        keys = [r["key"] for r in res]

        if q["expected"] is None:
            if any(r.get("expires") for r in res):
                stale += 1
            continue
        if q["expected"] in keys:
            hits += 1
            rr.append(1 / (keys.index(q["expected"]) + 1))
            body = res[keys.index(q["expected"])]["body"]
            if "must_contain" in q and q["must_contain"] not in body:
                stale += 1
        else:
            rr.append(0.0)
            print(f"  MISS [{name}] {q['q']!r} -> {keys}")

    if hasattr(mem, "close"):
        mem.close()

    print(f"{name:7s} recall@{k}={hits}/{len(graded)}"
          f"  MRR={statistics.mean(rr):.2f}"
          f"  stale={stale}"
          f"  p50={pct(latencies, 50):.2f}ms"
          f"  p95={pct(latencies, 95):.2f}ms"
          f"  n={len(latencies)}")


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("backends", nargs="+", choices=["files", "sqlite", "mcp"])
    ap.add_argument("--repeats", type=int, default=50)
    ap.add_argument("-k", type=int, default=3)
    args = ap.parse_args()
    for b in args.backends:
        run(b, args.repeats, args.k)

Run it:

python bench.py files sqlite mcp --repeats 50

What the metrics mean:

Step 6: verification before you trust the numbers

A benchmark that silently tests the wrong thing is worse than no benchmark. Before reading results, check:

  1. Supersession actually happened. After a run:
    cat store_files/staging-db.md
    sqlite3 store.db "SELECT key, body FROM records WHERE key='staging-db';"
    sqlite3 store.db "SELECT count(*) FROM records_fts WHERE records_fts MATCH 'staging';"
    Each should show exactly one record mentioning db-b. If the FTS count is 2, the index still holds the old row — the delete-sync in remember is broken.
  2. The MCP backend is really going through MCP. Temporarily add import sys; print("recall called", file=sys.stderr) inside the server's recall tool. You should see the line in your terminal for each call. If you don't, your adapter is importing the store directly somewhere.
  3. Expiry is evaluated as you expect. Run once with today's date; then temporarily change the freeze record's expires to a future date and confirm the freeze query now does return it (and is therefore counted as a miss against expected: null — that is the correct behaviour for the test as written).
  4. Latency is stable. Run the whole benchmark three times. If p50 swings by more than you are prepared to call a difference, increase --repeats or close other workloads. Report the spread, not the best run.
  5. Misses are inspected, not just counted. The harness prints each miss with what came back instead. Read them. A miss is usually a vocabulary mismatch ("API keys" vs "secrets"), and that tells you more about keyword memory than any aggregate.

Step 7: the dimensions a stopwatch cannot measure

Accuracy and latency come out of the harness. Portability and data control need a checklist. Answer each question for each backend, with a yes/no and one sentence of evidence from your setup.

Portability

Data control

For orientation, here is how the three minimal implementations in this article answer those questions by construction. These are properties of the code above, not measurements:

A: FilesB: SQLite FTS5C: MCP service (stdio, wrapping B)
Human-readableYes, any editorVia sqlite3 CLI or a viewerSame as B, plus via tools
Reviewable in gitYes, line diffsBinary file; diffs are not meaningfulSame as B
Usable by other agent frameworksIf they can read filesIf they can read SQLiteAny MCP client, without your code
Where data livesLocal directoryLocal fileLocal file (stdio); moves wherever the server runs if you switch to a network transport
Precise deletionDelete the file (and git history if committed)Delete the row and the FTS entry; then VACUUM to drop freed pagesNeeds a forget tool, which this minimal server does not have
Audit trailgit log, if committedNone in this versionNone in this version; the server is a natural place to add one
Who can writeAnything with filesystem accessAnything with filesystem accessThe client that launched the server, plus filesystem access to the DB

Step 8: record your results

Fill this in from your own runs. Leave the hardware line in: latency numbers without it are not comparable to anyone else's.

Backendrecall@3MRRStalep50, msp95, msNotes
A: Files—————
B: SQLite FTS5—————
C: MCP (stdio)—————

Hardware / OS / Python / SQLite / mcp versions: ____ · Records: ____ · Repeats: ____ · Runs: ____

What to expect qualitatively, so you can spot a broken run (these are reasoned expectations, not results):

Scaling the test

Ten records tell you about correctness, not scale. To probe latency growth, generate filler records that do not collide with the graded keys, then re-run:

python - <<'EOF'
import json, random
random.seed(0)
words = "alpha beta gamma delta parser cache queue worker deploy schema index token".split()
with open("data/memories.jsonl", "a", encoding="utf-8") as f:
    for i in range(5000):
        body = " ".join(random.choices(words, k=12))
        f.write(json.dumps({"key": f"filler-{i}", "topic": "filler", "body": body,
                            "updated": "2026-08-01", "expires": None}) + "\n")
EOF
python bench.py files sqlite mcp --repeats 20

Keep a copy of the original ten-record file so you can return to the correctness test. Synthetic filler shares no vocabulary with the real queries, so it tests scan cost but under-states ranking difficulty; real project notes overlap far more.

Failure cases you will meet

The stale fact wins

Symptom: stale > 0 for staging-db. Causes: memories were written with a new key each time (e.g. staging-db-2) so nothing was superseded; or in B, the FTS index was not updated on replace. Fix the key discipline first — it is the cause in most real systems.

Vocabulary mismatch

Symptom: "where are API keys stored" misses secrets. Keyword search cannot know that "API keys" are secrets. Options, in order of cost: write a broader topic when saving ("security, secrets, api keys, credentials"); let the agent rewrite the query before recall; add embedding-based search. Only the last changes the architecture, and it brings its own model, index and versioning problems.

The memory becomes a transcript dump

Symptom: hundreds of records, many near-duplicates, recall drowning in partial matches. Cause: the agent was allowed to "remember" freely. Fix: require a key and topic, cap body length, and periodically list all records and have a human (or a supervised agent pass) merge and delete.

Memory as an injection channel

Anything written to memory is replayed into future prompts. If an agent saves text it read from an untrusted web page or email, that text can carry prompt injection into every later session. Mitigations: store facts in the agent's own words rather than verbatim quotes, mark the source of each record, and do not let content from untrusted sources be written without review. The secrets example record is there for a reason: credentials must never be saved to memory, in any backend.

MCP session start-up dominates

Symptom: C's p95 is wildly higher than its p50. Usually the first call includes import and initialisation time. Either warm up with one discarded call before timing, or report cold-start separately — both are legitimate numbers, they just answer different questions.

FTS5 query errors

Symptom: sqlite3.OperationalError: fts5: syntax error. Raw query text reached MATCH. The tokens()-and-quote step in recall prevents this; if you change it, keep the quoting.

Limitations of this comparison

How to choose

Once your table is filled in, a reasonable default reading is:

Whichever you pick, the data model from the start of this article — one fact per record, stable keys that supersede, dates, expiry, deliberate writes — matters more than the storage engine. A well-keyed folder of Markdown files will outperform a sophisticated service full of duplicates.

Next steps

More step-by-step material is collected in the guides, and every term used above is defined in the glossary.

We publish what works for us—and implement the same solutions for your business. We design AI automation, Telegram bots, chats, and AI agents for real-world processes. Discuss your project →