Python API reference

Generated from the source, so it never drifts from the code.

For task-oriented examples, see the Python library guide.

---

Memory store

localmem_mcp.store.MemoryStore

MemoryStore(db_path=None, embedder=None, model_name=DEFAULT_MODEL)

Local memory store backed by SQLite and on-device embeddings.

Source code in src/localmem_mcp/core/store.py

def __init__(
    self,
    db_path: str | Path | None = None,
    embedder: Embedder | None = None,
    model_name: str = DEFAULT_MODEL,
):
    self.db_path = Path(db_path).expanduser() if db_path else default_db_path()
    if str(self.db_path) != ":memory:":
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
    self.embedder: Embedder = embedder or FastEmbedEmbedder(model_name)
    self._conn = sqlite3.connect(str(self.db_path), check_same_thread=False)
    self._conn.row_factory = sqlite3.Row
    self._conn.execute("PRAGMA journal_mode=WAL")
    self._conn.execute("PRAGMA synchronous=NORMAL")
    self._lock = threading.Lock()
    with self._conn:
        self._conn.executescript(_SCHEMA)

add

add(content, tags=None, source=None, metadata=None, created_at=None)

Embed and persist a memory. Returns the stored record.

created_at overrides the timestamp used for both created_at and updated_at, which is what an import needs: a memory restored from a JSONL file should keep the day it was first recorded, not the day it was restored. Leave it unset for normal writes.

Source code in src/localmem_mcp/core/store.py

def add(
    self,
    content: str,
    tags: Iterable[str] | str | None = None,
    source: str | None = None,
    metadata: dict[str, Any] | None = None,
    created_at: str | None = None,
) -> Memory:
    """Embed and persist a memory. Returns the stored record.

    ``created_at`` overrides the timestamp used for both ``created_at`` and
    ``updated_at``, which is what an import needs: a memory restored from a
    JSONL file should keep the day it was first recorded, not the day it was
    restored. Leave it unset for normal writes.
    """
    content = (content or "").strip()
    if not content:
        raise ValueError("content must not be empty")

    tag_list = _normalize_tags(tags)
    vector = self.embedder.embed([content])[0]
    timestamp = created_at or _now()

    with self._lock, self._conn:
        cursor = self._conn.execute(
            """
            INSERT INTO memories
                (content, tags, source, metadata, created_at, updated_at,
                 embedding, embedding_model, dim)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                content,
                ",".join(tag_list),
                source,
                json.dumps(metadata or {}),
                timestamp,
                timestamp,
                _pack(vector),
                self.embedder.name,
                len(vector),
            ),
        )
    assert cursor.lastrowid is not None
    return Memory(
        id=cursor.lastrowid,
        content=content,
        tags=tag_list,
        source=source,
        metadata=metadata or {},
        created_at=timestamp,
        updated_at=timestamp,
    )

update

update(memory_id, content=None, tags=None, source=None)

Correct an existing memory in place. Returns None if there's no such memory.

Only the fields you pass are changed. Changing content re-embeds the memory so search finds the correction; a tag- or source-only update leaves the vector alone. created_at is preserved and updated_at refreshed. The FTS5 index stays in sync via the AFTER UPDATE trigger.

Source code in src/localmem_mcp/core/store.py

def update(
    self,
    memory_id: int,
    content: str | None = None,
    tags: Iterable[str] | str | None = None,
    source: str | None = None,
) -> Memory | None:
    """Correct an existing memory in place. Returns None if there's no such memory.

    Only the fields you pass are changed. Changing ``content`` re-embeds the
    memory so search finds the correction; a tag- or source-only update
    leaves the vector alone. ``created_at`` is preserved and ``updated_at``
    refreshed. The FTS5 index stays in sync via the AFTER UPDATE trigger.
    """
    sets: _list[str] = []
    params: _list[Any] = []

    if content is not None:
        content = content.strip()
        if not content:
            raise ValueError("content must not be empty")
        vector = self.embedder.embed([content])[0]
        sets += ["content = ?", "embedding = ?", "embedding_model = ?", "dim = ?"]
        params += [content, _pack(vector), self.embedder.name, len(vector)]

    if tags is not None:
        tag_list = _normalize_tags(tags)
        sets.append("tags = ?")
        params.append(",".join(tag_list))

    if source is not None:
        sets.append("source = ?")
        params.append(source)

    if not sets:
        raise ValueError("nothing to update: pass content, tags, or source")

    sets.append("updated_at = ?")
    params.append(_now())
    params.append(memory_id)

    with self._lock, self._conn:
        cursor = self._conn.execute(
            f"UPDATE memories SET {', '.join(sets)} WHERE id = ?", params
        )
        if cursor.rowcount == 0:
            return None
        row = self._conn.execute("SELECT * FROM memories WHERE id = ?", (memory_id,)).fetchone()
    return _row_to_memory(row)

delete

delete(memory_id)

Delete a memory. Returns True if it existed, False if it didn't.

Source code in src/localmem_mcp/core/store.py

def delete(self, memory_id: int) -> bool:
    """Delete a memory. Returns True if it existed, False if it didn't."""
    with self._lock, self._conn:
        cursor = self._conn.execute("DELETE FROM memories WHERE id = ?", (memory_id,))
    return cursor.rowcount > 0

matching

matching(tags=None, older_than_days=None)

Memories a bulk delete would remove, newest first.

Same filters as :meth:delete_many, without deleting — the CLI uses this to show what forget --tag stale would remove before asking.

Source code in src/localmem_mcp/core/store.py

def matching(
    self,
    tags: Iterable[str] | str | None = None,
    older_than_days: int | None = None,
) -> _list[Memory]:
    """Memories a bulk delete would remove, newest first.

    Same filters as :meth:`delete_many`, without deleting — the CLI uses
    this to show what ``forget --tag stale`` would remove before asking.
    """
    where, params = _bulk_filters(tags, older_than_days)
    with self._lock:
        rows = self._conn.execute(
            f"SELECT * FROM memories WHERE {where} ORDER BY id DESC", params
        ).fetchall()
    return [_row_to_memory(row) for row in rows]

delete_many

delete_many(tags=None, older_than_days=None)

Bulk-delete memories matching all given tags and/or age.

Requires at least one of tags or older_than_days; an unfiltered call raises :class:ValueError so the store can't be wiped by accident. Deletion is hard — rows are gone, not hidden — and the FTS5 index stays in sync via the AFTER DELETE trigger. Returns the number removed.

Source code in src/localmem_mcp/core/store.py

def delete_many(
    self,
    tags: Iterable[str] | str | None = None,
    older_than_days: int | None = None,
) -> int:
    """Bulk-delete memories matching all given tags and/or age.

    Requires at least one of ``tags`` or ``older_than_days``; an unfiltered
    call raises :class:`ValueError` so the store can't be wiped by accident.
    Deletion is hard — rows are gone, not hidden — and the FTS5 index stays
    in sync via the AFTER DELETE trigger. Returns the number removed.
    """
    where, params = _bulk_filters(tags, older_than_days)
    with self._lock, self._conn:
        cursor = self._conn.execute(f"DELETE FROM memories WHERE {where}", params)
    return cursor.rowcount

get

get(memory_id)

Fetch one memory by id, or None if there's no such memory.

Source code in src/localmem_mcp/core/store.py

def get(self, memory_id: int) -> Memory | None:
    """Fetch one memory by id, or None if there's no such memory."""
    with self._lock:
        row = self._conn.execute(
            "SELECT * FROM memories WHERE id = ?", (memory_id,)
        ).fetchone()
    return _row_to_memory(row) if row else None

list

list(tags=None, limit=20, offset=0, order='newest')

List stored memories with tag filtering, ordering, and pagination.

Returns (memories, total_count) where total_count is the total number of stored memories matching all specified tags, regardless of limit and offset.

Source code in src/localmem_mcp/core/store.py

def list(
    self,
    tags: Iterable[str] | str | None = None,
    limit: int = 20,
    offset: int = 0,
    order: str = "newest",
) -> tuple[_list[Memory], int]:
    """List stored memories with tag filtering, ordering, and pagination.

    Returns ``(memories, total_count)`` where ``total_count`` is the total
    number of stored memories matching all specified tags, regardless of
    ``limit`` and ``offset``.
    """
    tag_list = _normalize_tags(tags)
    order_clean = (order or "").strip().lower()
    if order_clean not in ("newest", "oldest", "id_desc"):
        raise ValueError(f"invalid order {order!r}: must be 'newest' or 'oldest'")
    if limit <= 0:
        raise ValueError("limit must be greater than 0")
    if offset < 0:
        raise ValueError("offset must be non-negative")

    clauses: _list[str] = []
    params: _list[Any] = []
    for tag in tag_list:
        clauses.append("(',' || tags || ',') LIKE ? ESCAPE '\\'")
        params.append(f"%,{_escape_like(tag)},%")

    where_sql = f" WHERE {' AND '.join(clauses)}" if clauses else ""

    if order_clean == "newest":
        order_sql = "ORDER BY created_at DESC, id DESC"
    elif order_clean == "oldest":
        order_sql = "ORDER BY created_at ASC, id ASC"
    else:
        order_sql = "ORDER BY id DESC"

    with self._lock:
        count_row = self._conn.execute(
            f"SELECT COUNT(*) FROM memories{where_sql}", params
        ).fetchone()
        total = int(count_row[0]) if count_row else 0

        query = f"SELECT * FROM memories{where_sql} {order_sql} LIMIT ? OFFSET ?"
        rows = self._conn.execute(query, params + [limit, offset]).fetchall()

    memories = [_row_to_memory(row) for row in rows]
    return memories, total

recent

recent(limit=10, tags=None)

Most recently stored memories, newest first.

Source code in src/localmem_mcp/core/store.py

def recent(self, limit: int = 10, tags: Iterable[str] | str | None = None) -> _list[Memory]:
    """Most recently stored memories, newest first."""
    memories, _ = self.list(tags=tags, limit=max(1, limit), offset=0, order="id_desc")
    return memories

search

search(query, limit=5, tags=None, min_score=0.0, offset=0)

Semantic search, nudged by exact keyword matches.

Every stored memory is scored by cosine similarity against the query embedding; memories that also match the FTS5 index get a bounded keyword bonus so literal terms are not lost to paraphrase.

offset skips that many ranked results before limit is applied, so limit=5, offset=5 is the second page. Ranking is deterministic (score descending, ties broken by id descending), so pages neither overlap nor skip. Negative offsets are clamped to 0, and an offset past the end returns an empty list rather than raising.

Source code in src/localmem_mcp/core/store.py

def search(
    self,
    query: str,
    limit: int = 5,
    tags: Iterable[str] | str | None = None,
    min_score: float = 0.0,
    offset: int = 0,
) -> _list[SearchResult]:
    """Semantic search, nudged by exact keyword matches.

    Every stored memory is scored by cosine similarity against the query
    embedding; memories that also match the FTS5 index get a bounded
    keyword bonus so literal terms are not lost to paraphrase.

    ``offset`` skips that many ranked results before ``limit`` is applied,
    so ``limit=5, offset=5`` is the second page. Ranking is deterministic
    (score descending, ties broken by id descending), so pages neither
    overlap nor skip. Negative offsets are clamped to 0, and an offset past
    the end returns an empty list rather than raising.
    """
    query = (query or "").strip()
    if not query:
        return []

    tag_list = _normalize_tags(tags)
    # Computed before the lock, like add()/update() do — embedding is the
    # slow part, and it doesn't touch the connection.
    query_vector = self.embedder.embed([query])[0]

    with self._lock:
        rows = self._conn.execute("SELECT * FROM memories").fetchall()
        if not rows:
            return []
        keyword_hits = self._keyword_hits(query)

    results: _list[SearchResult] = []
    for row in rows:
        memory = _row_to_memory(row)
        if tag_list and not _has_tags(memory, tag_list):
            continue
        similarity = (
            _cosine(query_vector, _unpack(row["embedding"])) if row["embedding"] else 0.0
        )
        # Keyword matching is additive on top of cosine so scores stay on
        # the familiar 0-1 similarity scale that `min_score` filters on.
        score = min(1.0, similarity + KEYWORD_WEIGHT * keyword_hits.get(memory.id, 0.0))
        if score >= min_score:
            results.append(SearchResult(memory=memory, score=score))

    results.sort(key=lambda r: (-r.score, -r.memory.id))
    start = max(0, offset)
    return results[start : start + max(1, limit)]

contains

contains(content)

Whether a memory with exactly this content is already stored.

Content equality is how the import path recognises a re-import, which is deliberately cruder than semantic similarity: a duplicate here means the same text, not merely a memory that means the same thing.

Source code in src/localmem_mcp/core/store.py

def contains(self, content: str) -> bool:
    """Whether a memory with exactly this content is already stored.

    Content equality is how the import path recognises a re-import, which
    is deliberately cruder than semantic similarity: a duplicate here means
    the same text, not merely a memory that means the same thing.
    """
    content = (content or "").strip()
    if not content:
        return False
    with self._lock:
        row = self._conn.execute(
            "SELECT 1 FROM memories WHERE content = ? LIMIT 1", (content,)
        ).fetchone()
    return row is not None

export_records

export_records(tags=None, with_embeddings=False)

Yield every memory as a plain dict, oldest first.

The shape matches :meth:Memory.to_dict, so an exported record is the same structure --json already prints. tags filters conjunctively, like :meth:search.

Embeddings are left out unless with_embeddings is set: a vector is only meaningful to a machine running the same model, and the point of an export is to be portable. When included, each record also carries embedding_model and dim so a reader can tell what produced it.

Rows are fetched under the store's lock so a concurrent write can't produce a torn read, then yielded one at a time.

Source code in src/localmem_mcp/core/store.py

def export_records(
    self,
    tags: Iterable[str] | str | None = None,
    with_embeddings: bool = False,
) -> Iterator[dict[str, Any]]:
    """Yield every memory as a plain dict, oldest first.

    The shape matches :meth:`Memory.to_dict`, so an exported record is the
    same structure ``--json`` already prints. ``tags`` filters
    conjunctively, like :meth:`search`.

    Embeddings are left out unless ``with_embeddings`` is set: a vector is
    only meaningful to a machine running the same model, and the point of
    an export is to be portable. When included, each record also carries
    ``embedding_model`` and ``dim`` so a reader can tell what produced it.

    Rows are fetched under the store's lock so a concurrent write can't
    produce a torn read, then yielded one at a time.
    """
    tag_list = _normalize_tags(tags)
    with self._lock:
        rows = self._conn.execute("SELECT * FROM memories ORDER BY id").fetchall()
    for row in rows:
        memory = _row_to_memory(row)
        if tag_list and not _has_tags(memory, tag_list):
            continue
        record = memory.to_dict()
        if with_embeddings:
            record["embedding"] = _unpack(row["embedding"]) if row["embedding"] else None
            record["embedding_model"] = row["embedding_model"]
            record["dim"] = row["dim"]
        yield record

count

count()

Total number of stored memories.

Source code in src/localmem_mcp/core/store.py

def count(self) -> int:
    """Total number of stored memories."""
    with self._lock:
        return int(self._conn.execute("SELECT COUNT(*) FROM memories").fetchone()[0])

stats

stats()

Database location, memory count, and the embedding model in use.

Source code in src/localmem_mcp/core/store.py

def stats(self) -> dict[str, Any]:
    """Database location, memory count, and the embedding model in use."""
    return {
        "db_path": str(self.db_path),
        "memories": self.count(),
        "embedding_model": self.embedder.name,
    }

close

close()

Close the underlying SQLite connection.

Source code in src/localmem_mcp/core/store.py

def close(self) -> None:
    """Close the underlying SQLite connection."""
    self._conn.close()

---

Data types

localmem_mcp.store.Memory dataclass

Memory(id, content, tags=list(), source=None, metadata=dict(), created_at='', updated_at='')

localmem_mcp.store.SearchResult dataclass

SearchResult(memory, score)

localmem_mcp.store.ImportReport dataclass

ImportReport(imported=0, skipped_duplicates=0, dry_run=False, errors=list())

What an import did, or — with dry_run — what it would have done.

failed property

failed

Number of lines that could not be read as a memory.

---

Import

localmem_mcp.store.import_records

import_records(store, lines, dry_run=False, allow_duplicates=False)

Store one memory per JSONL line, skipping (and reporting) bad ones.

lines is any iterable of strings — an open file, so a large import streams rather than loading the whole thing. Blank lines are ignored.

Each memory is re-embedded by store; embeddings present in the file are ignored. created_at is preserved when the record has one. By default a record whose content already exists in the store is skipped, so importing the same file twice doesn't double the store — pass allow_duplicates to store it anyway. With dry_run nothing is written and nothing is embedded; the report says what would have happened.

Source code in src/localmem_mcp/core/portability.py

def import_records(
    store: MemoryStore,
    lines: Iterable[str],
    dry_run: bool = False,
    allow_duplicates: bool = False,
) -> ImportReport:
    """Store one memory per JSONL line, skipping (and reporting) bad ones.

    ``lines`` is any iterable of strings — an open file, so a large import
    streams rather than loading the whole thing. Blank lines are ignored.

    Each memory is re-embedded by ``store``; embeddings present in the file are
    ignored. ``created_at`` is preserved when the record has one. By default a
    record whose content already exists in the store is skipped, so importing
    the same file twice doesn't double the store — pass ``allow_duplicates`` to
    store it anyway. With ``dry_run`` nothing is written and nothing is
    embedded; the report says what would have happened.
    """
    report = ImportReport(dry_run=dry_run)
    # Dry runs write nothing, so duplicates within the file have to be tracked
    # here — the store can't tell us about a memory it never saw.
    seen: set[str] = set()

    for number, raw in enumerate(lines, start=1):
        if not raw.strip():
            continue
        try:
            record = _parse_record(raw)
        except ValueError as exc:
            report.errors.append(f"line {number}: {exc}")
            continue

        if not allow_duplicates:
            content = record["content"].strip()
            if content in seen or store.contains(content):
                report.skipped_duplicates += 1
                continue
            seen.add(content)

        if not dry_run:
            store.add(**record)
        report.imported += 1

    return report

---

Embedders

localmem_mcp.store.Embedder

Bases: Protocol

Anything that can turn text into a fixed-length vector.

embed

embed(texts)

Return one vector per input text, all of the same length.

Source code in src/localmem_mcp/core/embedders.py

def embed(self, texts: Sequence[str]) -> list[list[float]]:
    """Return one vector per input text, all of the same length."""
    ...

localmem_mcp.store.FastEmbedEmbedder

FastEmbedEmbedder(model_name=DEFAULT_MODEL, cache_dir=None)

Local ONNX embeddings via fastembed.

The model is loaded lazily so importing this module (and starting the MCP server) stays fast — the first store_memory/search_memory call pays the load cost, not process startup.

Source code in src/localmem_mcp/core/embedders.py

def __init__(self, model_name: str = DEFAULT_MODEL, cache_dir: str | None = None):
    self.name = model_name
    self._cache_dir = cache_dir
    self._model: Any = None
    self._lock = threading.Lock()

embed

embed(texts)

Embed texts locally, loading the model on the first call.

Source code in src/localmem_mcp/core/embedders.py

def embed(self, texts: Sequence[str]) -> list[list[float]]:
    """Embed texts locally, loading the model on the first call."""
    model = self._ensure_model()
    return [list(map(float, vec)) for vec in model.embed(list(texts))]

---

Helpers

localmem_mcp.store.default_db_path

default_db_path()

Where memories live unless told otherwise.

Honours LOCALMEM_DB_PATH, then LOCALMEM_HOME, then ~/.localmem.

Source code in src/localmem_mcp/core/utils.py

def default_db_path() -> Path:
    """Where memories live unless told otherwise.

    Honours ``LOCALMEM_DB_PATH``, then ``LOCALMEM_HOME``, then ``~/.localmem``.
    """
    env_path = os.environ.get("LOCALMEM_DB_PATH")
    if env_path:
        return Path(env_path).expanduser()
    home = Path(os.environ.get("LOCALMEM_HOME", "~/.localmem")).expanduser()
    return home / "memories.db"

---

Server

The MCP tool functions are documented in the MCP tools guide. These are the module-level helpers for embedding the server in your own process.

localmem_mcp.server.get_store

get_store()

Return the process-wide store, opening it on first use.

Source code in src/localmem_mcp/mcp/app.py

def get_store() -> MemoryStore:
    """Return the process-wide store, opening it on first use."""
    global _store
    if _store is None:
        _store = MemoryStore(
            db_path=os.environ.get("LOCALMEM_DB_PATH"),
            model_name=os.environ.get("LOCALMEM_MODEL", DEFAULT_MODEL),
        )
    return _store

localmem_mcp.server.configure

configure(db_path=None, model_name=None)

Point the server at a specific database/model before serving.

Source code in src/localmem_mcp/mcp/app.py

def configure(db_path: str | Path | None = None, model_name: str | None = None) -> MemoryStore:
    """Point the server at a specific database/model before serving."""
    global _store
    if _store is not None:
        _store.close()
    _store = MemoryStore(db_path=db_path, model_name=model_name or DEFAULT_MODEL)
    return _store