mem0 Integration - Memproof

mem0 is a memory storage and retrieval system for AI applications. Memproof can sit in front of mem0, adding policy evaluation, risk assessment, content redaction, and audit logging to every memory operation. Your agents continue to use mem0 for storage, but Memproof enforces governance before anything is persisted or retrieved.

Configuration

Set the adapter to mem0 and provide your mem0 credentials. Memproof will route all memory operations through the mem0 API while applying the full control path.

Python
TypeScript

from memproof import Memproof

mp = Memproof(
    policy="./memproof.yaml",
    adapter="mem0",
    mem0_api_key="your-api-key",
    mem0_org_id="your-org-id",
    mem0_project_id="your-project-id",
)

import { Memproof } from "@kyberon/memproof";

const mp = new Memproof({
    policy: "./memproof.yaml",
    config: {
        adapter: "mem0",
        mem0ApiKey: "your-api-key",
        mem0OrgId: "your-org-id",
        mem0ProjectId: "your-project-id",
    },
});

How It Works

When you call mp.remember(), mp.search(), or any other memory operation, Memproof executes the following pipeline before reaching mem0:

Auth check

If an auth hook is configured, it validates the caller before anything else runs.

Content redaction

PII and secrets are stripped from the content using the ContentRedactor.

Risk assessment

The risk engine scores the operation based on content signals, scope, and operation type.

Policy evaluation

The policy engine evaluates rules against the operation context. The result is allow, deny, require_approval, or quarantine.

mem0 execution

If the policy allows the operation, Memproof calls the mem0 API to persist or retrieve the memory.

Audit logging

An immutable event is recorded in the Trailproof audit trail with the full decision trail.

Usage Example

Once configured, use Memproof exactly as you would with any other adapter. The mem0 backend is transparent to the caller.

Python
TypeScript

import asyncio
from memproof import Memproof

async def main():
    mp = Memproof(
        policy="./memproof.yaml",
        adapter="mem0",
        mem0_api_key="your-api-key",
        mem0_org_id="your-org-id",
        mem0_project_id="your-project-id",
    )

    # Store a memory -- Memproof evaluates policy, then persists via mem0
    result = await mp.remember(
        content="User prefers dark mode and metric units",
        scope={"tenant_id": "acme", "project_id": "assistant", "agent_id": "pref-bot"},
        context={"actor_type": "agent", "actor_id": "pref-bot",
                 "source": "api", "timestamp": "2026-01-15T10:00:00Z"},
    )
    print(result.status)  # "committed"

    # Search memories -- policy is evaluated on reads too
    hits = await mp.search(
        query="user display preferences",
        scope={"tenant_id": "acme", "project_id": "assistant", "agent_id": "pref-bot"},
        context={"actor_type": "agent", "actor_id": "pref-bot",
                 "source": "api", "timestamp": "2026-01-15T10:01:00Z"},
    )
    for hit in hits:
        print(hit.content, hit.score)

asyncio.run(main())

import { Memproof } from "@kyberon/memproof";

const mp = new Memproof({
    policy: "./memproof.yaml",
    config: {
        adapter: "mem0",
        mem0ApiKey: "your-api-key",
        mem0OrgId: "your-org-id",
        mem0ProjectId: "your-project-id",
    },
});

const result = await mp.remember({
    content: "User prefers dark mode and metric units",
    scope: { tenantId: "acme", projectId: "assistant", agentId: "pref-bot" },
    context: { actorType: "agent", actorId: "pref-bot",
               source: "api", timestamp: "2026-01-15T10:00:00Z" },
});
console.log(result.status); // "committed"

Custom Mem0 Client

If you need to customize the HTTP client, connection pooling, or retry behavior, you can provide your own Mem0Client implementation and inject it into the adapter.

from memproof import Memproof
from memproof.config import MemproofConfig
from memproof.adapters.mem0 import Mem0Adapter, Mem0Client

class CustomMem0Client(Mem0Client):
    """Custom client with retry logic and connection pooling."""

    async def add(self, content: str, metadata: dict) -> dict:
        # Your custom implementation
        ...

config = MemproofConfig(policy_path="./memproof.yaml", adapter="mem0")
mp = Memproof(config=config)
mp._adapter = Mem0Adapter(client=CustomMem0Client(
    api_key="your-api-key",
    org_id="your-org-id",
    project_id="your-project-id",
))

When providing a custom Mem0Client, ensure your implementation raises AdapterError with canonical error codes. The orchestrator depends on these codes to decide whether to quarantine, retry, or surface the error.

The mem0 adapter maps Memproof scopes to mem0 metadata fields. tenant_id, project_id, and agent_id are stored as mem0 metadata so that scoped searches work correctly across both systems.

​Configuration

​How It Works

​Usage Example

​Custom Mem0 Client

Configuration

How It Works

Usage Example

Custom Mem0 Client