claiv-memory

Official Python SDK for the Claiv Memory API, aligned with the current V6.3 API.

Installation

pip install claiv-memory

Quick Start

from claiv import ClaivClient

client = ClaivClient(api_key="your-api-key")

# Store a memory
result = client.ingest({
    "user_id": "user-123",
    "conversation_id": "conv-session-456",
    "type": "message",
    "content": "User prefers dark mode and uses VS Code",
})
print(result["event_id"])

# Recall relevant memory
result = client.recall({
    "user_id": "user-123",
    "conversation_id": "conv-session-456",
    "query": "Help the user configure their editor",
})
# result["llm_context"]["text"] → pre-synthesized context, inject as system prompt
# result["answer_facts"]        → [{"fact_id", "subject", "predicate", "object_text", ...}]
system_prompt = result["llm_context"]["text"] or "No memory found."

# Forget memory for a user
result = client.forget({"user_id": "user-123"})
print(f"Deleted: {result['deleted_counts']}")

Async Support

from claiv import AsyncClaivClient

async with AsyncClaivClient(api_key="your-api-key") as client:
    result = await client.ingest({
        "user_id": "user-123",
        "type": "message",
        "content": "User prefers dark mode",
    })

API Reference

ClaivClient(*, api_key, base_url, timeout, max_retries, http_client)

AsyncClaivClient accepts the same parameters (with httpx.AsyncClient).

Core Methods

client.ingest(request) -> IngestResponse

result = client.ingest({
    "user_id": "user-123",         # required
    "conversation_id": "conv-session-456",  # required
    "type": "message",             # required: "message" | "tool_call" | "app_event"
    "content": "The actual text",  # required
    "project_id": "project-xyz",   # optional: for project-scoped facts
    "scope": "global",             # optional: "global" | "project" | "conversation"
    "metadata": {"source": "chat"},# optional
    "event_time": "2025-01-01T00:00:00Z",  # optional: ISO 8601
    "idempotency_key": "unique-1", # optional: prevents duplicates
})
# result: {"event_id": str, "deduped": bool}

client.recall(request) -> RecallResponse

Automatically includes matching document chunks in llm_context["text"] when documents exist for the user.

Document retrieval mode is auto-detected from the query:

• Semantic (default): top-K chunks by cosine similarity, controlled by limits["document_chunks"].
• Working-set: triggered by structural references ("chapter 3", "the introduction") — returns the full section plus a document summary and summaries of other sections.

result = client.recall({
    "user_id": "user-123",                    # required
    "conversation_id": "conv-session-456",    # required
    "project_id": "project-xyz",              # optional: include project-scoped facts
    "document_id": "doc_uuid",                # optional: restrict to a specific document
    "query": "Help configure their editor",   # required: natural-language question
    "reference_time": None,                   # optional: ISO datetime for temporal anchoring
    "limits": {
        "answer_facts": 12,                   # default: 12
        "document_chunks": 5,                 # default: 5 — max chunks in semantic mode (1–50)
    },
})
# result["llm_context"]["text"]  → synthesized narrative, inject as system prompt
# result["answer_facts"]         → [{"fact_id", "subject", "predicate", "object_text", ...}]
# result["supporting_facts"]     → corroborating facts
# result["background_context"]   → broader context facts

Use result["llm_context"]["text"] directly as your LLM system prompt:

system_prompt = result["llm_context"]["text"] or "No memory found."

client.upload_document(request) -> DocumentUploadResponse

Upload and index a document for RAG retrieval. Chunking and embedding happen synchronously — the document is fully indexed when the response returns.

result = client.upload_document({
    "user_id": "user-123",              # required
    "content": document_text,           # required: full document text (up to 5 MB)
    "document_name": "Product Manual",  # required: shown as citation in llm_context
    "document_id": "manual-v2",         # optional: stable ID; re-uploading replaces all chunks
    "conversation_id": "conv-session-456",  # optional: scope to this conversation
    "project_id": "project-xyz",        # optional: scope to this project
    "scope": "global",                  # optional: "global" | "project" | "conversation"
    "chunk_size": 800,                  # optional: target chars per chunk (200–4000)
    "chunk_overlap": 100,               # optional: overlap between chunks (0–500)
})
# result["document_id"]      → use on recall to target this document
# result["chunks_ingested"]  → number of chunks stored
# result["chunk_ids"]        → list of chunk UUIDs

# The document is now ready — no polling needed.
# Subsequent recalls will automatically include matching chunks in llm_context["text"].

# To target recall at this document only:
recall = client.recall({
    "user_id": "user-123",
    "conversation_id": "conv-session-456",
    "query": "What does the manual say about installation?",
    "document_id": result["document_id"],
})

# To delete the document later:
client.forget({
    "user_id": "user-123",
    "document_id": result["document_id"],
})

Async version:

result = await client.upload_document({
    "user_id": "user-123",
    "content": document_text,
    "document_name": "Product Manual",
})

client.forget(request) -> ForgetResponse

result = client.forget({
    "user_id": "user-123",                     # required
    "conversation_id": "conv-session-456",     # optional
    "project_id": "project-xyz",               # optional
    "document_id": "manual-v2",                # optional: removes all chunks for this document
    "from_time": "2025-01-01T00:00:00Z",       # optional
    "to_time": "2025-06-01T00:00:00Z",         # optional
})
# result: {"receipt_id": str, "deleted_counts": {...}}

Usage Methods

summary = client.get_usage_summary("30d")    # "7d" | "30d" | "month" | "today"
breakdown = client.get_usage_breakdown("today")
limits = client.get_usage_limits()

Health Check

result = client.health_check()  # no auth required
# {"ok": True}

Error Handling

All errors inherit from ClaivError.

from claiv import ClaivApiError, ClaivTimeoutError, ClaivNetworkError

try:
    client.ingest({...})
except ClaivApiError as e:
    print(e.status)      # HTTP status code
    print(e.code)        # "invalid_request" | "unauthorized" | "quota_exceeded" | ...
    print(e.request_id)  # server request ID for support
    print(e.details)     # validation errors, quota info, etc.
except ClaivTimeoutError:
    pass  # request timed out
except ClaivNetworkError:
    pass  # DNS failure, connection refused, etc.

Retries

The SDK automatically retries on 429 (rate limited) and 5xx (server error) responses with exponential backoff and jitter. Client errors (400, 401, 403, 404) are never retried.

# Default: 2 retries (3 total attempts)
client = ClaivClient(api_key="key")

# Disable retries
client = ClaivClient(api_key="key", max_retries=0)

# More retries for critical paths
client = ClaivClient(api_key="key", max_retries=5)

Context Manager

Both clients support context managers for automatic cleanup:

with ClaivClient(api_key="key") as client:
    client.ingest({...})

async with AsyncClaivClient(api_key="key") as client:
    await client.ingest({...})

Type Hints

All request/response types are exported as TypedDicts:

from claiv import (
    IngestRequest, IngestResponse,
    RecallRequest, RecallResponse, RecallFact, V6LLMContext, ContextPack,
    ForgetRequest, ForgetResponse, DeletedCounts,
    DocumentUploadRequest, DocumentUploadResponse,
    UsageSummaryResponse, UsageBreakdownResponse, UsageLimitsResponse,
)