How to Implement a Custom Storage Backend in MemPalace: A Complete Guide

To implement a custom storage backend in MemPalace, subclass BaseBackend and BaseCollection from mempalace/backends/base.py, implement the required CRUD methods, expose the backend via the mempalace.backends entry-point group in pyproject.toml, and declare capabilities via the capabilities class attribute.

MemPalace is an open-source vector memory library that isolates storage behind a strict contract defined in mempalace/backends/base.py. Whether you are integrating a proprietary database, optimising for specific hardware, or bridging to a cloud service, implementing a custom storage backend lets you extend the framework while preserving its isolation guarantees. This guide walks through the concrete implementation steps using the actual source code contracts and patterns found in the repository.

Step 1: Subclass BaseBackend and Define the Contract

Every custom backend must inherit from BaseBackend and satisfy the abstract contract defined in mempalace/backends/base.py (lines 365-385). At minimum, you must:

• Define a unique name class attribute (e.g., "mybackend").
• Populate the capabilities frozenset with feature tokens such as "supports_metadata_filters", "requires_explicit_embeddings", or "supports_lexical_search".
• Implement the abstract get_collection() method, which returns a BaseCollection instance tied to a specific palace and collection name.

The get_collection() signature must accept palace: PalaceRef, collection_name: str, create: bool, and options: Optional[dict]. This method is the factory that instantiates your collection handler.

Step 2: Implement BaseCollection for CRUD Operations

The storage engine itself lives in a BaseCollection subclass (abstract specification in mempalace/backends/base.py, lines 333-368). You must implement all required read/write methods:

• add() – Insert new documents with optional embeddings and metadata.
• upsert() – Update existing documents or insert new ones.
• query() – Perform similarity search using query_embeddings or query_texts, respecting n_results, where filters, and include parameters.
• get() – Fetch documents by ID or metadata filters with pagination (limit/offset).
• delete() – Remove documents by ID or metadata predicates.
• count() – Return the total number of documents in the collection.

Optional methods like lexical_search() can be added if your backend supports text-based retrieval. Follow the reference implementations in mempalace/backends/sqlite_exact.py (lines 14-34) and mempalace/backends/chroma.py for patterns on connection pooling and transaction handling.

Step 3: Expose Your Backend via Entry Points

MemPalace discovers backends at import time using Python entry points. Add the following to your pyproject.toml:

[project.entry-points."mempalace.backends"]
mybackend = "myproject.mybackend:MyBackend"

The discovery logic resides in mempalace/backends/registry.py (lines 55-73). After installing your package with pip install -e ., MemPalace will automatically register mybackend alongside built-in options like sqlite_exact and chroma.

Step 4: Declare Capabilities for Feature Routing

Declare your backend’s feature set via the capabilities class attribute. Standard tokens include:

• "requires_explicit_embeddings" – Your backend does not compute embeddings internally.
• "supports_metadata_filters" – Your storage can filter by metadata predicates.
• "supports_lexical_search" – Your backend implements full-text search.
• "local_mode" – Indicates local file-based storage.

These tokens allow the MemPalace core to route queries correctly and skip unsupported operations. The recognized capability names are documented in the base module and used by the conformance test suite.

Step 5: Implement Health Checks

Override the health() method to report backend status. The default implementation in mempalace/backends/base.py (lines 403-406) returns HealthStatus.healthy(). Your implementation should accept an optional palace: PalaceRef and return HealthStatus.healthy() or HealthStatus.unhealthy(reason):

def health(self, palace: Optional[PalaceRef] = None) -> HealthStatus:
    if self._closed:
        return HealthStatus.unhealthy("backend closed")
    if palace and not os.path.exists(self._data_path(palace.local_path)):
        return HealthStatus.unhealthy("data file missing")
    return HealthStatus.healthy()

Step 6: Add Auto-Detection Logic (Optional)

If you want MemPalace to automatically select your backend when opening existing palaces, implement the detect() class method. This method receives a path: str and should return True when your backend’s artifacts are present on disk:

@classmethod
def detect(cls, path: str) -> bool:
    return os.path.isfile(os.path.join(path, "mybackend.db"))

The registry calls this during auto-resolution (see mempalace/backends/registry.py, lines 88-96).

Step 7: Register for Testing (Optional)

In unit tests, you can bypass entry-point discovery by calling mempalace.backends.register():

from mempalace.backends import register
from myproject.mybackend import MyBackend

register("mybackend", MyBackend)

This function is defined in mempalace/backends/registry.py (lines 35-44) and allows rapid iteration without rebuilding packages.

Step 8: Validate with Conformance Tests

MemPalace ships with a backend-conformance test suite (tests/test_backend_conformance.py). Your implementation must pass these tests to guarantee isolation per PalaceRef.id and optional namespace handling. Run the suite against your backend to verify that add, query, delete, and get operations behave correctly across collection boundaries.

Complete Implementation Example

Below is a minimal yet complete skeleton showing MyBackend and MyCollection integrated with the MemPalace contracts:

# myproject/mybackend.py

from __future__ import annotations
from dataclasses import dataclass
from typing import Optional, ClassVar, List, Dict
import os

from mempalace.backends.base import (
    BaseBackend,
    BaseCollection,
    PalaceRef,
    QueryResult,
    GetResult,
    HealthStatus,
    BackendClosedError,
)

# ----------------------------------------------------------------------

# Collection implementation

# ----------------------------------------------------------------------

class MyCollection(BaseCollection):
    def __init__(self, handle: object, collection_name: str):
        self.handle = handle
        self.name = collection_name

    def add(self, *, documents: List[str], ids: List[str],
            metadatas: Optional[List[Dict]] = None,
            embeddings: Optional[List[List[float]]] = None) -> None:
        # Insert logic here

        pass

    def upsert(self, *, documents: List[str], ids: List[str],
               metadatas: Optional[List[Dict]] = None,
               embeddings: Optional[List[List[float]]] = None) -> None:
        # Upsert logic here

        pass

    def query(self, *, query_texts: Optional[List[str]] = None,
              query_embeddings: Optional[List[List[float]]] = None,
              n_results: int = 10,
              where: Optional[dict] = None,
              where_document: Optional[dict] = None,
              include: Optional[List[str]] = None) -> QueryResult:
        # Similarity search logic here

        return QueryResult(ids=[], embeddings=None, metadatas=None, documents=None, distances=None)

    def get(self, *, ids: Optional[List[str]] = None,
            where: Optional[dict] = None,
            where_document: Optional[dict] = None,
            limit: Optional[int] = None,
            offset: Optional[int] = None,
            include: Optional[List[str]] = None) -> GetResult:
        # Fetch logic here

        return GetResult(ids=[], embeddings=None, metadatas=None, documents=None)

    def delete(self, *, ids: Optional[List[str]] = None,
               where: Optional[dict] = None) -> None:
        # Deletion logic here

        pass

    def count(self) -> int:
        return 0

# ----------------------------------------------------------------------

# Backend implementation

# ----------------------------------------------------------------------

class MyBackend(BaseBackend):
    """Custom backend that stores vectors in <your-store>."""
    name: ClassVar[str] = "mybackend"
    capabilities: ClassVar[frozenset[str]] = frozenset({
        "requires_explicit_embeddings",
        "supports_metadata_filters",
        "supports_lexical_search",
        "local_mode",
    })

    def __init__(self) -> None:
        self._clients: dict[str, object] = {}
        self._closed = False

    @staticmethod
    def _data_path(palace_path: str) -> str:
        return os.path.join(palace_path, "mybackend.db")

    def get_collection(self, *, palace: PalaceRef,
                       collection_name: str,
                       create: bool = False,
                       options: Optional[dict] = None) -> MyCollection:
        if self._closed:
            raise BackendClosedError("MyBackend is closed")
        db_path = self._data_path(palace.local_path or "")
        handle = self._clients.setdefault(db_path, object())  # Replace with real handle

        return MyCollection(handle, collection_name)

    def health(self, palace: Optional[PalaceRef] = None) -> HealthStatus:
        if self._closed:
            return HealthStatus.unhealthy("backend closed")
        if palace and palace.local_path and not os.path.isfile(self._data_path(palace.local_path)):
            return HealthStatus.unhealthy("data file missing")
        return HealthStatus.healthy()

    @classmethod
    def detect(cls, path: str) -> bool:
        return os.path.isfile(cls._data_path(path))

__all__ = ["MyBackend"]

After adding the entry-point to pyproject.toml and installing the package, MemPalace will discover mybackend automatically.

Summary

• Subclass BaseBackend in mempalace/backends/base.py to define your backend’s name, capabilities, and collection factory.
• Implement BaseCollection to handle add, upsert, query, get, delete, and count operations.
• Expose via entry points in pyproject.toml under the group mempalace.backends for automatic discovery.
• Declare capabilities so MemPalace knows which features your storage supports.
• Override health() to provide runtime status checks.
• Optionally implement detect() for automatic backend selection based on filesystem artifacts.
• Pass conformance tests in tests/test_backend_conformance.py to guarantee correct isolation semantics.

Frequently Asked Questions

What is the minimum set of methods I must implement for a custom storage backend?

You must subclass BaseBackend and implement get_collection() to return a BaseCollection instance. Within your collection class, you must implement add, upsert, query, get, delete, and count. These methods are defined as abstract in mempalace/backends/base.py (lines 333-368), and skipping any will raise NotImplementedError at runtime.

How does MemPalace discover my custom backend without importing my module explicitly?

MemPalace uses Python’s entry-point system. When you declare [project.entry-points."mempalace.backends"] in your pyproject.toml, the registry logic in mempalace/backends/registry.py (lines 55-73) iterates over these entries at import time and instantiates your backend class. This allows zero-configuration discovery once your package is installed in the environment.

Can I use a custom backend without publishing a package or modifying pyproject.toml?

Yes. For development and testing, you can manually register your backend using mempalace.backends.register(name, BackendClass) as defined in mempalace/backends/registry.py (lines 35-44). This bypasses entry-point discovery and is useful for unit tests or notebook environments where you want to iterate quickly without rebuilding the package.

What happens if my backend raises an error during a collection operation?

MemPalace expects backends to raise specific exception types defined in mempalace/backends/base.py, such as BackendClosedError when operating on a closed backend. If you raise standard Python exceptions, they will propagate to the caller unhandled. For production stability, wrap low-level storage errors (e.g., connection failures, constraint violations) into the appropriate MemPalace exception types so the core library can handle retries or cleanup correctly.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}