> ## Documentation Index
> Fetch the complete documentation index at: https://memproof.kyberon.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# Memory Adapters

> Configure built-in adapters or write your own to connect Memproof to any memory backend.

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

<Tabs>
  <Tab title="in_memory">
    A simple dictionary-backed store for **local development and testing**. No external dependencies. Data is lost when the process exits.

    ```python theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
    from memproof import Memproof

    mp = Memproof(policy="memproof.yaml", adapter="in_memory")
    ```
  </Tab>

  <Tab title="langgraph">
    Connects to LangGraph's checkpointer and memory store APIs. Pass backend connection details to enable real provider communication.

    ```python theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
    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.
  </Tab>

  <Tab title="openai_sessions">
    Stores memories as session metadata via the OpenAI Responses API.

    ```python theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
    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.
  </Tab>

  <Tab title="mcp">
    Communicates with an MCP-compliant memory server via JSON-RPC over HTTP.

    ```python theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
    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.
  </Tab>
</Tabs>

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

```python theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
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

```python theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
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 []
```

<Warning>
  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.
</Warning>

### 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.

```python theme={"theme":{"light":"github-light","dark":"github-dark-dimmed"}}
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")
```

<Tip>
  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.
</Tip>
