Python SDK

pip install engram-subnet
# For PDF support
pip install engram-subnet pypdf

from engram.sdk import EngramClient
client = EngramClient(
    miner_url="http://72.62.2.34:8091",   # or use from_subnet() for auto-discovery
    timeout=30.0,
)

# One line — no miner URL needed
client = EngramClient.from_subnet(netuid=450)

private = EngramClient(
    "http://miner:8091",
    namespace="company-docs",
    namespace_key="your-secret-key-min-16-chars",
)
cid = private.ingest("Q4 revenue was $4.2M")  # encrypted before leaving your machine
results = private.query("revenue figures")      # decrypted client-side

cid: str = client.ingest(text: str, metadata: dict = None)

cid = client.ingest(
    "BERT uses bidirectional encoder representations.",
    metadata={"source": "arxiv", "year": "2018"}
)
print(cid)  # v1::a3f2b1c4d5e6f7...

result = client.ingest_image(
    "photo.jpg",                      # path, or raw bytes
    xai_api_key="xai-...",            # get one at console.x.ai
    metadata={"user_id": "u_123"},    # optional extra metadata
)
print(result["cid"])          # v1::a3f2b1... — use this for search
print(result["description"])  # "A photograph of a whiteboard showing..."
print(result["content_cid"])  # sha256:abc123... — integrity check
print(result["filename"])     # "photo.jpg"
# Search by what's in the image later:
results = client.query("whiteboard diagram with architecture")

pip install pypdf

result = client.ingest_pdf(
    "research_paper.pdf",             # path, or raw bytes
    metadata={"category": "research"},
)
print(result["cid"])          # v1::...
print(result["pages"])        # 12
print(result["chars"])        # 48293
print(result["content_cid"])  # sha256:...
# Search the PDF content later:
results = client.query("transformer attention mechanism")

result = client.ingest_url(
    "https://arxiv.org/abs/1706.03762",
    metadata={"category": "research"},
)
print(result["cid"])    # v1::...
print(result["title"])  # "Attention Is All You Need"
print(result["chars"])  # 6842
print(result["url"])    # final URL after redirects
# Search later:
results = client.query("transformer architecture paper")

messages = [
    {"role": "user",      "content": "What's the capital of France?"},
    {"role": "assistant", "content": "The capital of France is Paris."},
    {"role": "user",      "content": "Tell me more about Paris."},
]
cids = client.ingest_conversation(
    messages,
    session_id="session_abc123",
    metadata={"user_id": "u_456"},
)
print(cids)
# ["v1::a3f2...", "v1::b2e8...", "v1::c9f4..."]
# Retrieve conversation turns later:
results = client.query("capital city France")
# Returns the turn that mentioned Paris

results: list[dict] = client.query(text: str, top_k: int = 10, filter: dict = None)

# Basic search
results = client.query("how does self-attention work?", top_k=10)
# [
#   {"cid": "v1::a3f2b1...", "score": 0.9821, "metadata": {"source": "arxiv"}},
#   {"cid": "v1::b2e8c1...", "score": 0.8847, "metadata": {"type": "url"}},
# ]
# Filter by metadata — AND semantics (all conditions must match)
results = client.query(
    "revenue figures",
    top_k=5,
    filter={"user_id": "u_123", "type": "text"},
)
# Only conversation turns for a specific session
turns = client.query(
    "Paris",
    filter={"session_id": "session_abc123", "role": "assistant"},
)

record = client.get("v1::a3f2b1c4d5e6f7...")
if record:
    print(record["cid"])       # v1::a3f2b1...
    print(record["metadata"])  # {"source": "arxiv", "title": "Attention Is All You Need"}
else:
    print("Not found")

deleted = client.delete("v1::a3f2b1c4d5e6f7...")
print(deleted)  # True if it existed, False if not found

# All records (first page)
records = client.list(limit=50, offset=0)
# Filter by type
image_records = client.list(filter={"type": "image"})
# All memories for a user, paginated
page1 = client.list(filter={"user_id": "u_123"}, limit=20, offset=0)
page2 = client.list(filter={"user_id": "u_123"}, limit=20, offset=20)
for r in page1:
    print(r["cid"], r["metadata"].get("title", ""))

# data.jsonl format:
# {"text": "First entry"}
# {"text": "Second entry", "metadata": {"category": "ml"}}
cids = client.batch_ingest_file("data/corpus.jsonl")
print(f"Ingested {len(cids)} records")
# With error tracking
cids, errors = client.batch_ingest_file("corpus.jsonl", return_errors=True)
for err in errors:
    print(f"Skipped: {err}")

# Check liveness — raises MinerOfflineError if unreachable
info = client.health()
# {"status": "ok", "vectors": 42156, "uid": 7}
# Safe check — never raises
if client.is_online():
    cid = client.ingest("...")

from engram.sdk import EngramClient, MinerOfflineError
miners = [
    EngramClient("http://miner1:8091"),
    EngramClient("http://miner2:8091"),
    EngramClient("http://miner3:8091"),
]
cids = []
for miner in miners:
    try:
        cids.append(miner.ingest("Critical knowledge."))
    except MinerOfflineError:
        print(f"Miner offline: {miner.miner_url}")
print(f"Stored on {len(cids)}/3 miners")