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