Python SDK

pip install engramma-cloud

import os
from engramma_cloud import EngrammaClient

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

result = client.text.store(
    text='The deployment window is Tuesday 2-4pm UTC',
    metadata={'category': 'ops', 'priority': 'high'}
)
print(result.pattern_id)       # pat_7f3a2b...
print(result.embedding_dim)    # 384
print(result.patterns_used)    # 42
print(result.patterns_limit)   # 10000

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

response = client.text.retrieve(
    query='When can we deploy?',
    top_k=5
)
print(f'Latency: {response.latency_ms}ms')
for r in response.results:
    print(f'{r.text} (similarity: {r.similarity}, id: {r.pattern_id})')

response = client.text.recall(
    query='What happened before the outage?',
    top_k=10
)
for r in response.results:
    print(f'{r.text} (similarity: {r.similarity})')

explanation = client.text.explain(
    query='When can we deploy?',
    lang='en'
)
print(explanation.explanation)
print(explanation.method)      # e.g. 'cosine_similarity'
print(explanation.confidence)  # 0.93

client.text.forget(
    query='deployment window',
    category='ops',
    threshold=0.85
)

result = client.text.embed(text='The deployment window is Tuesday')
print(f'Dimension: {result.dim}')       # 384
print(f'Vector: {result.embedding[:5]}')  # first 5 values

result = client.text.similarity(
    text_a='The server crashed at 3am',
    text_b='There was a 3am outage on the server'
)
print(f'Similarity: {result.similarity}')  # 0.91

response = client.text.predict(
    query='What will happen next quarter?',
    top_k=5
)
for r in response.results:
    print(f'{r.text} (similarity: {r.similarity})')

stats = client.text.stats()
print(f'Total patterns: {stats.total_patterns}')
print(f'Embedding dim: {stats.embedding_dim}')
print(f'Storage: {stats.storage_bytes} bytes')

facts = client.text.important()
for fact in facts:
    print(f'{fact.text} (id: {fact.pattern_id})')

result = client.consolidation.sleep(
    mode='full',
    protect_categories=['critical', 'compliance']
)
print(f'Consolidation started: {result}')

client.consolidation.wake()

status = client.consolidation.status()
print(f'State: {status.state}')  # 'awake', 'sleeping', etc.

preview = client.consolidation.preview()
print(f'Merge candidates: {preview.merge_candidates}')
print(f'Patterns affected: {preview.patterns_affected}')

result = client.consolidation.merge_duplicates(threshold=0.85)
print(f'Merged: {result.merged_count} patterns')

regime = client.regime.current()
print(f'Current regime: {regime.name}')
print(f'Parameters: {regime.parameters}')

history = client.regime.history(last_n=10)
for entry in history:
    print(f'{entry.timestamp}: {entry.from_regime} -> {entry.to_regime}')

result = client.causal.strength(
    text_a='Server load increased to 95%',
    text_b='Response times degraded'
)
print(f'Causal strength: {result.strength}')  # 0.82

neighbors = client.causal.neighbors(
    text='The database connection pool was exhausted',
    direction='effects',     # 'causes' or 'effects'
    min_strength=0.6
)
for n in neighbors:
    print(f'{n.text} (strength: {n.strength})')

import asyncio
from engramma_cloud import AsyncEngrammaClient

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

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

    # Concurrent operations
    stores = await asyncio.gather(
        client.text.store(text='Fact A'),
        client.text.store(text='Fact B'),
        client.text.store(text='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:
    response = client.text.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,
    RetrieveResponse,
    RetrieveResult,
    ExplainResult,
    EmbedResult,
    TextStats
)

# Type hints work everywhere
result: StoreResult = client.text.store(text='typed result')
response: RetrieveResponse = client.text.retrieve(query='query')

# Access typed fields
result.pattern_id       # str
result.embedding_dim    # int
result.patterns_used    # int
result.patterns_limit   # int

for r in response.results:
    r.text        # str
    r.similarity  # float
    r.pattern_id  # str
    r.metadata    # dict | None

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)