xtrace_sdk.integrations¶

Submodules¶

xtrace_sdk.integrations.xtrace

Classes¶

XTraceIntegration

Async HTTP client for the XTrace encrypted vector database API.

Package Contents¶

class xtrace_sdk.integrations.XTraceIntegration(org_id, api_key=None, admin_key=None, api_url='https://api.production.xtrace.ai', admin_api_url=None)¶

Async HTTP client for the XTrace encrypted vector database API.

Handles all network communication with XTrace: uploading encrypted chunks, querying Hamming distances, metadata search, and execution context management.

Supports use as an async context manager for automatic session cleanup:

async with XTraceIntegration(org_id="your_org_id") as xtrace:
    await xtrace.store_db(...)

Parameters:

org_id (str) – Your XTrace organisation ID.
api_key (str | None) – XTrace API key. If omitted, read from the XTRACE_API_KEY environment variable.
api_url (str) – API base URL, defaults to https://api.production.xtrace.ai.
admin_key (str | None)
admin_api_url (str | None)

org_id¶

api_url = 'https://api.production.xtrace.ai'¶

admin_api_url¶

batch_size = 500¶

session: aiohttp.ClientSession | None = None¶

_loop: asyncio.AbstractEventLoop | None = None¶

_lock¶

api_key = None¶

admin_key: str | None¶

async init_session()¶

Ensure a live aiohttp.ClientSession exists for the current event loop.

Creates or recreates the session if it is closed or belongs to a different loop. Called automatically before every API request.

Returns:: The active client session.
Return type:: aiohttp.ClientSession

async close()¶

Close the underlying HTTP session and release all connections.

Return type:: None

async __aenter__()¶

Return type:: XTraceIntegration

async __aexit__(exc_type, exc, tb)¶

Parameters:

exc_type (Any)
exc (Any)
tb (Any)

Return type:

None

_headers()¶

Return type:: dict[str, str]

_admin_headers()¶

Return type:: dict[str, str]

_encode_index(index_item)¶

Parameters:: index_item (gmpy2.mpz)
Return type:: str

_preprocess_chunks(db, index, update=False)¶

Parameters:

db (xtrace_sdk.x_vec.utils.xtrace_types.EncryptedDB)
index (xtrace_sdk.x_vec.utils.xtrace_types.EncryptedIndex)
update (bool)

Return type:

list[list[dict]]

async get_chunk_by_id(chunk_ids, kb_id, projection=None)¶

Fetch one or more chunks by their IDs.

Parameters:

chunk_ids (list[int]) – List of integer chunk IDs to retrieve.
kb_id (str) – Knowledge-base ID the chunks belong to.
projection (list[str], optional) – Optional list of fields to include in the response (e.g. ["chunk_id", "meta_data"]). All fields returned if omitted.

Returns:

List of chunk dicts. The index field, if present, is base64-decoded to bytes.

Return type:

list[dict]

async store_db(db, index, kb_id, context_id, concurrent=False)¶

Upload an encrypted database to XTrace, automatically batching into groups of 500.

Parameters:

db (EncryptedDB) – Encrypted document collection produced by DataLoader.
index (EncryptedIndex) – Encrypted embedding vectors, one list per chunk.
kb_id (str) – Destination knowledge-base ID.
context_id (str) – Execution context ID to associate with the data.
concurrent (bool, optional) – If True, upload batches concurrently with asyncio.gather. Defaults to sequential upload.

Returns:

List of server responses, one per batch.

Return type:

list[dict]

async insert_chunks(kb_id, chunks, context_id)¶

Insert a pre-batched list of chunks. Retries up to 3 times on server errors.

Prefer store_db() for most use cases — it handles batching automatically.

Parameters:

kb_id (str) – Destination knowledge-base ID.
chunks (list[dict]) – List of serialised chunk dicts ready for the API.
context_id (str) – Execution context ID to associate with the data.

Returns:

Server response dict.

Return type:

dict

async update_chunks(index_updates, chunk_updates, context_id, kb_id)¶

Replace existing chunks with updated content and re-encrypted vectors.

Each entry in chunk_updates must include a chunk_id field.

Parameters:

index_updates (EncryptedIndex) – New encrypted embedding vectors, one list per chunk.
chunk_updates (EncryptedDB) – Updated chunk dicts, each containing a chunk_id.
context_id (str) – Execution context ID.
kb_id (str) – Knowledge-base ID the chunks belong to.

Returns:

List of server responses, one per batch.

Return type:

list[dict]

async delete_chunks(chunk_ids, kb_id)¶

Delete chunks by ID from a knowledge base.

Parameters:

chunk_ids (list[int]) – IDs of the chunks to delete.
kb_id (str) – Knowledge-base ID the chunks belong to.

Returns:

Server response dict.

Return type:

dict

async patch_chunks(kb_id, context_id, chunk_patches)¶

Apply partial updates to existing chunks (e.g. metadata only).

Parameters:

kb_id (str) – Knowledge-base ID the chunks belong to.
context_id (str) – Execution context ID.
chunk_patches (list[dict]) – List of patch dicts, each containing a chunk_id and the fields to update.

Returns:

Server response dict.

Return type:

dict

async compute_hamming_distances(query, context_id, kb_id, meta_filter=None, range=None)¶

Submit an encrypted query vector and retrieve encrypted Hamming distances.

The server computes distances between query and all stored vectors without decrypting either side. Results must be decrypted client-side with the homomorphic secret key.

Parameters:

query (list[gmpy2.mpz]) – Encrypted query vector produced by the homomorphic client.
context_id (str) – Execution context ID that matches the stored data.
kb_id (str) – Knowledge-base ID to search.
meta_filter (str | dict | None) – Optional metadata filter expression (dict or JSON string).
range (list[int] | None) – Optional [min, max] range to limit which chunks are searched.

Returns:

List of (chunk_id, encrypted_distance) tuples.

Return type:

list[tuple[int, list[int | bytes]]]

async meta_search(kb_id, meta_filter, context_id, return_content=False, projection=None)¶

Search chunks by metadata filter expression, returning all matching results.

Parameters:

kb_id (str) – Knowledge-base ID to search.
meta_filter (str or dict) – Filter expression as a dict or JSON string. Uses MongoDB-style operators ($eq, $in, $and, etc.).
context_id (str) – Execution context ID.
return_content (bool, optional) – If True, include the (AES-encrypted) chunk content in results. Defaults to False.
projection (list[str], optional) – Optional list of fields to include in each result.

Returns:

List of matching chunk dicts.

Return type:

list[dict]

async meta_search_paginated(kb_id, context_id, meta_filter=None, projection=None, limit=None, offset=0, sort_order='desc', sort_by='chunk_id', return_content=False)¶

Search chunks by metadata filter with pagination support.

Parameters:

kb_id (str) – Knowledge-base ID to search.
context_id (str) – Execution context ID.
meta_filter (str or dict, optional) – Optional filter expression as a dict or JSON string.
projection (list[str], optional) – Optional list of fields to include in each result.
limit (int, optional) – Maximum number of results to return. Returns all if omitted.
offset (int, optional) – Number of results to skip (for pagination), defaults to 0.
sort_order (str, optional) – "asc" or "desc", defaults to "desc".
sort_by (str, optional) – Field to sort by, defaults to "chunk_id".
return_content (bool, optional) – If True, include the (AES-encrypted) chunk content in results. Defaults to False.

Returns:

Dict containing results (list of chunk dicts) and pagination metadata.

Return type:

dict

async delete_chunks_by_meta(kb_id, context_id, meta_filter, dry_run=False)¶

Delete all chunks matching a metadata filter.

Parameters:

kb_id (str) – Knowledge-base ID to delete from.
context_id (str) – Execution context ID.
meta_filter (str or dict) – Filter expression as a dict or JSON string.
dry_run (bool, optional) – If True, return a count of chunks that would be deleted without actually deleting them. Defaults to False.

Returns:

Server response dict (includes deleted_count or would_delete_count).

Return type:

dict

async patch_chunks_by_meta(kb_id, context_id, meta_filter, patch, dry_run=False)¶

Apply a metadata patch to all chunks matching a filter expression.

Parameters:

kb_id (str) – Knowledge-base ID to update.
context_id (str) – Execution context ID.
meta_filter (str or dict) – Filter expression as a dict or JSON string.
patch (dict) – Metadata fields to set on each matching chunk.
dry_run (bool, optional) – If True, return a count of chunks that would be updated without modifying anything. Defaults to False.

Returns:

Server response dict.

Return type:

dict

async store_exec_context(exec_context_dict, context_id)¶

Upload a serialised execution context to XTrace remote storage.

The secret key inside exec_context_dict must already be AES-encrypted before calling this method. Use save_to_remote() instead of calling this directly.

Parameters:

exec_context_dict (dict) – Encrypted execution context produced by to_dict_enc().
context_id (str) – Desired context ID (typically the SHA-256 fingerprint of the context).

Returns:

The context_id confirmed by the server.

Return type:

str

async get_serialized_exec_context(context_id)¶

Fetch a serialised execution context from XTrace remote storage.

Parameters:: context_id (str) – ID of the context to retrieve.
Returns:: Encrypted execution context dict (secret key is still AES-encrypted).
Return type:: dict

async list_exec_contexts()¶

List all execution context IDs stored under this organisation.

Returns:: List of context ID strings.
Return type:: list[str]

async delete_exec_context(context_id)¶

Delete an execution context from XTrace remote storage.

Parameters:: context_id (str) – ID of the context to delete.
Returns:: True on success.
Return type:: bool

async get_ctx_for_kb(kb_id)¶

Return the execution context IDs bound to a knowledge base.

Parameters:: kb_id (str) – Knowledge-base ID to look up.
Returns:: List of context ID strings, or None if the lookup fails.
Return type:: list[str] or None

async get_kbs_for_ctx(ctx_id)¶

Return the knowledge-base IDs bound to an execution context.

Parameters:: ctx_id (str) – Execution context ID to look up.
Returns:: List of knowledge-base ID strings, or None if the lookup fails.
Return type:: list[str] or None

async register_kb_binding(kb_id, ctx_id)¶

Bind a knowledge base to an execution context on the server.

Parameters:

kb_id (str) – Knowledge-base ID to bind.
ctx_id (str) – Execution context ID to bind to.

Returns:

True on success, False otherwise.

Return type:

bool

async list_kbs()¶

List all knowledge bases for the organisation.

Requires an admin key (admin_key constructor arg or XTRACE_ADMIN_KEY env var).

Returns:: List of KB records.
Return type:: list[dict]

async get_kb(kb_id)¶

Fetch metadata for a single knowledge base.

Requires an admin key (admin_key constructor arg or XTRACE_ADMIN_KEY env var).

Parameters:: kb_id (str) – Knowledge base ID to fetch.
Returns:: KB record dict.
Return type:: dict

async list_api_keys()¶

List all API keys for the organisation.

Requires an admin key (admin_key constructor arg or XTRACE_ADMIN_KEY env var).

Returns:: List of API key records.
Return type:: list[dict]

async get_key_permissions(key_hash)¶

Get KB permissions for an API key.

Requires an admin key (admin_key constructor arg or XTRACE_ADMIN_KEY env var).

Parameters:: key_hash (str) – Hash of the API key.
Returns:: List of permission records.
Return type:: list[dict]

async set_key_permission(key_hash, kb_id, permission)¶

Set KB permission for an API key.

Requires an admin key (admin_key constructor arg or XTRACE_ADMIN_KEY env var).

Parameters:

key_hash (str) – Hash of the API key to update.
kb_id (str) – Knowledge base ID to set permission on.
permission (int) – Permission level: 1=read, 3=write, 7=delete.

Returns:

Server response dict.

Return type:

dict

async create_kb(name, description='')¶

Create a new knowledge base.

Requires an admin key (admin_key constructor arg or XTRACE_ADMIN_KEY env var).

Parameters:

name (str) – Human-readable name for the knowledge base.
description (str) – Optional description.

Returns:

Created KB record with id, name, and description.

Return type:

dict

async delete_kb(kb_id)¶

Permanently delete a knowledge base and all its chunks.

Requires an admin key (admin_key constructor arg or XTRACE_ADMIN_KEY env var).

Parameters:: kb_id (str) – ID of the knowledge base to delete.
Return type:: None

async create_api_key(name, description='')¶

Create a new API key for the organisation.

Requires an admin key (admin_key constructor arg or XTRACE_ADMIN_KEY env var).

Parameters:

name (str) – Human-readable name for the API key.
description (str) – Optional description.

Returns:

Dict with apiKey (metadata) and secretKey (the actual key — store it, it won’t be shown again).

Return type:

dict

async revoke_api_key(key_hash)¶

Revoke an API key by its hash.

Requires an admin key (admin_key constructor arg or XTRACE_ADMIN_KEY env var).

Parameters:: key_hash (str) – Hash of the API key to revoke (from create_api_key()).
Return type:: None