claiv-memory 0.3.0

Official Python SDK for the Claiv Memory API (V3 - Priority Memory + Retention + Feedback)

claiv-memory

Official Python SDK for the Claiv Memory 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",
    "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",
    "task": "Help the user configure their editor",
    "token_budget": 2000,
})
for block in result["memory_blocks"]:
    print(f"[{block['type']}] {block['content']}")

# 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)

Parameter	Type	Default	Description
api_key	str	required	API key (sent as Bearer token)
base_url	str	https://api.claiv.io	API base URL
timeout	float	30.0	Request timeout in seconds
max_retries	int	2	Retries on 429/5xx (0 to disable)
http_client	httpx.Client	None	Custom httpx client

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

Core Methods

client.ingest(request) -> IngestResponse

result = client.ingest({
    "user_id": "user-123",         # required
    "type": "message",             # required: "message" | "tool_call" | "app_event"
    "content": "The actual text",  # required
    "thread_id": "thread-456",     # optional
    "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

result = client.recall({
    "user_id": "user-123",                  # required
    "task": "Help configure their editor",  # required
    "token_budget": 2000,                   # required: 200–8000
    "thread_id": "thread-456",              # optional
    "scope": {"project": "claiv"},          # optional
})
# result: {
#   "system_context": str,
#   "memory_blocks": [{"type", "content", "source_ids", "score"}, ...],
#   "citations": [str, ...],
#   "token_estimate": int,
# }

Memory block types: open_loop, fact, claim, episode, chunk.

client.forget(request) -> ForgetResponse

result = client.forget({
    "user_id": "user-123",                     # required
    "thread_id": "thread-456",                 # optional
    "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, ContextPack, MemoryBlock,
    ForgetRequest, ForgetResponse, DeletedCounts,
    UsageSummaryResponse, UsageBreakdownResponse, UsageLimitsResponse,
)