Memory Adapters - Memproof

Memproof uses an adapter pattern to decouple its policy and audit layer from the underlying memory storage. Every memory operation flows through the control path (risk, policy, events) before the adapter persists it.

Built-in Adapters

in_memory
langgraph
openai_sessions
mcp

A simple dictionary-backed store for local development and testing. No external dependencies. Data is lost when the process exits.

from memproof import Memproof

mp = Memproof(policy="memproof.yaml", adapter="in_memory")

Connects to LangGraph’s checkpointer and memory store APIs. Pass backend connection details to enable real provider communication.

mp = Memproof(
    policy="memproof.yaml",
    adapter="langgraph",
    langgraph_url="http://localhost:8123",
    langgraph_api_key="lg-xxx",
)

Without langgraph_url, the adapter falls back to an in-memory store.

Stores memories as session metadata via the OpenAI Responses API.

mp = Memproof(
    policy="memproof.yaml",
    adapter="openai_sessions",
    openai_api_key="sk-xxx",
    openai_organization="org-xxx",
)

Without openai_api_key, the adapter falls back to an in-memory store.

Communicates with an MCP-compliant memory server via JSON-RPC over HTTP.

mp = Memproof(
    policy="memproof.yaml",
    adapter="mcp",
    mcp_server_url="http://localhost:3333",
)

Without mcp_server_url, the adapter falls back to an in-memory store.

How Adapters Work

Each adapter translates five canonical operations to the backend:

Canonical Method	Description
`create_memory()`	Persist a new `MemoryRecord`
`update_memory()`	Apply a partial patch to an existing record
`delete_memory()`	Remove a record by ID
`get_memory()`	Retrieve a single record by ID
`search_memories()`	Run a scoped query, return ranked hits

Adapters also normalize backend errors into canonical error codes: NOT_FOUND, CONFLICT, VALIDATION_ERROR, PERMISSION_DENIED, and PROVIDER_UNAVAILABLE.

Writing a Custom Adapter

Subclass MemoryAdapter and implement all five abstract methods plus the provider_name property.

The Base Interface

from memproof.adapters.base import MemoryAdapter, AdapterError
from memproof.models.core import MemoryRecord, MemorySearchHit

class MemoryAdapter(ABC):
    @property
    @abstractmethod
    def provider_name(self) -> str: ...

    @abstractmethod
    async def create_memory(self, memory: MemoryRecord) -> MemoryRecord: ...

    @abstractmethod
    async def update_memory(self, memory_id: str, patch: dict) -> MemoryRecord: ...

    @abstractmethod
    async def delete_memory(self, memory_id: str) -> bool: ...

    @abstractmethod
    async def get_memory(self, memory_id: str) -> MemoryRecord | None: ...

    @abstractmethod
    async def search_memories(
        self, query: str, scope: dict, limit: int = 20,
        filters: dict | None = None,
    ) -> list[MemorySearchHit]: ...

Example: Redis Adapter

from memproof.adapters.base import MemoryAdapter, AdapterError
from memproof.models.core import MemoryRecord, MemorySearchHit

class RedisAdapter(MemoryAdapter):
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        import redis.asyncio as redis
        self._client = redis.from_url(redis_url)

    @property
    def provider_name(self) -> str:
        return "redis"

    async def create_memory(self, memory: MemoryRecord) -> MemoryRecord:
        key = f"mem:{memory.memory_id}"
        if await self._client.exists(key):
            raise AdapterError("CONFLICT", f"{memory.memory_id} exists", self.provider_name)
        await self._client.set(key, memory.model_dump_json())
        return memory

    async def update_memory(self, memory_id: str, patch: dict) -> MemoryRecord:
        raw = await self._client.get(f"mem:{memory_id}")
        if not raw:
            raise AdapterError("NOT_FOUND", f"{memory_id} not found", self.provider_name)
        data = MemoryRecord.model_validate_json(raw).model_dump()
        for k, v in patch.items():
            if v is not None and k in data:
                data[k] = v
        updated = MemoryRecord(**data)
        await self._client.set(f"mem:{memory_id}", updated.model_dump_json())
        return updated

    async def delete_memory(self, memory_id: str) -> bool:
        result = await self._client.delete(f"mem:{memory_id}")
        if not result:
            raise AdapterError("NOT_FOUND", f"{memory_id} not found", self.provider_name)
        return True

    async def get_memory(self, memory_id: str) -> MemoryRecord | None:
        raw = await self._client.get(f"mem:{memory_id}")
        return MemoryRecord.model_validate_json(raw) if raw else None

    async def search_memories(
        self, query: str, scope: dict, limit: int = 20,
        filters: dict | None = None,
    ) -> list[MemorySearchHit]:
        # Implement using RediSearch or scan-based matching
        return []

Raise AdapterError with one of the canonical error codes. The orchestrator relies on these codes to decide whether to quarantine, retry, or surface the error.

Using a Custom Adapter with Memproof

Pass a pre-built MemproofConfig and override the adapter after construction, or contribute it upstream so it can be selected by name.

from memproof import Memproof
from memproof.config import MemproofConfig

config = MemproofConfig(policy_path="memproof.yaml", adapter="in_memory")
mp = Memproof(config=config)
mp._adapter = RedisAdapter(redis_url="redis://localhost:6379")

For production backends that require HTTP calls, follow the backend pattern used by LangGraphBackend, OpenAISessionsBackend, and MCPMemoryBackend — create a separate backend class and inject it into the adapter constructor.

​Built-in Adapters

​How Adapters Work

​Writing a Custom Adapter

​The Base Interface

​Example: Redis Adapter

​Using a Custom Adapter with Memproof

Built-in Adapters

How Adapters Work

Writing a Custom Adapter

The Base Interface

Example: Redis Adapter

Using a Custom Adapter with Memproof