Koios Client

Typed Python SDK for the Koios IoT platform. Provides two API layers:

• High-level resource API — Device, Tag, Model objects with intuitive methods like .enable(), .update(), .delete()
• Low-level GraphQL client — fully-typed generated client via client.gql with IDE autocomplete

Plus REST helpers for file operations, import/export, backups, and trend data.

Installation

pip install koios-client

Or install from source:

pip install -e ".[dev]"

Compatibility

The client's minor tracks the server's: install the line that matches the Koios you run.

Client	Koios server
1.2.x	1.2.x
1.1.x	1.1.x
1.0.x	1.0.x

pip install "koios-client~=1.2.0"

The client is generated from the server's GraphQL schema, so a newer client against an older server will offer operations that server does not have.

Two changes in 1.2.0 follow removals on the 1.2 server, and are the only ones that can break working code:

• Events are read-and-manage only. Creating or editing an event through the API is no longer possible, and the request header that grouped new events under a root event is gone. Acknowledging, deleting, clearing, and marking events seen all still work. Integrations that wrote events into Koios need another route.
• Sorts that never worked are gone: status on models and on model bindings, and filename on model files. Passing them previously raised an error, so dropping them is the fix rather than a regression. Model binding filters no longer require status, which makes that filter usable on its own.

Quick Start

from koios_client import KoiosClient

client = KoiosClient(
    hostname="koios.example.com",
    client_id="your-client-uuid",
    client_secret="your-client-secret",
)

# High-level resource API — fetch by ID (default) or slug
device = client.device(1)
device.enable()
for tag in device.tags():
    print(tag.name, tag.value)

# Low-level GraphQL (fully typed — IDE autocomplete works)
result = client.gql.get_devices()
for device in result.devices.results:
    print(device.name, device.status)

client.close()

Use as a context manager for automatic cleanup:

with KoiosClient("koios.example.com", "client-id", "secret") as client:
    device = client.device(1)
    print(device.name, device.enabled)

High-Level Resource API

The resource layer wraps the generated GraphQL client with intuitive Python objects. All resource classes are importable from the top-level package:

from koios_client import Device, Tag, Model, PaginatedList

Fetching Resources

Every resource can be fetched by ID (default positional arg) or slug (UUID).

# By ID (default — integer database primary key)
device = client.device(1)
tag = client.tag(42)
model = client.model(5)

# By slug (UUID — useful when you already have the identifier)
device = client.device(slug="550e8400-e29b-41d4-a716-446655440000")

# By name — use the list query with filters
from koios_client.types import DeviceFilter, StrFilterLookup
devices = client.devices(filters=DeviceFilter(name=StrFilterLookup(exact="My Device")))
device = devices[0]

# List with filtering, ordering, and pagination
from koios_client.types import OffsetPaginationInput

devices = client.devices(
    filters=DeviceFilter(enabled=True),
    pagination=OffsetPaginationInput(limit=50),
)
print(f"{devices.total_count} total devices")
for device in devices:
    print(device.name, device.slug, device.enabled)

PaginatedList supports iteration, indexing, len(), and bool():

tags = client.tags()
print(len(tags))       # Number of items on this page
print(tags.total_count) # Total matching items on server
first = tags[0]         # Index access

CRUD Operations

from koios_client.types import DeviceInput, OneToManyInput

# Create — `protocol` takes a OneToManyInput pointing at the protocol's ID
device = client.create_device(DeviceInput(
    name="New Device",
    protocol=OneToManyInput(set="1"),
    scan_rate=5.0,
))

# Update (returns a fresh object with all fields refreshed)
device = device.update(name="Renamed Device", scan_rate=5.0, enabled=True)

# Enable / Disable (convenience wrappers around update)
device = device.enable()
device = device.disable()

# Duplicate
copy = device.duplicate("Device Copy", description="Cloned from original")

# Delete
device.delete()

Parent-Child Accessors

# Device → Tags
device = client.device(1)
tags = device.tags(
    filters=TagFilter(enabled=True),
    pagination=OffsetPaginationInput(limit=100),
)

# Model → Bindings
model = client.model(5)
for binding in model.bindings():
    print(binding.name, binding.usage, binding.normalization_type)
    binding.update(normalization_type="MIN_MAX")

Bulk Operations

# Bulk enable / disable
client.enable_devices(["1", "2", "3"])
client.disable_tags(["10", "11", "12"])

# Bulk delete
client.delete_devices(["1", "2", "3"])
client.delete_tags(["10", "11", "12"])
client.delete_models(["20", "21"])
client.delete_scan_groups(["30", "31"])
client.delete_device_sets(["40", "41"])

Live Data

Fetch real-time values from the Redis cache:

device = client.device(1)
live = device.live()
print(live.get("status"), live.get("error_message"))

tag = client.tag(42)
live = tag.live()
print(live.get("value"), live.get("timestamp"), live.get("quality"))

Import / Export with Resources

# Import with dry_run (default) — returns preview without applying
preview = client.import_devices("devices.csv")
print(preview)

# Import for real
result = client.import_devices("devices.csv", dry_run=False)

# Export a single device/model
csv_bytes = device.export_csv()
zip_bytes = model.export()

All Resource Types

All singular methods accept id (positional, integer) or slug= (UUID). Exactly one must be provided.

Method	Returns	Notes
client.device(id, slug=)	Device
client.devices(...)	PaginatedList[Device]	filters, order, pagination
client.create_device(data)	Device
client.enable_devices(ids)	list[Device]
client.disable_devices(ids)	list[Device]
client.delete_devices(ids)	None
client.import_devices(file)	dict	dry_run=True by default
client.tag(id, slug=)	Tag
client.tags(...)	PaginatedList[Tag]
client.create_tag(data)	Tag
client.enable_tags(ids)	list[Tag]
client.disable_tags(ids)	list[Tag]
client.delete_tags(ids)	None
client.import_tags(file)	dict	dry_run=True by default
client.model(id, slug=)	Model
client.models(...)	PaginatedList[Model]
client.create_model(data)	Model
client.enable_models(ids)	list[Model]
client.disable_models(ids)	list[Model]
client.delete_models(ids)	None
client.import_models(file)	dict	dry_run=True by default
client.scan_group(id, slug=)	ScanGroup
client.scan_groups(...)	PaginatedList[ScanGroup]
client.create_scan_group(data)	ScanGroup
client.enable_scan_groups(ids)	list[ScanGroup]
client.disable_scan_groups(ids)	list[ScanGroup]
client.delete_scan_groups(ids)	None
client.device_set(id, slug=)	DeviceSet	supports .update(), .delete(), .set_active_device(item_id)
client.device_sets(...)	PaginatedList[DeviceSet]
client.create_device_set(data)	DeviceSet
client.delete_device_sets(ids)	None
client.protocol(id, slug=)	Protocol	read-only, supports .set_visible(bool)
client.protocols()	list[Protocol]	not paginated

Device Sets (Redundancy Groups)

Device sets group multiple devices for failover. Member devices are added as DeviceSetItem records via the low-level GraphQL client, and the active member is controlled by set_active_device:

from koios_client.types import (
    DeviceSetInput,
    DeviceSetItemInput,
    OneToManyInput,
)

ds = client.create_device_set(DeviceSetInput(
    name="Redundant Pair",
    protocol=OneToManyInput(set="1"),
))

# Add two member devices with priority 1 (primary) and 2 (backup).
item1 = KoiosClient.check_operation(
    client.gql.create_device_set_item(
        data=DeviceSetItemInput(
            device_set=OneToManyInput(set=ds.id),
            device=OneToManyInput(set="10"),
            priority=1,
        )
    ).create_device_set_item
)
item2 = KoiosClient.check_operation(
    client.gql.create_device_set_item(
        data=DeviceSetItemInput(
            device_set=OneToManyInput(set=ds.id),
            device=OneToManyInput(set="11"),
            priority=2,
        )
    ).create_device_set_item
)

# Switch which member is active — identified by the DeviceSetItem ID, not the Device ID.
ds = ds.set_active_device(item2.id)

Endpoint Clients

# List endpoint clients
clients = client.endpoint_clients()

# Get a single endpoint client by ID
ec = client.endpoint_client("123")

# Create a new endpoint client (returns one-time secret)
result = client.create_endpoint_client("My Integration", description="Data pipeline")

# Set permissions
client.set_endpoint_client_permissions(
    client_id="123",
    permission_ids=["1", "2", "3"],
)

Low-Level GraphQL Client

All GraphQL queries and mutations are available via client.gql with fully typed parameters and return values.

Querying Data

from koios_client.types import (
    DeviceFilter,
    OffsetPaginationInput,
    StatusChoices,
)

# Paginated queries
devices = client.gql.get_devices(
    pagination=OffsetPaginationInput(limit=50),
)
print(f"Total: {devices.devices.total_count}")
for device in devices.devices.results:
    print(device.name, device.status, device.protocol.name)

# Filtered queries
tags = client.gql.get_tags(
    filters=TagFilter(enabled=True),
    pagination=OffsetPaginationInput(limit=100),
)

# Single item by slug
device = client.gql.get_device(slug="my-device")
tag = client.gql.get_tag(slug="my-tag")

Mutations

Single-item mutations return a union of the entity type or OperationInfo. Use check_operation to unwrap:

from koios_client.types import DeviceInput, OneToManyInput

raw = client.gql.create_device(data=DeviceInput(
    name="New Device",
    protocol=OneToManyInput(set="1"),
))

# Raises OperationError if the server returned validation/permission errors
device = KoiosClient.check_operation(raw.create_device)
print(f"Created: {device.name} ({device.slug})")

Available Enums

All GraphQL enums are available as typed Python enums:

from koios_client.types import (
    StatusChoices,        # RUNNING, STOPPED, FAILED
    UsageChoices,         # INPUT, OUTPUT, VIRTUAL
    AggregateFunction,    # MEAN, SUM, MIN, MAX, FIRST, LAST, COUNT
    DeviceErrorCodeChoices,
    TagErrorCodeChoices,
    ProtocolReferenceCodeChoices,  # OPCUA, MODBUS_TCP, ETHERNET_IP, ...
)

REST API

Operations that involve file uploads, background tasks, or binary responses use REST endpoints exposed directly on the client.

Export / Import

# Export devices as CSV
csv_bytes = client.export_devices_csv()
with open("devices.csv", "wb") as f:
    f.write(csv_bytes)

# Export specific tags
csv_bytes = client.export_tags_csv(ids=[1, 2, 3])

# Export models as ZIP (includes bindings)
zip_bytes = client.export_models()

# Two-step import: preview then confirm
preview = client.import_tags_preview("tags.csv")
print(f"Will import {preview['result']['total_rows']} rows")
if preview["can_import"]:
    result = client.import_tags_confirm(
        preview["tmp_storage_name"],
        preview["file_name"],
    )
    print(f"Imported {result['imported_count']} tags")

Model Files

# Upload a model file
result = client.upload_model_file(
    "model.onnx",
    model_slug="my-model",
    version="1.0",
    set_active=True,
)

# Download a model file
model_bytes = client.download_model_file("model-file-slug")

# Get model structure (weights stripped for ONNX)
structure = client.get_model_file_structure("model-file-slug")

Backups

# Create a backup
task = client.create_backup(tier="full")

# Poll for completion
import time
while True:
    status = client.get_backup_status(task["task_id"])
    if status["status"] == "completed":
        break
    time.sleep(2)

# List and download
backups = client.list_backups()
for backup in backups["backups"]:
    print(backup["filename"], backup["size_bytes"])

data = client.download_backup(backups["backups"][0]["filename"])

# Restore from backup
upload = client.upload_restore_file("backup.tar.gz")
task = client.start_restore(upload["restore_file_path"])

Trend Data Export

# The export streams: write it as it arrives, however large it is
with open("trend.csv", "wb") as fh:
    for chunk in client.stream_trend_export(
        tag_ids=[1, 2, 3],
        start="2026-01-01T00:00:00Z",
        stop="2026-02-01T00:00:00Z",
        mode="resampled",
        resample_interval="5m",
        aggregate_fn="mean",
        output_format="csv",
    ):
        fh.write(chunk)

# Or the whole history, as a ZIP of one file per day
with open("trend.zip", "wb") as fh:
    for chunk in client.stream_trend_export([1, 2, 3], export_all=True, split_by_day=True):
        fh.write(chunk)

# Estimate first, if you want to know what you are asking for
estimate = client.estimate_trend_export(tag_ids=[1, 2, 3], export_all=True)
print(f"~{estimate['estimated_size_bytes'] / 1e6:.1f} MB")

Exporting requires the Can export trend data permission.

Nothing is built on the server and nothing is stored: the export is produced as it is sent, so there is no file to collect afterwards and none left behind. That also means there is nothing to resume — if the connection drops part-way you have an incomplete file and should run it again.

Everything that can refuse an export is answered before the first byte, as an HTTP error: exports estimated over 10 GB, a malformed resample_interval, and too many exports already running at once.

output_format="parquet" writes the timestamp index as a real timestamp column, so it loads back as a datetime rather than needing to be parsed from text.

Component Libraries

result = client.upload_component_library("my-component.kcl")
print(f"Uploaded: {result['name']} v{result['version']}")

OPC-UA Certificates

# Upload a certificate + key pair
result = client.upload_opcua_certificate(
    "cert.der", "key.pem", name="My Certificate"
)

# Download cert or key
cert_bytes = client.download_opcua_certificate("cert-slug")
key_bytes = client.download_opcua_key("cert-slug")

EDS Files

# Upload an EDS file for a device
result = client.upload_eds_file("device.eds")

# Delete the EDS file from a device
client.delete_eds_file(device_id=1)

Log Downloads

log = client.download_device_log("my-device")
log = client.download_model_log("my-model")
log = client.download_scan_group_log("scan-group-1")
log = client.download_component_log("instance-1")
log = client.download_service_log("datacollector")

Connection Options

client = KoiosClient(
    hostname="koios.example.com",
    client_id="your-uuid",
    client_secret="your-secret",
    port=443,           # Default: 443 (HTTPS) or 80 (HTTP)
    ssl=True,           # Default: True
    verify_ssl=True,    # Default: True (set False for self-signed certs)
    timeout=30.0,       # Default: 30 seconds
)

Error Handling

from koios_client import (
    KoiosError,           # Base exception
    AuthenticationError,  # Bad credentials or expired token
    NotFoundError,        # Resource not found (e.g. client.device(999))
    OperationError,       # Mutation returned OperationInfo
    GraphQLError,         # GraphQL response errors
    KoiosConnectionError, # Network connectivity issues
    KoiosPermissionError, # Insufficient permissions
    ValidationError,      # Input validation failures
)

try:
    device = client.device(1)
    device.update(name="New Name")
except NotFoundError:
    print("Device not found")
except OperationError as e:
    for msg in e.messages:
        print(f"{msg.kind}: {msg.field}: {msg.message}")
except AuthenticationError:
    print("Check your client credentials")

Development

# Install with dev dependencies
make install

# Format and lint
make format      # ruff check --fix + format
make lint        # ruff check + format --check

# Type check and run unit tests
make typecheck   # basedpyright
make test        # pytest (mocked HTTP; no live server needed)
make ci          # lint + typecheck + test

# Regenerate GraphQL client (requires a running Koios dev server at :8080)
make codegen

Integration tests

make test-integration runs the suite under tests/integration/ against a live Koios server. These tests are skipped by the default make test / make ci runs — they exist so releases can be verified against a real container.

Copy tests/integration/.env.example to tests/integration/.env and fill in an endpoint client ID/secret from the target server:

cp tests/integration/.env.example tests/integration/.env
# Edit tests/integration/.env with KOIOS_HOST, KOIOS_CLIENT_ID, KOIOS_CLIENT_SECRET
make test-integration

Releasing

# Bump pyproject.toml version, land via PR, then tag:
make tag         # reads version from pyproject.toml, creates matching git tag

License

Licensed under the Apache License, Version 2.0 — see LICENSE.

Copyright 2024 Ai-OPs, Inc.