omnictl

Async-first Python SDK and CLI for interacting with the Sidero Omni API.

PyPI package: omnictl
Repository: github.com/SparkAIUR/omni-sdk

Highlights

• Async-first transport (httpx) with retries and conservative caching.
• Service-account request signing compatible with Omni API (siderov1).
• Unified sync/async DX:
  • sync methods: get_cluster(...)
  • async methods: aget_cluster(...)
• Config + persistent preferences (OMNI_SDK_CONFIG, YAML/JSON/TOML, SQLite state).
• CLI entrypoint: omnipy.
• Thin talosctl wrapper: omnipy talos (alias omnipy t).
• Compatibility import alias: omnisdk.

Install

Install SDK in a project

uv pip install omnictl

Install global CLI with uv

Install from PyPI:

uv tool install omnictl

Install directly from GitHub:

uv tool install git+https://github.com/SparkAIUR/omni-sdk.git

Upgrade global install:

uv tool upgrade omnictl

Verify CLI:

omnipy --help

Configuration

Config loading precedence:

1. Runtime dict/model passed to client constructor.
2. Explicit config_path=....
3. OMNI_SDK_CONFIG environment variable.
4. Default path search:
   • Linux/macOS: ~/.config/omnisdk/config.yml|yaml|toml|json
   • Windows: %LOCALAPPDATA%/omnisdk/config.yml|yaml|toml|json

Example ~/.config/omnisdk/config.yml:

version: "1"
default_instance: prod
instances:
  prod:
    url: https://my-omni-instance.example.com
    verify_tls: true
    auth:
      service_account_key: null
preferences:
  default_cluster: my-cluster

Common CLI Tasks

These workflows mirror common omnictl tasks from Sidero docs, but use omnipy.

Inspect resources and CRDs

omnipy get Clusters.omni.sidero.dev
omnipy get Clusters.omni.sidero.dev my-cluster -o yaml
omnipy get Clusters.omni.sidero.dev my-cluster --explain -o yaml
omnipy crd list --query cluster
omnipy crd explain clusterstatus

Apply and remove resources

omnipy apply -f cluster-resources.yaml --verbose
omnipy delete Clusters.omni.sidero.dev my-cluster --teardown

Cluster and machine operations

omnipy cluster status my-cluster
omnipy cluster lock my-cluster
omnipy cluster unlock my-cluster
omnipy cluster machine lock <machine-id>
omnipy cluster machine unlock <machine-id>
omnipy cluster machine delete <machine-id>

Kubernetes and Talos access

omnipy kubeconfig --cluster my-cluster > my-cluster.kubeconfig
omnipy talosconfig --cluster my-cluster > my-cluster.talosconfig
omnipy talos --cluster my-cluster version --client
# first run with --nodes persists defaults for later talos wrapper calls
omnipy talos --cluster my-cluster --nodes 10.0.0.11 get members

Support and audit flows

omnipy support --cluster my-cluster --output support.zip
omnipy audit-log 2026-01-01 2026-01-07 > audit.jsonl
omnipy machine-logs <machine-id> --tail-lines 200

Access and automation resources

omnipy serviceaccount list
omnipy serviceaccount create ci-bot
omnipy jointoken create bootstrap --ttl 3600
omnipy jointoken list
omnipy user list
omnipy user create dev@example.com --role Operator
omnipy infraprovider list

Common SDK Tasks

Unified client (sync + async methods on one client)

import asyncio
from omni import OmniClient

async def main() -> None:
    async with OmniClient() as client:
        status_sync = client.get_cluster_status("my-cluster")
        status_async = await client.aget_cluster_status("my-cluster")
        print(status_sync["body"])
        print(status_async["body"])

asyncio.run(main())

List resources and inspect by ID

from omni import OmniClient

client = OmniClient()
clusters = client.resources.list({"namespace": "default", "type": "Clusters.omni.sidero.dev"})
cluster = client.resources.get(
    {"namespace": "default", "type": "Clusters.omni.sidero.dev", "id": "my-cluster"}
)
print(clusters)
print(cluster)
client.close()

Download and persist kubeconfig/talosconfig

import base64
from pathlib import Path
from omni import OmniClient

client = OmniClient()
kcfg = client.management.kubeconfig({"service_account": False, "break_glass": False}, metadata={"cluster": "my-cluster"})
Path("my-cluster.kubeconfig").write_bytes(base64.b64decode(kcfg["kubeconfig"]))

tcfg_path = client.ensure_talosconfig(cluster="my-cluster")
print(f"managed talosconfig at {tcfg_path}")
client.close()

Stream audit events (async)

import asyncio
import base64
from omni import AsyncOmniClient

async def main() -> None:
    client = AsyncOmniClient()
    async for event in client.management.read_audit_log({"start_time": "2026-01-01", "end_time": "2026-01-02"}):
        chunk = event.get("audit_log")
        if isinstance(chunk, str):
            print(base64.b64decode(chunk).decode("utf-8"), end="")
    await client.aclose()

asyncio.run(main())

Compatibility Alias

import omnisdk

client = omnisdk.OmniClient()
print(client.get_cluster("my-cluster"))
client.close()

More Examples

See runnable examples in:

• examples/README.md
• examples/sync_resource_queries.py
• examples/async_resource_queries.py
• examples/config_sources.py
• examples/global_singleton.py
• examples/export_access_configs.py
• examples/audit_log_stream.py
• examples/compat_alias.py

Development

uv sync --extra dev
uv run python scripts/update_upstream.py
uv run ruff check .
uv run mypy src
uv run pytest

Versioning

omnictl tracks upstream Omni versioning (X.Y.Z) where possible.
SDK-only patches use .postN.

License

MPL-2.0