Installation
- Python
- TypeScript
pip install memproof
npm install @kyberon/memproof
1. Create a Policy File
Creatememproof.yaml in your project root:
version: 0.1.0
mode: enforce
defaults:
on_policy_miss: deny
on_adapter_error: quarantine
require_idempotency: true
risk_thresholds:
low_max: 0.30
medium_max: 0.60
high_max: 0.80
critical_max: 1.00
rules:
- id: allow_trusted_reads
description: Allow reads from trusted sources.
enabled: true
priority: 100
action: allow
reason_codes: [SAFE_READ]
match: all
when:
- field: operation_type
operator: in
value: [search, get]
- field: context.source
operator: in
value: [langgraph, openai_sessions, mcp]
- id: allow_low_risk_writes
description: Allow low-risk writes from trusted sources.
enabled: true
priority: 150
action: allow
reason_codes: [TRUSTED_LOW_RISK]
match: all
when:
- field: operation_type
operator: in
value: [remember, update, forget]
- field: context.source
operator: in
value: [langgraph, openai_sessions, mcp]
- field: risk_level
operator: in
value: [low, medium]
- id: approve_high_risk
description: Require approval for high-risk mutations.
enabled: true
priority: 200
action: require_approval
reason_codes: [HIGH_RISK]
match: all
when:
- field: operation_type
operator: in
value: [remember, update, forget]
- field: risk_level
operator: in
value: [high, critical]
2. Initialize Memproof
- Python
- TypeScript
import asyncio
from memproof import Memproof
async def main():
mp = Memproof(policy="./memproof.yaml")
# Create a memory
result = await mp.remember(
content="user prefers dark mode",
scope={
"tenant_id": "acme-corp",
"project_id": "chatbot",
"agent_id": "assistant-1",
},
context={
"actor_type": "agent",
"actor_id": "assistant-1",
"source": "langgraph",
"timestamp": "2026-01-15T10:30:00Z",
},
tags=["preference", "ui"],
)
print(f"Status: {result.status}") # committed
print(f"Decision: {result.decision.action}") # allow
print(f"Risk: {result.risk_assessment.level}") # low
print(f"Memory ID: {result.memory.memory_id}")
# Read it back
memory = await mp.get(result.memory.memory_id)
print(f"Content: {memory.content}")
# Search
search = await mp.search(
query="dark mode",
scope={"tenant_id": "acme-corp", "project_id": "chatbot", "agent_id": "assistant-1"},
context={"actor_type": "agent", "actor_id": "assistant-1",
"source": "langgraph", "timestamp": "2026-01-15T10:31:00Z"},
)
print(f"Found {len(search.hits)} results")
asyncio.run(main())
import { Memproof } from "@kyberon/memproof";
const mp = new Memproof({ policy: "./memproof.yaml" });
// Create a memory
const result = await mp.remember({
content: "user prefers dark mode",
scope: {
tenantId: "acme-corp",
projectId: "chatbot",
agentId: "assistant-1",
},
context: {
actorType: "agent",
actorId: "assistant-1",
source: "langgraph",
timestamp: "2026-01-15T10:30:00Z",
},
tags: ["preference", "ui"],
});
console.log(`Status: ${result.status}`); // committed
console.log(`Decision: ${result.decision.action}`); // allow
console.log(`Risk: ${result.riskAssessment.level}`); // low
console.log(`Memory ID: ${result.memory.memoryId}`);
// Read it back
const memory = await mp.get(result.memory.memoryId);
console.log(`Content: ${memory.content}`);
// Search
const search = await mp.search({
query: "dark mode",
scope: { tenantId: "acme-corp", projectId: "chatbot", agentId: "assistant-1" },
context: { actorType: "agent", actorId: "assistant-1",
source: "langgraph", timestamp: "2026-01-15T10:31:00Z" },
});
console.log(`Found ${search.hits.length} results`);
3. Handle Policy Decisions
- Python
- TypeScript
from memproof import Memproof, PolicyDeniedError, QuarantinedError
mp = Memproof(policy="./memproof.yaml")
try:
result = await mp.remember(
content="some sensitive data",
scope=scope,
context=context,
)
if result.status == "pending_approval":
print(f"Waiting for approval: {result.operation_id}")
# Later: await mp.approve(result.operation_id, actor_id="admin-1")
else:
print(f"Committed: {result.memory.memory_id}")
except PolicyDeniedError as e:
print(f"Blocked by policy: {e.code}")
except QuarantinedError as e:
print(f"Quarantined for review: {e.details}")
import { Memproof, PolicyDeniedError, QuarantinedError } from "@kyberon/memproof";
const mp = new Memproof({ policy: "./memproof.yaml" });
try {
const result = await mp.remember({ content: "some sensitive data", scope, context });
if (result.status === "pending_approval") {
console.log(`Waiting for approval: ${result.operationId}`);
// Later: await mp.approve(result.operationId, "admin-1");
} else {
console.log(`Committed: ${result.memory.memoryId}`);
}
} catch (e) {
if (e instanceof PolicyDeniedError) {
console.log(`Blocked by policy: ${e.code}`);
} else if (e instanceof QuarantinedError) {
console.log(`Quarantined for review: ${e.details}`);
}
}
4. Use a Real Adapter
- Python
- TypeScript
# LangGraph
mp = Memproof(
policy="./memproof.yaml",
adapter="langgraph",
langgraph_url="http://localhost:8123",
langgraph_api_key="your-key",
)
# OpenAI Sessions
mp = Memproof(
policy="./memproof.yaml",
adapter="openai_sessions",
openai_api_key="sk-...",
)
# MCP Memory Server
mp = Memproof(
policy="./memproof.yaml",
adapter="mcp",
mcp_server_url="http://localhost:3333",
)
// LangGraph
const mp = new Memproof({
policy: "./memproof.yaml",
adapter: "langgraph",
langgraphUrl: "http://localhost:8123",
langgraphApiKey: "your-key",
});
// OpenAI Sessions
const mp = new Memproof({
policy: "./memproof.yaml",
adapter: "openai_sessions",
openaiApiKey: "sk-...",
});
// MCP Memory Server
const mp = new Memproof({
policy: "./memproof.yaml",
adapter: "mcp",
mcpServerUrl: "http://localhost:3333",
});
Next Steps
Control Path
Understand the 6-stage pipeline every operation passes through.
Policy Authoring
Write custom policy rules with conditions and operators.