tooluniverse.embedding_database module

Embedding Database Tool for ToolUniverse

A unified tool for managing embedding databases with FAISS vector search and SQLite metadata storage. Supports creating databases from documents, adding documents, searching, and loading existing databases. Uses OpenAI’s embedding models for text-to-vector conversion, with support for Azure OpenAI.

class tooluniverse.embedding_database.Path(*args, **kwargs)

Bases: PurePath

PurePath subclass that can make system calls.

Path represents a filesystem path but unlike PurePath, also offers methods to do system calls on path objects. Depending on your system, instantiating a Path will return either a PosixPath or a WindowsPath object. You can also instantiate a PosixPath or WindowsPath directly, but cannot instantiate a WindowsPath on a POSIX system or vice versa.

classmethod cwd()

Return a new path pointing to the current working directory (as returned by os.getcwd()).

classmethod home()

Return a new path pointing to the user’s home directory (as returned by os.path.expanduser(‘~’)).

samefile(other_path)

Return whether other_path is the same or not as this file (as returned by os.path.samefile()).

iterdir()

Iterate over the files in this directory. Does not yield any result for the special paths ‘.’ and ‘..’.

glob(pattern)

Iterate over this subtree and yield all existing files (of any kind, including directories) matching the given relative pattern.

rglob(pattern)

Recursively yield all existing files (of any kind, including directories) matching the given relative pattern, anywhere in this subtree.

absolute()

Return an absolute version of this path. This function works even if the path doesn’t point to anything.

No normalization is done, i.e. all ‘.’ and ‘..’ will be kept along. Use resolve() to get the canonical path to a file.

resolve(strict=False)

Make the path absolute, resolving all symlinks on the way and also normalizing it (for example turning slashes into backslashes under Windows).

stat(*, follow_symlinks=True)

Return the result of the stat() system call on this path, like os.stat() does.

owner()

Return the login name of the file owner.

group()

Return the group name of the file gid.

open(mode='r', buffering=-1, encoding=None, errors=None, newline=None)

Open the file pointed by this path and return a file object, as the built-in open() function does.

read_bytes()

Open the file in bytes mode, read it, and close the file.

read_text(encoding=None, errors=None)

Open the file in text mode, read it, and close the file.

write_bytes(data)

Open the file in bytes mode, write to it, and close the file.

write_text(data, encoding=None, errors=None, newline=None)

Open the file in text mode, write to it, and close the file.

readlink()

Return the path to which the symbolic link points.

touch(mode=438, exist_ok=True)

Create this file with the given access mode, if it doesn’t exist.

mkdir(mode=511, parents=False, exist_ok=False)

Create a new directory at this given path.

chmod(mode, *, follow_symlinks=True)

Change the permissions of the path, like os.chmod().

lchmod(mode)

Like chmod(), except if the path points to a symlink, the symlink’s permissions are changed, rather than its target’s.

unlink(missing_ok=False)

Remove this file or link. If the path is a directory, use rmdir() instead.

rmdir()

Remove this directory. The directory must be empty.

lstat()

Like stat(), except if the path points to a symlink, the symlink’s status information is returned, rather than its target’s.

rename(target)

Rename this path to the target path.

The target path may be absolute or relative. Relative paths are interpreted relative to the current working directory, not the directory of the Path object.

Returns the new Path instance pointing to the target path.

replace(target)

Rename this path to the target path, overwriting if that path exists.

The target path may be absolute or relative. Relative paths are interpreted relative to the current working directory, not the directory of the Path object.

Returns the new Path instance pointing to the target path.

symlink_to(target, target_is_directory=False)

Make this path a symlink pointing to the target path. Note the order of arguments (link, target) is the reverse of os.symlink.

hardlink_to(target)

Make this path a hard link pointing to the same file as target.

Note the order of arguments (self, target) is the reverse of os.link’s.

link_to(target)

Make the target path a hard link pointing to this path.

Note this function does not make this path a hard link to target, despite the implication of the function and argument names. The order of arguments (target, link) is the reverse of Path.symlink_to, but matches that of os.link.

Deprecated since Python 3.10 and scheduled for removal in Python 3.12. Use hardlink_to() instead.

exists()

Whether this path exists.

is_dir()

Whether this path is a directory.

is_file()

Whether this path is a regular file (also True for symlinks pointing to regular files).

is_mount()

Check if this path is a POSIX mount point

is_symlink()

Whether this path is a symbolic link.

is_block_device()

Whether this path is a block device.

is_char_device()

Whether this path is a character device.

is_fifo()

Whether this path is a FIFO.

is_socket()

Whether this path is a socket.

expanduser()

Return a new path with expanded ~ and ~user constructs (as returned by os.path.expanduser)

class tooluniverse.embedding_database.OpenAI(*, api_key: str | None | Callable[[], str] = None, organization: str | None = None, project: str | None = None, webhook_secret: str | None = None, base_url: str | URL | None = None, websocket_base_url: str | URL | None = None, timeout: float | Timeout | None | NotGiven = NOT_GIVEN, max_retries: int = 2, default_headers: Mapping[str, str] | None = None, default_query: Mapping[str, object] | None = None, http_client: Client | None = None, _strict_response_validation: bool = False)

Bases: SyncAPIClient

__init__(*, api_key: str | None | Callable[[], str] = None, organization: str | None = None, project: str | None = None, webhook_secret: str | None = None, base_url: str | URL | None = None, websocket_base_url: str | URL | None = None, timeout: float | Timeout | None | NotGiven = NOT_GIVEN, max_retries: int = 2, default_headers: Mapping[str, str] | None = None, default_query: Mapping[str, object] | None = None, http_client: Client | None = None, _strict_response_validation: bool = False) → None

Construct a new synchronous OpenAI client instance.

This automatically infers the following arguments from their corresponding environment variables if they are not provided: - api_key from OPENAI_API_KEY - organization from OPENAI_ORG_ID - project from OPENAI_PROJECT_ID - webhook_secret from OPENAI_WEBHOOK_SECRET

property audio: Audio

property auth_headers: dict[str, str]

property batches: Batches

property beta: Beta

property chat: Chat

property completions: Completions

property containers: Containers

property conversations: Conversations

copy(*, api_key: str | None | Callable[[], str] = None, organization: str | None = None, project: str | None = None, webhook_secret: str | None = None, websocket_base_url: str | URL | None = None, base_url: str | URL | None = None, timeout: float | Timeout | None | NotGiven = NOT_GIVEN, http_client: Client | None = None, max_retries: int | NotGiven = NOT_GIVEN, default_headers: Mapping[str, str] | None = None, set_default_headers: Mapping[str, str] | None = None, default_query: Mapping[str, object] | None = None, set_default_query: Mapping[str, object] | None = None, _extra_kwargs: Mapping[str, Any] = {}) → Self

Create a new client instance re-using the same options given to the current client with optional overriding.

property default_headers: dict[str, str | Omit]

property embeddings: Embeddings

property evals: Evals

property files: Files

property fine_tuning: FineTuning

property images: Images

property models: Models

property moderations: Moderations

property qs: Querystring

property realtime: Realtime

property responses: Responses

property uploads: Uploads

property vector_stores: VectorStores

property webhooks: Webhooks

with_options(*, api_key: str | None | Callable[[], str] = None, organization: str | None = None, project: str | None = None, webhook_secret: str | None = None, websocket_base_url: str | URL | None = None, base_url: str | URL | None = None, timeout: float | Timeout | None | NotGiven = NOT_GIVEN, http_client: Client | None = None, max_retries: int | NotGiven = NOT_GIVEN, default_headers: Mapping[str, str] | None = None, set_default_headers: Mapping[str, str] | None = None, default_query: Mapping[str, object] | None = None, set_default_query: Mapping[str, object] | None = None, _extra_kwargs: Mapping[str, Any] = {}) → Self

Create a new client instance re-using the same options given to the current client with optional overriding.

property with_raw_response: OpenAIWithRawResponse

property with_streaming_response: OpenAIWithStreamedResponse

api_key: str

organization: str | None

project: str | None

webhook_secret: str | None

websocket_base_url: str | httpx.URL | None

class tooluniverse.embedding_database.AzureOpenAI(*, azure_endpoint: str, azure_deployment: str | None = None, api_version: str | None = None, api_key: str | Callable[[], str] | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: Callable[[], str] | None = None, organization: str | None = None, webhook_secret: str | None = None, websocket_base_url: str | URL | None = None, timeout: float | Timeout | None | NotGiven = NOT_GIVEN, max_retries: int = DEFAULT_MAX_RETRIES, default_headers: Mapping[str, str] | None = None, default_query: Mapping[str, object] | None = None, http_client: Client | None = None, _strict_response_validation: bool = False)

class tooluniverse.embedding_database.AzureOpenAI(*, azure_deployment: str | None = None, api_version: str | None = None, api_key: str | Callable[[], str] | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: Callable[[], str] | None = None, organization: str | None = None, webhook_secret: str | None = None, websocket_base_url: str | URL | None = None, timeout: float | Timeout | None | NotGiven = NOT_GIVEN, max_retries: int = DEFAULT_MAX_RETRIES, default_headers: Mapping[str, str] | None = None, default_query: Mapping[str, object] | None = None, http_client: Client | None = None, _strict_response_validation: bool = False)

class tooluniverse.embedding_database.AzureOpenAI(*, base_url: str, api_version: str | None = None, api_key: str | Callable[[], str] | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: Callable[[], str] | None = None, organization: str | None = None, webhook_secret: str | None = None, websocket_base_url: str | URL | None = None, timeout: float | Timeout | None | NotGiven = NOT_GIVEN, max_retries: int = DEFAULT_MAX_RETRIES, default_headers: Mapping[str, str] | None = None, default_query: Mapping[str, object] | None = None, http_client: Client | None = None, _strict_response_validation: bool = False)

Bases: BaseAzureClient[Client, Stream[Any]], OpenAI

__init__(*, azure_endpoint: str, azure_deployment: str | None = None, api_version: str | None = None, api_key: str | Callable[[], str] | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: Callable[[], str] | None = None, organization: str | None = None, webhook_secret: str | None = None, websocket_base_url: str | URL | None = None, timeout: float | Timeout | None | NotGiven = NOT_GIVEN, max_retries: int = DEFAULT_MAX_RETRIES, default_headers: Mapping[str, str] | None = None, default_query: Mapping[str, object] | None = None, http_client: Client | None = None, _strict_response_validation: bool = False) → None

__init__(*, azure_deployment: str | None = None, api_version: str | None = None, api_key: str | Callable[[], str] | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: Callable[[], str] | None = None, organization: str | None = None, webhook_secret: str | None = None, websocket_base_url: str | URL | None = None, timeout: float | Timeout | None | NotGiven = NOT_GIVEN, max_retries: int = DEFAULT_MAX_RETRIES, default_headers: Mapping[str, str] | None = None, default_query: Mapping[str, object] | None = None, http_client: Client | None = None, _strict_response_validation: bool = False) → None

__init__(*, base_url: str, api_version: str | None = None, api_key: str | Callable[[], str] | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: Callable[[], str] | None = None, organization: str | None = None, webhook_secret: str | None = None, websocket_base_url: str | URL | None = None, timeout: float | Timeout | None | NotGiven = NOT_GIVEN, max_retries: int = DEFAULT_MAX_RETRIES, default_headers: Mapping[str, str] | None = None, default_query: Mapping[str, object] | None = None, http_client: Client | None = None, _strict_response_validation: bool = False) → None

Construct a new synchronous azure openai client instance.

This automatically infers the following arguments from their corresponding environment variables if they are not provided: - api_key from AZURE_OPENAI_API_KEY - organization from OPENAI_ORG_ID - project from OPENAI_PROJECT_ID - azure_ad_token from AZURE_OPENAI_AD_TOKEN - api_version from OPENAI_API_VERSION - azure_endpoint from AZURE_OPENAI_ENDPOINT

Parameters:

• azure_endpoint – Your Azure endpoint, including the resource, e.g. https://example-resource.azure.openai.com/

• azure_ad_token – Your Azure Active Directory token, https://www.microsoft.com/en-us/security/business/identity-access/microsoft-entra-id

• azure_ad_token_provider – A function that returns an Azure Active Directory token, will be invoked on every request.

• azure_deployment – A model deployment, if given with azure_endpoint, sets the base client URL to include /deployments/{azure_deployment}. Not supported with Assistants APIs.

copy(*, api_key: str | None | Callable[[], str] = None, organization: str | None = None, project: str | None = None, webhook_secret: str | None = None, websocket_base_url: str | URL | None = None, api_version: str | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: Callable[[], str] | None = None, base_url: str | URL | None = None, timeout: float | Timeout | None | NotGiven = NOT_GIVEN, http_client: Client | None = None, max_retries: int | NotGiven = NOT_GIVEN, default_headers: Mapping[str, str] | None = None, set_default_headers: Mapping[str, str] | None = None, default_query: Mapping[str, object] | None = None, set_default_query: Mapping[str, object] | None = None, _extra_kwargs: Mapping[str, Any] = {}) → Self

Create a new client instance re-using the same options given to the current client with optional overriding.

with_options(*, api_key: str | None | Callable[[], str] = None, organization: str | None = None, project: str | None = None, webhook_secret: str | None = None, websocket_base_url: str | URL | None = None, api_version: str | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: Callable[[], str] | None = None, base_url: str | URL | None = None, timeout: float | Timeout | None | NotGiven = NOT_GIVEN, http_client: Client | None = None, max_retries: int | NotGiven = NOT_GIVEN, default_headers: Mapping[str, str] | None = None, set_default_headers: Mapping[str, str] | None = None, default_query: Mapping[str, object] | None = None, set_default_query: Mapping[str, object] | None = None, _extra_kwargs: Mapping[str, Any] = {}) → Self

Create a new client instance re-using the same options given to the current client with optional overriding.

class tooluniverse.embedding_database.BaseTool(tool_config)

Bases: object

__init__(tool_config)

classmethod get_default_config_file()

Get the path to the default configuration file for this tool type.

This method uses a robust path resolution strategy that works across different installation scenarios:

1. Installed packages: Uses importlib.resources for proper package resource access

2. Development mode: Falls back to file-based path resolution

3. Legacy Python: Handles importlib.resources and importlib_resources

Override this method in subclasses to specify a custom defaults file.

Returns:

Path or resource object pointing to the defaults file

classmethod load_defaults_from_file()

Load defaults from the configuration file

run(arguments=None)

Execute the tool.

The default BaseTool implementation accepts an optional arguments mapping to align with most concrete tool implementations which expect a dictionary of inputs.

check_function_call(function_call_json)

get_required_parameters()

Retrieve required parameters from the endpoint definition. Returns: list: List of required parameters for the given endpoint.

tooluniverse.embedding_database.register_tool(tool_type_name=None, config=None)

Decorator to automatically register tool classes and their configs.

Usage:

@register_tool(‘CustomToolName’, config={…}) class MyTool:

pass

tooluniverse.embedding_database.get_logger(name: str | None = None) → Logger

Get a logger instance

Parameters:

name (str, optional) – Logger name (usually __name__)

Returns:

Logger instance

Return type:

logging.Logger

class tooluniverse.embedding_database.EmbeddingDatabase(tool_config)

Bases: BaseTool

Unified embedding database tool supporting multiple operations: - create_from_docs: Create new database from documents - add_docs: Add documents to existing database - search: Search for similar documents - load_database: Load existing database from path

__init__(tool_config)

run(arguments)

Main entry point for the tool