Python SDK

The official Python SDK for the Engram Memory platform.

PyPI: engrammemory-ai Version: 0.2.0 Python: 3.9+ License: MIT

---

Installation

pip install engrammemory-ai

Dependencies: httpx>=0.25.0

---

Quick Start

from engrammemory import Engram

client = Engram(api_key="pr_live_xxxxx")

# Store a memory
result = client.store("User prefers dark mode and vim keybindings.")
print(result.id)  # "mem_8f3a..."

# Search memories
results = client.search("What are the user's preferences?")
for memory in results.memories:
    print(f"[{memory.score:.3f}] {memory.content}")

---

Constructor

client = Engram(
    api_key="pr_live_xxxxx",
    base_url="https://api.engrammemory.ai/v1",  # default
    timeout=30.0,       # seconds, default 30
    max_retries=3,      # default 3
)

---

Methods

store

Store a memory with optional classification and metadata.

result = client.store(
    text="The deployment pipeline uses GitHub Actions with Docker.",
    category="technical",
    importance=0.8,
    metadata={"project": "infra", "author": "eddy"},
    collection="engineering-notes"
)

print(result.id)         # "mem_a1b2c3..."
print(result.category)   # "technical"
print(result.stored)     # True

Parameters:

Returns: MemoryResult

---

search

Semantic search across stored memories.

results = client.search(
    query="deployment pipeline",
    limit=10,
    category="technical",
    min_score=0.7,
    collection="engineering-notes"
)

print(results.total)  # Number of matches
for memory in results.memories:
    print(f"[{memory.score:.3f}] {memory.content}")
    print(f"  Category: {memory.category}")
    print(f"  ID: {memory.id}")

Parameters:

Returns: SearchResult

---

recall

Alias for search. Provided for semantic convenience in agent workflows.

memories = client.recall(
    query="What does the user prefer for editors?",
    limit=5,
    min_score=0.6
)

for memory in memories.memories:
    print(memory.content)

Parameters: Same as search.

Returns: SearchResult

---

forget

Delete memories by ID or by semantic query match.

# Forget by ID
result = client.forget(memory_id="mem_a1b2c3")
print(result.deleted)  # 1

# Forget by query (deletes best match)
result = client.forget(query="deployment pipeline", collection="engineering-notes")
print(result.deleted)  # Number of memories removed

Parameters:

At least one of memory_id or query is required.

Returns: ForgetResult

---

connect

Health check. Verifies API key and connectivity.

status = client.connect()
print(status.connected)   # True
print(status.latency_ms)  # 42
print(status.tier)         # "builder"

Returns: ConnectionResult

---

consolidate

Trigger memory consolidation. Merges duplicates and optimizes storage.

result = client.consolidate(collection="engineering-notes")
print(result.merged)       # 3 (memories merged)
print(result.removed)      # 2 (duplicates removed)
print(result.total_after)  # 145

Parameters:

Returns: ConsolidateResult

---

embed

Generate an embedding vector without storing anything.

result = client.embed("The quick brown fox jumps over the lazy dog.")
print(len(result.vector))    # 768
print(result.model)          # "engram-embed-v1"
print(result.dimensions)     # 768

Parameters:

Returns: EmbedResult

---

intelligence

Full pipeline in one call: embed, classify, deduplicate, compress, and store.

result = client.intelligence(
    text="User switched from VS Code to Neovim on 2026-03-15.",
    check_dedup=True,
    compress=True,
    target_bits=4
)

print(result.memory_id)          # "mem_8f3a..."
print(result.category)           # "decision"
print(result.is_duplicate)       # False
print(result.compressed)         # True
print(result.compression_ratio)  # 6.2
print(result.cosine_preserved)   # 0.9947

Parameters:

Returns: IntelligenceResult

---

Async Client

AsyncEngram provides the same interface with async/await support.

import asyncio
from engrammemory import AsyncEngram

async def main():
    client = AsyncEngram(api_key="pr_live_xxxxx")

    # Store
    result = await client.store("Async memory storage works great.")
    print(result.id)

    # Search
    results = await client.search("async storage")
    for memory in results.memories:
        print(memory.content)

    # Intelligence pipeline
    intel = await client.intelligence(
        text="User prefers async Python patterns.",
        check_dedup=True
    )
    print(intel.category)

asyncio.run(main())

All methods on AsyncEngram accept the same parameters as Engram.

---

Models

MemoryResult

class MemoryResult:
    id: str              # Memory ID ("mem_...")
    stored: bool         # Whether the memory was stored
    category: str        # Assigned or auto-classified category
    created_at: str      # ISO 8601 timestamp

SearchResult

class SearchResult:
    memories: list[Memory]  # List of matching memories
    total: int              # Total matches found
    query: str              # Original query

class Memory:
    id: str
    content: str
    score: float          # Similarity score 0.0-1.0
    category: str
    metadata: dict
    created_at: str

EmbedResult

class EmbedResult:
    vector: list[float]   # Embedding vector
    dimensions: int       # Vector dimensions (768)
    model: str            # Model used for embedding

IntelligenceResult

class IntelligenceResult:
    memory_id: str
    category: str
    is_duplicate: bool
    compressed: bool
    compression_ratio: float
    cosine_preserved: float

ForgetResult

class ForgetResult:
    deleted: int          # Number of memories deleted

ConnectionResult

class ConnectionResult:
    connected: bool
    latency_ms: int
    tier: str

ConsolidateResult

class ConsolidateResult:
    merged: int
    removed: int
    total_after: int

---

Exceptions

All exceptions inherit from EngramError.

from engrammemory import (
    EngramError,
    AuthenticationError,
    RateLimitError,
    NotFoundError,
    ValidationError,
)

try:
    result = client.store("Some memory")
except AuthenticationError:
    print("Invalid or expired API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except NotFoundError:
    print("Resource not found")
except ValidationError as e:
    print(f"Invalid input: {e.message}")
except EngramError as e:
    print(f"Unexpected error: {e}")

---

Multi-Agent Example

Use collection namespaces to isolate memory per agent or project.

from engrammemory import Engram

client = Engram(api_key="pr_live_xxxxx")

# Agent 1: Research assistant
client.store(
    text="Found 3 papers on transformer attention mechanisms.",
    category="research",
    collection="agent-researcher"
)

# Agent 2: Code assistant
client.store(
    text="Refactored the auth module to use JWT tokens.",
    category="engineering",
    collection="agent-coder"
)

# Agent 3: Coordinator queries across all agents
research = client.search("attention mechanisms", collection="agent-researcher")
code = client.search("authentication changes", collection="agent-coder")

# Or search globally (no collection filter)
everything = client.search("recent changes", limit=20)

---

Environment Variables

The SDK reads these environment variables as fallbacks:

import os
os.environ["ENGRAM_API_KEY"] = "pr_live_xxxxx"

# No api_key needed — reads from environment
client = Engram()

---

Next Steps

• JavaScript SDK — TypeScript/Node.js SDK
• API Reference — Full endpoint documentation
• MCP Server — Connect Engram to AI agents via MCP