Data Storage Guide

Anchor provides governed key-value data storage with policy enforcement. Every write is checked against policies and logged to the audit trail.

What Goes in Data Storage?

Agents typically store:

• Facts learned from conversations: "user speaks spanish", "prefers morning meetings"
• User context: timezone, communication preferences, past topics
• Task state: current step, accumulated results, pending actions
• Conversation summaries: compressed history for context windows

Writing Data

from anchor import Anchor

anchor = Anchor(api_key="anc_...")
agent = anchor.agents.create(name="support-bot")

# Configure policies
anchor.config.update(agent.id, {
    "policies": {
        "block_pii": True,
        "block_secrets": True
    }
})

# Write data (policy-checked)
result = anchor.data.write(agent.id, "user:123:language", "spanish")

if result.allowed:
    print(f"Stored with audit_id: {result.audit_id}")
else:
    print(f"Blocked by {result.blocked_by}: {result.message}")

What Gets Blocked

With block_pii and block_secrets policies:

# Blocked - email is PII
result = anchor.data.write(agent.id, "user:123:contact", "john@example.com")
# result.allowed = False, result.blocked_by = "policy:block_pii"

# Blocked - API key pattern
result = anchor.data.write(agent.id, "config:api_key", "sk-1234567890abcdef")
# result.allowed = False, result.blocked_by = "policy:block_secrets"

# Allowed - just a fact
result = anchor.data.write(agent.id, "user:123:prefers_dark_mode", "true")
# result.allowed = True

Reading Data

Get by Key

value = anchor.data.read(agent.id, "user:123:language")
# Returns: "spanish"

Read Full Entry

entry = anchor.data.read_full(agent.id, "user:123:language")

if entry:
    print(f"Value: {entry.value}")
    print(f"Created: {entry.created_at}")
    print(f"Metadata: {entry.metadata}")

List Keys

# All keys for a prefix
keys = anchor.data.list(agent.id, prefix="user:123:")

# Pagination
keys = anchor.data.list(agent.id, limit=50, offset=100)

Search

# Semantic search across stored data
results = anchor.data.search(agent.id, "user preferences", limit=5)

for r in results:
    print(f"{r.key}: {r.value} (similarity: {r.similarity})")

Key Naming Conventions

Use a consistent naming scheme:

# Pattern: {entity_type}:{entity_id}:{attribute}

# User facts
"user:123:language"
"user:123:timezone"
"user:123:meeting_preference"

# Session/conversation context
"session:abc:summary"
"session:abc:last_topic"

# Task state
"task:456:status"
"task:456:current_step"

Batch Operations

# Write multiple items at once
results = anchor.data.write_batch(agent.id, [
    {"key": "user:123:name", "value": "John"},
    {"key": "user:123:plan", "value": "enterprise"}
])

for result in results:
    if result.allowed:
        print(f"Stored: {result.key}")
    else:
        print(f"Blocked: {result.key} - {result.blocked_by}")

Deleting Data

# Delete a single key
anchor.data.delete(agent.id, "user:123:language")

# Delete all keys with a prefix
anchor.data.delete_prefix(agent.id, "session:abc:")

TypeScript Example

import { Anchor } from 'anchorai';

const anchor = new Anchor({ apiKey: 'anc_...' });
const agent = await anchor.agents.create('support-bot');

// Configure policies
await anchor.config.update(agent.id, {
  policies: {
    block_pii: true,
    block_secrets: true
  }
});

// Write data
const result = await anchor.data.write(agent.id, 'user:123:language', 'spanish');
if (result.allowed) {
  console.log(`Stored: ${result.auditId}`);
} else {
  console.log(`Blocked: ${result.blockedBy}`);
}

// Read data
const value = await anchor.data.read(agent.id, 'user:123:language');

// Search
const results = await anchor.data.search(agent.id, 'user preferences', { limit: 5 });

Memory vs Data Storage

You might be wondering: What's the difference between "memory" and "data storage"?

How Anchor Works with Memory Frameworks

Anchor works above memory frameworks like Mem0, LangChain Memory, and CrewAI:

• Memory frameworks answer: "What does the agent remember?"
• Anchor answers: "What is the agent allowed to remember, and can you prove it?"

Example: Mem0 Integration

When you use Mem0 with Anchor, Anchor wraps Mem0 to add governance:

from anchor import Anchor
from anchor.integrations.mem0 import AnchorMem0
from mem0 import Memory

anchor = Anchor(api_key="anc_...")
agent = anchor.agents.create("mem0-agent")

# Configure policies
anchor.config.update(agent.id, {
    "policies": {"block_pii": True, "block_secrets": True}
})

# Wrap Mem0 with Anchor governance
mem0_client = Memory()
wrapped = AnchorMem0(anchor=anchor, agent_id=agent.id, mem0_client=mem0_client)

# Mem0 handles semantic memory, retrieval, etc.
# Anchor adds: policy enforcement, audit trails, checkpoints
result = wrapped.add("User email is john@example.com", user_id="user_123")
if not result.allowed:
    print(f"Blocked by Anchor: {result.blocked_by}")  # PII detected
else:
    print("Stored in Mem0 with Anchor governance")

What Anchor Adds

• Policy Enforcement - Blocks PII, secrets before memory frameworks store them
• Audit Trail - Hash-chained logs of all memory operations
• Checkpoints - Snapshot and rollback memory state
• Retention Policies - Auto-delete memories after N days
• Compliance - Export and deletion certificates

In summary: Anchor doesn't replace memory frameworks—it governs them. Use Mem0/LangChain/CrewAI for semantic memory and retrieval, and Anchor for policy enforcement, audit trails, and compliance.

For more details, see the Policies guide and the Data API reference.