Get the handle for a model, with support for custom overrides. Args: model_name (str): The name of the model.

Close all resources and release connections.

Generate a test ID matching Mirix ID pattern (prefix-[8 hex chars]).

Start the background worker as an asyncio.Task.

Check if a configuration class is properly defined. - It should inherit from Pydantic's BaseModel. - It should have a model_config attrib

Create a new organization.

Create a new user in the system.

List records with before/after pagination, ordering by created_at. Can use both before and after to fetch a window of records.

Safely get config value from either dict or object.

Create a new agent using a config

Execute the tool and return the result.

Get client_id and org_id from headers or use defaults. Returns: tuple[str, str]: (client_id, org_id)

Initialize a meta agent with the given configuration. This creates a meta memory agent that manages multiple specialized memory agen

Build raw SQL WHERE clause fragments from filter_tags and scopes. Args: filter_tags: Dict of filter conditions. Keys "read_scopes" a

Convert a ChatCompletion message object into a Message object (synced to DB)

Escape single quotes in SQL string.

Create a Adder instance from a configuration factory.

Helper to get sub agent by type.

* Sanitize tool output to remove technical details

Search for memories across ALL users in the organization with optional temporal filtering. Results are automatically filtered by clie

Get connection from pool with health check.

Get a database session.

Return connection to pool.

Clear all messages from the queue.

Count number of vectors in the index. Returns: int: Total number of vectors.

Delete a memory by ID. Args: memory_id (str): ID of the memory to delete.

Format parameter value to handle both quoted and unquoted formats

Fetch an organization by ID (with cache - Redis or IPS Cache).

Get all cubes accessible by a user. Args: user_id (str): The user ID. Returns: list[Cube]: List of cubes acc

* Sanitize and validate string input

Create an edge from source node to target node. Args: source_id: ID of the source node. target_id: ID of the

Create a new block or update an existing one (with Redis Hash caching). Scope is always auto-injected from actor.write_scope unless

Get value (string/JSON) from Redis.

Get a value from the context.

Correct way to index into the memory.memory field, returns a Block

Retrieve a block by its ID (with cache - Redis or IPS Cache).

Get current UTC datetime with compatibility for different Python versions

Get the current UTC time

Save configuration to JSON file. Args: output_path: Path to save JSON configuration indent: JSON ind

Update an existing raw memory record. Args: memory_id: ID of the memory to update new_context: New context t

Add memories using the product/add API endpoint. Args: messages: List of message objects with role and content

Count items in the database, optionally with filter.

Create a user if it doesn't exist, or get existing user. This method ensures a user exists in the backend database before performing

Authenticate using either JWT token (dashboard) or Client API Key (programmatic). Args: authorization: Bearer JWT token from Authori

JSON serializer that handles datetime objects. Args: data: Data to serialize indent: JSON indentation level Returns:

List records with cursor-based pagination, ordering by created_at. Cursor is an ID, but pagination is based on the cursor object's cr

Create QueueMessage protobuf and send to queue. Args: actor: The Client performing the action (for auth/write operations)

Set updated_at to naive UTC (RawMemory uses TIMESTAMP WITHOUT TIME ZONE). Overrides mixin to avoid asyncpg "offset-naive and offset-a

Download the image and save it to the corresponding path. Args: task (dict): The task dict got from ``task_queue``. t

List all available agents to a user