# Python SDK (/python-sdk) ## Overview The Python SDK (`tetherto-qvac-sdk` on PyPI) is the Python client for QVAC. It is asyncio-native, drives the worker over `bare-rpc`, and exposes every QVAC capability through a typed API. ## Requirements * Python `>= 3.10` * Node.js `>= v22.17` (to host the worker) For host-level requirements (OS versions, GPU drivers, Vulkan runtime), see [System requirements](/system-requirements). ## Install Install the package: ```bash pip install "tetherto-qvac-sdk" ``` Optional extras: `vla`, `notebook`. Install the worker: ```bash python -m tetherto.qvac_sdk install-worker ``` `Client()` finds it automatically after this. To override the location, set `QVAC_WORKER_PATH` or pass `worker_path=` to `Client()`. ## Quickstart Run your first example using the Python SDK. Create the examples workspace: ```bash mkdir qvac-examples-py cd qvac-examples-py python -m venv .venv source .venv/bin/activate ``` Install the package: ```bash pip install tetherto-qvac-sdk ``` Install the worker: ```bash python -m tetherto.qvac_sdk install-worker ``` Create the quickstart script: ```python file=/packages/sdk-python/examples/quickstart.py title="quickstart.py" lineNumbers """Python port of packages/sdk/examples/quickstart.ts. Load a model (with download progress), run a streaming completion, unload. The one structural difference from the JS SDK: this client is asyncio-native, so top-level calls live inside `async def main()`. A `Client()` from a bundled wheel needs no configuration; against a thin install, point QVAC_SDK_DIR at a built `@qvac/sdk`. RUN: python examples/quickstart.py """ from __future__ import annotations import asyncio import sys from _common import print_progress from tetherto.qvac_sdk import Client, completion, load_model, unload_model from tetherto.qvac_sdk.models import LLAMA_3_2_1B_INST_Q4_0 async def main() -> int: async with Client() as client: t = client.transport try: model_id = await load_model( t, model_src=LLAMA_3_2_1B_INST_Q4_0, on_progress=print_progress ) run = completion( t, model_id=model_id, history=[ { "role": "user", "content": "Explain quantum computing in one sentence", } ], ) async for event in run.events: if event.type == "contentDelta": sys.stdout.write(event.text) sys.stdout.flush() print() await unload_model(t, model_id) except Exception as error: print(f"✖ {error}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(asyncio.run(main())) ``` Run the quickstart script: ```bash python quickstart.py ``` ## Running examples Follow these instructions to run any example in this documentation: * All examples are self-contained, runnable Python scripts. Use the `qvac-examples-py` workspace created in this quickstart to store and run them as you explore this documentation. * Run each example with Python `>= 3.10`, inside the virtualenv where you installed `tetherto-qvac-sdk`. * More examples can be found in the [Python SDK examples directory](https://github.com/tetherto/qvac/tree/main/packages/sdk-python/examples). * Some examples need companion files — sample audio, an image, or a config — that aren't part of the embedded code. These can be found in the same examples directory. * The `notebook.ipynb` example runs in Jupyter — see [Notebook / data science](#notebook--data-science-jupyter) below. One such example queries the model registry: ```python file=/packages/sdk-python/examples/registry_query.py title="registry_query.py" lineNumbers """Python port of packages/sdk/examples/registry-query.ts. Query the QVAC model registry: list, filter, search by engine / quantization / addon, and fetch one model's full record. These are request-reply methods on the flat `tetherto.qvac_sdk` surface; no model needs to be loaded, but a running worker (the transport) is required. RUN: python examples/registry_query.py """ from __future__ import annotations import asyncio import sys from tetherto.qvac_sdk import ( Client, model_registry_get_model, model_registry_list, model_registry_search, ) def val(x): """Registry enum fields (addon/engine) are pydantic enums; render the wire string, matching the JS SDK's plain-string fields.""" return x.value if hasattr(x, "value") else x def format_size(n) -> str: if n < 1024: return f"{n}B" if n < 1024**2: return f"{n / 1024:.1f}KB" if n < 1024**3: return f"{n / 1024**2:.1f}MB" return f"{n / 1024**3:.1f}GB" async def main() -> int: async with Client() as client: t = client.transport try: print("▸ QVAC Model Registry Query Examples\n") print("▸ Listing all models in QVAC model registry...") all_models = await model_registry_list(t) print(f"▸ Found {len(all_models)} models in registry\n") print("▸ Sample models:") for m in all_models[:5]: print( f" - {m.name} ({val(m.addon)}, {val(m.engine)}, {format_size(m.expected_size)})" ) print() print('▸ Searching for "whisper" models...') whisper = await model_registry_search(t, filter="whisper") print(f"▸ Found {len(whisper)} whisper-related models\n") print("▸ Searching by engine (llamacpp-embedding)...") embed_models = await model_registry_search(t, engine="llamacpp-embedding") print(f"▸ Found {len(embed_models)} embedding models") for m in embed_models[:3]: print(f" - {m.name} ({m.quantization})") print() print("▸ Searching for Q4 quantized models...") q4 = await model_registry_search(t, quantization="q4") print(f"▸ Found {len(q4)} Q4 quantized models") for m in q4[:3]: print(f" - {m.name}") print() if all_models: sample = all_models[0] print( f"▸ Getting specific model: {sample.registry_source}/{sample.registry_path}" ) model = await model_registry_get_model( t, sample.registry_path, sample.registry_source ) print(" Model details:") print(f" - Name: {model.name}") print(f" - Addon: {val(model.addon)}") print(f" - Engine: {val(model.engine)}") print(f" - Quantization: {model.quantization}") print(f" - Expected size: {format_size(model.expected_size)}") print() print("▸ QVAC model registry query examples completed successfully!") except Exception as error: print(f"✖ {error}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(asyncio.run(main())) ``` ## Configuration Pass the config to `Client()`: ```python async with Client(config={"cacheDirectory": "/data/qvac-models"}) as client: ... ``` `QVAC_CACHE_DIR` is a shortcut that sets `cacheDirectory` without code. See [Configuration](/configuration) for options and schema. ## Notebook / data science (Jupyter) For notebooks and REPLs, `tetherto.qvac_sdk.notebook.SyncClient` is a synchronous facade over `Client`: every call is a plain blocking call (no `await`), returns numpy/pandas-native data, and streams completions live into the cell. Install the `notebook` extra: ```bash pip install "tetherto-qvac-sdk[notebook]" ``` ```python file=/packages/sdk-python/examples/notebook.py title="notebook.py" lineNumbers """Notebook facade — the synchronous, data-science-native client. `tetherto.qvac_sdk.notebook.SyncClient` wraps the async client on a background event-loop thread so every call is a plain blocking call (no `await`), with numpy/pandas-native returns and live in-cell streaming. It's the ergonomic way to drive QVAC from a Jupyter notebook or a REPL. Needs the `notebook` extra (numpy + pandas): pip install "tetherto-qvac-sdk[notebook]" RUN: python examples/notebook.py """ from __future__ import annotations import sys from tetherto.qvac_sdk.models import EMBEDDINGGEMMA_300M_Q4_0, QWEN3_600M_INST_Q4 from tetherto.qvac_sdk.notebook import SyncClient def main() -> int: # SyncClient owns a Client (pass any Client kwargs, e.g. sdk_dir=...), or # wrap an already-connected transport with SyncClient(transport=...). # No async/await anywhere below -- a background thread runs the event loop. with SyncClient() as client: print("▸ Embeddings as numpy arrays") embed_model = client.load_model(model_src=EMBEDDINGGEMMA_300M_Q4_0) vector = client.embed(embed_model, "hello from the notebook facade") print(f" one text -> ndarray shape {vector.shape}, dtype {vector.dtype}") matrix = client.embed(embed_model, ["cats and dogs", "kittens and puppies"]) print(f" a batch -> ndarray shape {matrix.shape}") print("\n▸ Batch embeddings as a pandas DataFrame (indexed by text)") frame = client.embed_frame( embed_model, ["quantum computing", "espresso machine", "qubit entanglement"] ) print(f" DataFrame {frame.shape}, index={list(frame.index)}") client.unload_model(embed_model) print("\n▸ Completion, streaming live into the cell/stdout") llm = client.load_model( model_src=QWEN3_600M_INST_Q4, model_config={"n_ctx": 2048} ) text = client.completion( llm, "Explain what an embedding is in one sentence.", predict=256, temp=0, seed=42, ) print(f"\n returned {len(text)} chars") client.unload_model(llm) return 0 if __name__ == "__main__": sys.exit(main()) ``` Jupyter version: [`packages/sdk-python/examples/notebook.ipynb`](https://github.com/tetherto/qvac/tree/main/packages/sdk-python/examples/notebook.ipynb). ## API reference Everything you normally need is imported flat from `tetherto.qvac_sdk`; model constants live in `tetherto.qvac_sdk.models`. | Group | Names | | ------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Client | `Client` (`client.transport` is passed to every call) | | Ergonomic wrappers | `load_model`, `unload_model`, `completion`, `translate`, `cancel`, `delete_cache`, `invoke_plugin`, `invoke_plugin_stream`, `model_registry_list` / `_search` / `_get_model` | | Result types | `CompletionRun` (`.events`, `.final`), `CompletionFinal`, `ToolCall`, `TranslateRun` | | Generated methods | `embed`, `transcribe`, `text_to_speech`, `ocr_stream`, `diffusion_stream`, `classify`, `get_model_info`, `download_asset`, … (each takes a typed request model) | | Models & enums | `LoadModelRequest`, `ModelType`, … (also in `tetherto.qvac_sdk.schemas`) | | Errors | `QvacError`, `RPCError`, `InferenceCancelledError`, `ContextOverflowError`, … | | Logging / VLA | `logging_stream`, `subscribe_server_logs`, `SDK_LOG_ID`; `vla`, `vla_hparams`, `vla_preprocess_image`, `vla_pad_state` | | Notebook | `tetherto.qvac_sdk.notebook.SyncClient` — synchronous, numpy/pandas returns, live in-cell streaming | See [API reference](/reference/api) for the JS/TS surface; the Python API is generated from the same contract.