Python SDK - Thrindex docs

Installation

pip install thrindex

Requirements: Python 3.10+, httpx >= 0.28, pydantic >= 2.10.

Clients

The SDK provides two clients with identical method signatures:

Class	Module	Use case
`Thrindex`	`thrindex`	Synchronous — standard Python scripts, Django, Flask
`AsyncThrindex`	`thrindex`	Asynchronous — FastAPI, LangGraph, async frameworks

from thrindex import Thrindex, AsyncThrindex

Synchronous client

from thrindex import Thrindex

client = Thrindex(
    api_key="th_live_...",         # required
    base_url="https://api.thrindex.com",  # optional, this is the default
    timeout=30.0,                  # optional, seconds (default 30)
    max_retries=3,                 # optional (default 3)
)

The client is thread-safe and uses connection pooling internally. Create one instance and reuse it across your application.

Constructor parameters

Parameter	Type	Default	Description
`api_key`	`str`	required	Your API key beginning with `th_live_`
`base_url`	`str`	`"https://api.thrindex.com"`	Override for self-hosted or test environments
`timeout`	`float`	`30.0`	Per-request timeout in seconds
`max_retries`	`int`	`3`	Maximum retry attempts on `5xx` and network errors

Context manager

with Thrindex(api_key="th_live_...") as client:
    client.add(content="...", agent_id="...", user_id="...")
# Connection pool is closed automatically

Asynchronous client

from thrindex import AsyncThrindex

async with AsyncThrindex(api_key="th_live_...") as client:
    memory_id = await client.add(
        content="User prefers dark mode.",
        agent_id="agent-1",
        user_id="user-42",
    )

Same constructor parameters and method signatures as Thrindex. Use await on every method. Rate-limit waits and retry backoffs use asyncio.sleep — the event loop is never blocked.

Methods

`add()`

Store a memory for asynchronous processing.

memory_id: str = client.add(
    content="User prefers weekly email digests.",  # required
    agent_id="support-agent",                      # required
    user_id="user-42",                             # required
    confidence=0.95,                               # optional, float 0–1, default 1.0
    metadata={"source": "preference-survey"},      # optional, dict
    extract=True,                                  # optional, default True
)

Returns the memory UUID as a string. The memory is enqueued and available for search within a few seconds.

Parameters

Parameter	Type	Required	Default	Description
`content`	`str`	yes	—	The memory text. Max 8192 bytes (≈8 KB).
`agent_id`	`str`	yes	—	Agent identifier. Max 256 chars. Charset: `[A-Za-z0-9._:-]`
`user_id`	`str`	yes	—	User identifier. Same constraints as `agent_id`.
`confidence`	`float \| None`	no	`1.0` server-side	Your confidence in this memory’s accuracy, `0.0`–`1.0`.
`metadata`	`dict \| None`	no	`None`	Arbitrary key-value pairs stored with the memory.
`extract`	`bool`	no	`True`	When `True`, the cognition worker runs LLM extraction to transform raw content into discrete facts. Set to `False` to store content verbatim with no LLM call.

Returns

str — the memory UUID, e.g. "3f2e1d0c-1234-5678-abcd-ef0123456789".

`search()`

Retrieve the most relevant memories for a query.

from thrindex import SearchHit

results: list[SearchHit] = client.search(
    query="how does the user want to be contacted",  # required
    agent_id="support-agent",                         # required
    user_id="user-42",                                # required
    k=10,                                             # optional, default 10
    task_context="composing a notification email",    # optional
    level=None,                                       # optional, filter by level
)

for hit in results:
    print(hit.content, hit.score)

Parameters

Parameter	Type	Required	Default	Description
`query`	`str`	yes	—	Natural language query. Max 2048 characters.
`agent_id`	`str`	yes	—	Limits search to memories belonging to this agent.
`user_id`	`str`	yes	—	Limits search to memories belonging to this user.
`k`	`int`	no	`10`	Number of results to return. Range: `1`–`100`.
`task_context`	`str \| None`	no	`None`	A sentence describing what the agent is currently doing. Improves ranking for the current task.
`level`	`int \| None`	no	`None` (all levels)	Filter results to a specific memory level: `0` (raw), `1` (abstraction).

Returns

list[SearchHit] — ordered by descending relevance score.

`SearchHit` fields

Field	Type	Description
`id`	`str`	Memory UUID
`content`	`str`	The memory text
`agent_id`	`str`	Agent that owns the memory
`user_id`	`str`	User the memory belongs to
`level`	`int`	`0` or `1`
`status`	`str`	`"active"` or `"deprecated"`
`importance`	`float`	Importance score `0.0`–`1.0` assigned by the cognition worker
`confidence`	`float`	Confidence score `0.0`–`1.0`
`score`	`float`	Multi-signal fusion score — use this to compare results within a single response
`created_at`	`datetime`	When the memory was created

`list()`

List memories with cursor-based pagination.

from thrindex import ListResponse

page: ListResponse = client.list(
    agent_id="support-agent",  # optional
    user_id="user-42",         # optional
    limit=50,                  # optional, default 50, max 200
    cursor=None,               # optional, from previous page's next_cursor
)

for memory in page.results:
    print(memory.id, memory.importance)

# Iterate through all pages
while page.next_cursor:
    page = client.list(
        agent_id="support-agent",
        user_id="user-42",
        cursor=page.next_cursor,
    )

Parameters

Parameter	Type	Required	Default	Description
`agent_id`	`str \| None`	no	`None`	Filter to a specific agent. Omit to list all agents.
`user_id`	`str \| None`	no	`None`	Filter to a specific user. Omit to list all users.
`limit`	`int`	no	`50`	Results per page. Range: `1`–`200`.
`cursor`	`str \| None`	no	`None`	Opaque cursor from `next_cursor` of a previous response. Omit for the first page.

Returns

ListResponse with:

results: list[MemoryRecord] — current page of memories (content excluded)
next_cursor: str | None — pass to the next call to get the next page; None when no more pages

content is intentionally excluded from list results. Call get(memory_id) to retrieve the full record including content.

`MemoryRecord` fields

Field	Type	Description
`id`	`str`	Memory UUID
`org_id`	`str`	Your organization identifier
`agent_id`	`str`	Agent that owns the memory
`user_id`	`str`	User the memory belongs to
`level`	`int`	`0` or `1`
`status`	`str`	`"active"` or `"deprecated"`
`importance`	`float`	Importance score `0.0`–`1.0`
`confidence`	`float`	Confidence score `0.0`–`1.0`
`source`	`str`	Origin of the memory (e.g. `"agent"`)
`created_at`	`datetime`	Creation timestamp
`updated_at`	`datetime`	Last update timestamp

`get()`

Fetch a single memory by ID, including its content.

from thrindex import MemoryDetail

memory: MemoryDetail = client.get("3f2e1d0c-1234-5678-abcd-ef0123456789")
print(memory.content)

Parameters

Parameter	Type	Required	Description
`memory_id`	`str`	yes	The memory UUID

Returns

MemoryDetail — all fields of MemoryRecord plus content: str. Raises NotFoundError if the memory does not exist or belongs to a different organization.

`delete()`

Delete a memory by ID.

client.delete("3f2e1d0c-...")          # soft delete (default)
client.delete("3f2e1d0c-...", hard=True)  # hard delete (GDPR erasure)

Parameters

Parameter	Type	Required	Default	Description
`memory_id`	`str`	yes	—	The memory UUID
`hard`	`bool`	no	`False`	`False` = soft delete (status → `deprecated`, excluded from search). `True` = permanent erasure from all stores. Irreversible.

Returns

None. Raises NotFoundError if the memory does not exist.

`batch_add()`

Add multiple memories sequentially with per-item retry.

ids: list[str] = client.batch_add(
    memories=[
        {"content": "User timezone is Europe/Berlin."},
        {"content": "User language is German.", "confidence": 0.9},
        {"content": "User prefers dark mode.", "extract": False},
    ],
    agent_id="support-agent",  # default for all items
    user_id="user-42",         # default for all items
    extract=True,              # default extraction behaviour, per-item can override
)

Each item in memories must have a "content" key. Items may override agent_id, user_id, confidence, metadata, and extract individually. Returns a list[str] of memory UUIDs in the same order as the input. Raises ThrindexError on the first unretriable failure.

Error handling

from thrindex import (
    ThrindexError,
    AuthError,
    NotFoundError,
    RateLimitError,
)

try:
    results = client.search(query="...", agent_id="...", user_id="...")
except AuthError:
    # 401 — invalid or revoked API key
    print("Check your API key.")
except NotFoundError:
    # 404 — memory does not exist
    print("Memory not found.")
except RateLimitError as e:
    # 429 — rate limit exceeded (SDK already retried)
    print(f"Rate limited. Retry after {e.retry_after}s.")
except ThrindexError as e:
    # All other errors (400, 500, network failures)
    print(f"Error {e.status_code}: {e}")

Exception hierarchy

Exception	HTTP	Description
`ThrindexError`	any	Base class. Has `.status_code: int \| None`.
`AuthError`	401	Invalid, revoked, or missing API key; or missing scope.
`NotFoundError`	404	Memory or resource does not exist.
`RateLimitError`	429	Rate limit exceeded. Has `.retry_after: float` (seconds).

Retry behaviour

The SDK automatically retries on transient failures:

5xx errors (500, 502, 503, 504): up to max_retries times with exponential backoff + jitter
429 Rate limited: waits for the Retry-After header duration, then retries
Network errors (connection reset, timeout): retries with exponential backoff

4xx errors (except 429) are not retried. If max_retries is exhausted, the final exception is raised. Backoff formula: min(0.5 × 2^attempt + jitter, 30s).

Closing the client

# Explicit close
client.close()

# Context manager (recommended)
with Thrindex(api_key="th_live_...") as client:
    ...

# Async context manager
async with AsyncThrindex(api_key="th_live_...") as client:
    ...

Always close the client when done to release the connection pool. In long-running servers, create one client at startup and close it on shutdown.

​Installation

​Clients

​Synchronous client

​Constructor parameters

​Context manager

​Asynchronous client

​Methods

​add()

​Parameters

​Returns

​search()

​Parameters

​Returns

​SearchHit fields

​list()

​Parameters

​Returns

​MemoryRecord fields

​get()

​Parameters

​Returns

​delete()

​Parameters

​Returns

​batch_add()

​Error handling

​Exception hierarchy

​Retry behaviour

​Closing the client

Installation

Clients

Synchronous client

Constructor parameters

Context manager

Asynchronous client

Methods

`add()`

Parameters

Returns

`search()`

Parameters

Returns

`SearchHit` fields

`list()`

Parameters

Returns

`MemoryRecord` fields

`get()`

Parameters

Returns

`delete()`

Parameters

Returns

`batch_add()`

Error handling

Exception hierarchy

Retry behaviour

Closing the client