How to Implement a Custom Storage Backend for MemPalace: A Complete Developer's Guide

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

MemPalace isolates vector storage behind a well-defined contract, enabling you to integrate any underlying storage system—from SQL databases to cloud services—while preserving the library's isolation guarantees. Implementing a custom storage backend allows you to extend MemPalace with proprietary databases or hardware-optimized engines without modifying core library code.

Understanding the Backend Architecture

MemPalace architecture separates storage concerns into two primary abstractions defined in mempalace/backends/base.py. Understanding these contracts is essential before implementing your own backend.

The BaseBackend Contract

The BaseBackend abstract class (lines 365-385 in mempalace/backends/base.py) serves as the factory and lifecycle manager for storage connections. Your implementation must define a unique name class attribute, an optional capabilities frozenset, and the abstract get_collection() method. This method returns a BaseCollection instance tied to a specific PalaceRef, handling resource initialization and caching.

The BaseCollection Interface

The BaseCollection abstract class (lines 333-368 in mempalace/backends/base.py) specifies the data operations contract. You must implement all required CRUD methods: add, upsert, query, get, delete, and count. Optional methods such as lexical_search can be added based on your declared capabilities.

Step-by-Step Implementation Guide

Follow these steps to create a production-ready backend that integrates seamlessly with MemPalace's discovery and testing infrastructure.

1. Subclass BaseBackend

Create a class inheriting from BaseBackend and define the mandatory class attributes. The name attribute identifies your backend in configuration files, while capabilities declares feature support to the core engine.

from mempalace.backends.base import BaseBackend, BaseCollection, PalaceRef

class MyBackend(BaseBackend):
    name: ClassVar[str] = "mybackend"
    capabilities: ClassVar[frozenset[str]] = frozenset({
        "requires_explicit_embeddings",
        "supports_metadata_filters",
        "local_mode",
    })

2. Implement BaseCollection Methods

Provide concrete implementations for all required methods in your collection class. Reference SQLiteExactBackend in mempalace/backends/sqlite_exact.py (lines 14-34) for patterns on connection pooling and schema initialization. Your add method must handle document insertion with optional embeddings and metadata, while query must support both text and embedding-based similarity search.

3. Register via Entry Points

Expose your backend through the mempalace.backends entry-point group so MemPalace can discover it at import time. 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), which scans entry points during backend resolution.

4. Declare Capabilities

Populate the capabilities frozenset with standard tokens to enable feature routing:

• "requires_explicit_embeddings" – Indicates your backend does not generate embeddings internally
• "supports_metadata_filters" – Enables the where parameter in queries
• "supports_lexical_search" – Enables full-text search functionality
• "local_mode" – Indicates file-based or local-only storage

These flags allow MemPalace core to correctly route operations and enable appropriate UI features.

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(), but you should customize this to verify database connectivity or file accessibility:

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()

6. Add Auto-Detection (Optional)

Implement the detect() class method to enable automatic backend selection for existing palaces. The registry calls this during auto-detection (lines 88-96 in mempalace/backends/registry.py):

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

Complete Production-Ready Example

Below is a minimal yet complete implementation template following patterns from ChromaBackend and SQLiteExactBackend:

# 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,
)

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 documents with optional embeddings and metadata."""
        pass

    def upsert(self, *, documents: List[str], ids: List[str],
               metadatas: Optional[List[Dict]] = None,
               embeddings: Optional[List[List[float]]] = None) -> None:
        """Insert or update existing documents."""
        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:
        """Perform similarity search against stored embeddings."""
        pass

    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:
        """Retrieve documents by ID or filter."""
        pass

    def delete(self, *, ids: Optional[List[str]] = None,
               where: Optional[dict] = None) -> None:
        """Remove documents from storage."""
        pass

    def count(self) -> int:
        """Return the total number of documents."""
        return 0

class MyBackend(BaseBackend):
    """Custom backend implementation for MemPalace."""
    name: ClassVar[str] = "mybackend"
    capabilities: ClassVar[frozenset[str]] = frozenset({
        "requires_explicit_embeddings",
        "supports_metadata_filters",
        "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:
        """Return a collection instance for the specified palace."""
        if self._closed:
            raise BackendClosedError("MyBackend is closed")
        
        db_path = self._data_path(palace.local_path or "")
        handle = self._clients.setdefault(db_path, object())
        return MyCollection(handle, collection_name)

    def health(self, palace: Optional[PalaceRef] = None) -> HealthStatus:
        """Report backend health status."""
        if self._closed:
            return HealthStatus.unhealthy("backend closed")
        if palace and palace.local_path:
            if 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:
        """Auto-detect if this backend created the palace artifacts."""
        return os.path.isfile(cls._data_path(path))

__all__ = ["MyBackend"]

Testing and Conformance

MemPalace ships with a comprehensive conformance test suite in tests/test_backend_conformance.py that validates isolation guarantees across PalaceRef.id boundaries and optional namespaces. Your backend must pass these tests to ensure compatibility with the MemPalace ecosystem.

For unit testing, you can bypass entry-point discovery by manually registering your backend using mempalace.backends.register(name, BackendClass), as implemented in mempalace/backends/registry.py (lines 35-44).

Summary

• Subclass BaseBackend in mempalace/backends/base.py (lines 365-385) to define your backend's factory interface and capabilities.
• Implement BaseCollection (lines 333-368) with required methods: add, upsert, query, get, delete, and count.
• Expose via entry points using the mempalace.backends group in pyproject.toml for automatic discovery by the registry (lines 55-73).
• Declare capabilities such as "supports_metadata_filters" to enable core features.
• Add health checks by overriding the health() method to monitor storage connectivity.
• Implement detect() for automatic backend selection when opening existing palaces.
• Run conformance tests from tests/test_backend_conformance.py to verify isolation guarantees.

Frequently Asked Questions

What is the minimum set of methods required for a custom storage backend?

You must implement the BaseBackend.get_collection() method and all abstract methods in BaseCollection: add, upsert, query, get, delete, and count. These methods handle the core vector storage operations. Optional methods like lexical_search only need implementation if you declare the corresponding capability in your backend's capabilities frozenset.

How does MemPalace discover my custom backend?

MemPalace uses Python's entry-point system via the mempalace.backends group. When you add your backend to [project.entry-points."mempalace.backends"] in pyproject.toml, the registry module (mempalace/backends/registry.py, lines 55-73) automatically discovers and loads your class at runtime. For testing, you can manually register backends using mempalace.backends.register().

Can I implement a backend that connects to cloud-based vector databases?

Yes. While the example above shows a local file-based implementation, you can implement BaseBackend to connect to any cloud service or remote database. Simply manage your connection pools in __init__ or get_collection(), and omit the "local_mode" capability if your backend requires network access. Ensure your health() method checks remote connectivity rather than local file existence.

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