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)

Embed and persist a memory. Returns the stored record.

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,
) -> Memory:
    """Embed and persist a memory. Returns the stored record."""
    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 = _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),
            ),
        )
    return Memory(
        id=int(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)
    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."""
    row = self._conn.execute(
        "SELECT * FROM memories WHERE id = ?", (memory_id,)
    ).fetchone()
    return _row_to_memory(row) if row else None

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."""
    tag_list = _normalize_tags(tags)
    rows = self._conn.execute(
        "SELECT * FROM memories ORDER BY id DESC LIMIT ?",
        (max(1, limit) * (10 if tag_list else 1),),
    ).fetchall()
    memories = [_row_to_memory(row) for row in rows]
    if tag_list:
        memories = [m for m in memories if _has_tags(m, tag_list)]
    return memories[:limit]

search

search(query, limit=5, tags=None, min_score=0.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.

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,
) -> 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.
    """
    query = (query or "").strip()
    if not query:
        return []

    tag_list = _normalize_tags(tags)
    rows = self._conn.execute("SELECT * FROM memories").fetchall()
    if not rows:
        return []

    query_vector = self.embedder.embed([query])[0]
    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))
    return results[: max(1, limit)]

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."""
    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)

---

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