Python SDK

pip install engramma-cloud

import os
from engramma_cloud import EngrammaClient

client = EngrammaClient(api_key=os.environ['ENGRAMMA_API_KEY'])

result = client.store(
    text='The deployment window is Tuesday 2-4pm UTC',
    metadata={'category': 'ops', 'priority': 'high'},
    importance=0.9
)
print(result.pattern_id)  # pat_7f3a2b
print(result.confidence)  # 0.93

results = client.batch_store([
    {'text': 'Alice joined the team on Jan 5'},
    {'text': 'Project deadline is March 15', 'importance': 0.95},
    {'text': 'Budget approved for Q2', 'metadata': {'dept': 'finance'}}
])
print(f'Stored {results.stored} patterns')

results = client.retrieve(
    query='When can we deploy?',
    top_k=5,
    min_confidence=0.5,
    pathway='auto'
)
for r in results:
    print(f'{r.text} (confidence: {r.confidence}, pathway: {r.pathway})')

results = client.recall(
    query='What happened before the outage?',
    temporal_weight=0.8,
    causal_weight=0.7,
    limit=10
)
for r in results:
    print(f'{r.text} (relevance: {r.relevance_score})')

explanation = client.explain(
    query='When can we deploy?',
    level='detailed'  # 'brief', 'detailed', or 'debug'
)
print(explanation.summary)
print(explanation.pathway_scores)
print(explanation.reasoning_chain)

client.forget(pattern_id='pat_7f3a2b')
# or soft-delete (can be recovered within 30 days)
client.forget(pattern_id='pat_7f3a2b', soft=True)

results = client.text.search(
    query='deployment',
    filters={'category': 'ops'},
    sort='relevance',
    limit=20
)

summary = client.text.summarize(
    topic='Q1 project updates',
    max_length=200
)
print(summary.text)
print(f'Based on {summary.source_count} memories')

comparison = client.text.compare(
    pattern_id_a='pat_abc123',
    pattern_id_b='pat_def456'
)
print(f'Similarity: {comparison.similarity}')
print(f'Differences: {comparison.differences}')

# Tag memories
client.text.tag(pattern_id='pat_abc123', tags=['important', 'ops'])

# Create causal link
client.text.link(
    source_id='pat_abc123',
    target_id='pat_def456',
    relation='caused_by'
)

# Get timeline
timeline = client.text.timeline(
    topic='deployment',
    from_date='2026-01-01',
    to_date='2026-01-31'
)

# Preview what consolidation would do
preview = client.engine.consolidation.preview()
print(f'Would merge: {preview.merge_candidates}')
print(f'Would prune: {preview.prune_candidates}')

# Trigger consolidation
result = client.engine.consolidation.merge()
print(f'Merged: {result.merged}, Pruned: {result.pruned}')

# Prefetch for faster retrieval
client.engine.diagnostics.prefetch(queries=['deployment', 'budget'])

# Warmup cache
client.engine.diagnostics.warmup()

# Get memory space structure
structure = client.engine.diagnostics.structure()
print(f'Total patterns: {structure.total_patterns}')

# List organizations
orgs = client.organizations.list()

# Usage summary
usage = client.usage.summary()
print(f'Patterns: {usage.patterns.used}/{usage.patterns.limit}')

# Notifications
notifications = client.notifications.list(unread_only=True)
print(f'Unread: {notifications.unread_count}')

# List consents
consents = client.gdpr.consents()

# Request data export
export_req = client.gdpr.export(format='json')
print(f'Export ready in ~{export_req.estimated_time_minutes} min')

# Request erasure (irreversible after 7 days)
client.gdpr.erasure(confirm=True, reason='Account closure')

import asyncio
from engramma_cloud import AsyncEngrammaClient

async def main():
    client = AsyncEngrammaClient(api_key='nx_live_...')

    # All methods are awaitable
    result = await client.store('Async memory storage works')
    results = await client.retrieve('What was stored?')

    # Concurrent operations
    stores = await asyncio.gather(
        client.store('Fact A'),
        client.store('Fact B'),
        client.store('Fact C')
    )
    print(f'Stored {len(stores)} patterns concurrently')

    await client.close()

asyncio.run(main())

from engramma_cloud.exceptions import (
    EngrammaError,        # Base exception
    AuthenticationError,  # 401 — invalid or expired key
    PermissionError,      # 403 — insufficient scope
    NotFoundError,        # 404 — resource not found
    ValidationError,      # 422 — invalid parameters
    RateLimitError,       # 429 — rate limit exceeded
    ServerError           # 5xx — server-side error
)

try:
    result = client.retrieve(query='test')
except RateLimitError as e:
    print(f'Rate limited. Retry after {e.retry_after}s')
except AuthenticationError:
    print('Invalid API key')
except EngrammaError as e:
    print(f'Error {e.status_code}: {e.message}')

from engramma_cloud.types import (
    StoreResult,
    RetrieveResult,
    RecallResult,
    Explanation,
    UsageSummary,
    ConsentRecord
)

# Type hints work everywhere
result: StoreResult = client.store('typed result')
results: list[RetrieveResult] = client.retrieve('query')

# Access typed fields
result.pattern_id   # str
result.confidence   # float
result.pathway      # Literal['exact', 'energy', 'attention']

client = EngrammaClient(
    api_key='nx_live_...',
    max_retries=5,          # default: 3
    retry_backoff=2.0,      # backoff multiplier (default: 2.0)
    retry_max_wait=30       # max wait between retries in seconds
)

import logging
logging.basicConfig(level=logging.DEBUG)

# Or enable only for the SDK
logging.getLogger('engramma_cloud').setLevel(logging.DEBUG)