nined-memory 0.1.1

Python SDK for the 9D Labs Memory Runtime API (v1 + v2)

nined-memory

Python SDK for the 9D Labs Memory Runtime API. Supports both v1 and v2 APIs.

Give any AI agent persistent, auditable memory with deterministic context retrieval.

Install

pip install nined-memory

Zero dependencies. Uses only Python standard library.

v1 API (MemoryClient)

from nined.memory import MemoryClient

client = MemoryClient(base_url="https://api.9dlabs.xyz", api_key="your-key")

# Ingest a document
client.ingest(
    workspace_id="ws-1",
    actor_id="agent-1",
    artifact_type="document",
    raw_payload={"title": "Runbook", "content": "If auth fails, restart the pod."},
)

# Retrieve context (deterministic, budget-enforced)
pack = client.context_pack(query="What should I do about auth?", workspace_id="ws-1")
print(pack["context_text"])
print(f"Confidence: {pack['confidence']}")

# Audit trail
receipt = client.get_receipt(pack["receipt_id"])

v1 methods

Method	Description
ingest(workspace_id, actor_id, artifact_type, raw_payload, ...)	Ingest a single artifact
context_pack(query, workspace_id, max_tokens, mode, ...)	Retrieve context pack with rich tuning params
feedback(workspace_id, actor_id, action, ...)	Submit correction (pin, private, mark_wrong, ...)
list_receipts(workspace_id, limit, offset)	List decision receipts
get_receipt(receipt_id)	Get a single receipt
artifact_status(artifact_id)	Get indexing status
delete_workspace(workspace_id)	Delete all workspace data
ask(query, workspace_id, max_tokens)	LLM-synthesized answer
health()	Service liveness check
register(email)	Register tenant (multi-tenant mode)
regenerate_key(email)	Regenerate API key
billing_usage()	Current billing usage
billing_plan()	Current plan & entitlements

v2 API (MemoryClientV2)

from nined.memory import MemoryClientV2

client = MemoryClientV2(
    base_url="https://api.9dlabs.xyz",
    api_key="your-key",
    workspace_id="my-workspace",
)

# Batch ingest
client.ingest([
    {"artifact_type": "document", "raw_payload": {"content": "Deploy runbook..."}},
    {"artifact_type": "note", "raw_payload": {"content": "Meeting notes..."}},
])

# Retrieve with serving profile
pack = client.context_pack("What happened?", profile="high_recall")
for snippet in pack["snippets"]:
    print(snippet["content"])

# Async ingest + job tracking
result = client.ingest([...], async_index=True)
status = client.job_status(result["queued_jobs"][0]["job_id"])

v2 methods

Method	Description
ingest(artifacts, async_index=False)	Batch ingest (sync or async)
context_pack(query, max_tokens, profile)	Retrieve with serving profiles
feedback(action, artifact_id, ...)	Submit correction
list_receipts(limit, offset)	List receipts
get_receipt(pack_hash)	Get receipt by pack hash
artifact_status(artifact_id)	Indexing status
job_status(job_id)	Async job status
ask(query, max_tokens, profile)	LLM-synthesized answer
delete_workspace()	Delete workspace
health()	Liveness check
ready()	Readiness + storage health

Serving profiles (v2)

Profile	Use case
low_latency	Fastest retrieval, fewer results
balanced	Default — good recall with reasonable latency
high_recall	Maximum evidence, higher latency

Error handling

from nined.memory import MemoryClient, PolicyDeniedError, RateLimitError

try:
    pack = client.context_pack(query="...", workspace_id="...")
except PolicyDeniedError as e:
    print(f"Denied: {e.detail}")
except RateLimitError:
    print("Slow down!")

Exception	HTTP status	When
AuthorizationError	401, 403	Missing/invalid API key
PolicyDeniedError	403	Policy violation
NotFoundError	404	Resource not found
RateLimitError	429	Rate limit exceeded
MemoryAPIError	any	Base class for all API errors

License

MIT