Python API Client

koios-client is a typed Python SDK for the Koios API. It gives you three layers, and you can mix them freely in one script:

a resource API of Device, Tag, and Model objects with methods like .enable(), .update(), and .delete()
a fully typed GraphQL client at client.gql, with editor autocomplete for every query and mutation
file helpers for import and export, model files, backups, trend data, and log downloads

Installation

pip install koios-client

Choosing a version

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 API schema, so a newer client against an older server offers operations that server does not have.

Two changes in 1.2.0 can break working code

Events are read-and-manage only: creating or editing an event through the API is no longer possible, and the header that grouped new events under a root event is gone. Acknowledging, deleting, clearing, and marking events seen all still work. Separately, sort options that always raised an error were removed — status on models and model bindings, and filename on model files.

Credentials

You authenticate with an API client ID and secret, created under System. See API Clients for how to create one and grant it permissions. The secret is shown once, when the client is created.

Getting started

from koios_client import KoiosClient

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

device = client.device(1)
device.enable()
for tag in device.tags():
    print(tag.name, tag.value)

client.close()

Or as a context manager, which closes the connection for you:

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

Working with resources

Every resource is fetched by ID or by slug. Exactly one is required.

device = client.device(1)
device = client.device(slug="550e8400-e29b-41d4-a716-446655440000")

To find something by name, filter the list:

from koios_client.types import DeviceFilter, OffsetPaginationInput, StrFilterLookup

devices = client.devices(filters=DeviceFilter(name=StrFilterLookup(exact="My Device")))

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

A returned list supports iteration, indexing, len(), and truth testing. len() counts the current page; total_count is everything matching on the server.

Creating, updating, deleting

from koios_client.types import DeviceInput, OneToManyInput

device = client.create_device(DeviceInput(
    name="New Device",
    protocol=OneToManyInput(set="1"),
    scan_rate=5.0,
))

device = device.update(name="Renamed Device", scan_rate=5.0, enabled=True)
device = device.enable()
device = device.disable()

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

update returns a fresh object with every field refreshed, so use the return value rather than the object you called it on.

Following relationships

device = client.device(1)
tags = device.tags(pagination=OffsetPaginationInput(limit=100))

model = client.model(5)
for binding in model.bindings():
    print(binding.name, binding.usage, binding.normalization_type)

Acting on many at once

client.enable_devices(["1", "2", "3"])
client.disable_tags(["10", "11", "12"])
client.delete_models(["20", "21"])

Live values

Current values come from the live data cache rather than the configuration database, so they are cheap to poll:

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

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

What you can reach

Devices, tags, models, scan groups, device sets, protocols, and API clients each expose the same shape: a singular getter, a plural list with filters and pagination, a create, bulk enable and disable, and a bulk delete. Protocols are read-only apart from visibility, and are not paginated.

Device sets group devices for failover. Members are added through client.gql, and the active member is chosen by the member record's ID, not the device's:

from koios_client.types import DeviceSetInput, DeviceSetItemInput, OneToManyInput

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

item = 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
)

ds = ds.set_active_device(item.id)

See Device Sets.

The GraphQL layer

Everything the API offers is available on client.gql, typed end to end.

from koios_client.types import OffsetPaginationInput, TagFilter

devices = client.gql.get_devices(pagination=OffsetPaginationInput(limit=50))
for device in devices.devices.results:
    print(device.name, device.status, device.protocol.name)

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

device = client.gql.get_device(slug="my-device")

A mutation returns either the entity or a structured error. check_operation unwraps it, raising if the server rejected the request:

from koios_client.types import DeviceInput, OneToManyInput

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

Every API enum is importable, so you never compare against a bare string:

from koios_client.types import (
    AggregateFunction,
    ProtocolReferenceCodeChoices,
    StatusChoices,
    UsageChoices,
)

Files, imports, and exports

Import and export

csv_bytes = client.export_devices_csv()
csv_bytes = client.export_tags_csv(ids=[1, 2, 3])
zip_bytes = client.export_models()

preview = client.import_tags_preview("tags.csv")
if preview["can_import"]:
    result = client.import_tags_confirm(
        preview["tmp_storage_name"],
        preview["file_name"],
    )

Imports preview by default, so client.import_devices("devices.csv") shows you what would change and applies nothing until you pass dry_run=False. See Importing & Exporting Devices.

Model files

result = client.upload_model_file(
    "model.onnx",
    model_slug="my-model",
    version="1.0",
    set_active=True,
)
model_bytes = client.download_model_file("model-file-slug")
structure = client.get_model_file_structure("model-file-slug")

See Managing Model Files.

Backups

task = client.create_backup(tier="full")

status = client.get_backup_status(task["task_id"])
backups = client.list_backups()
data = client.download_backup(backups["backups"][0]["filename"])

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

Backups and restores run in the background, so poll the task until it reports completion. See Backup & Restore.

Trend exports

task = client.create_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",
)

status = client.get_trend_export_status(task["task_id"])
if status["status"] == "completed":
    data = client.download_trend_export(status["filename"])

client.estimate_trend_export(...) reports the approximate size before you commit to a large range, along with the size limit. The server refuses an export estimated over that limit, or a raw export whose size it cannot estimate at the time, so client.create_trend_export(...) raises an error for that request instead of returning a task. See Settings & Export.

Other uploads and downloads

client.upload_component_library("my-component.kcl")
client.upload_opcua_certificate("cert.der", "key.pem", name="My Certificate")
client.upload_eds_file("device.eds")

log = client.download_device_log("my-device")
log = client.download_service_log("datacollector")

Connection options

client = KoiosClient(
    hostname="koios.example.com",
    client_id="your-uuid",
    client_secret="your-secret",
    port=443,
    ssl=True,
    verify_ssl=True,
    timeout=30.0,
)

Set verify_ssl=False only against a server using a self-signed certificate. Installing your own certificate is the better answer for anything long-lived — see Certificates.

Handling errors

from koios_client import (
    AuthenticationError,
    GraphQLError,
    KoiosConnectionError,
    KoiosError,
    KoiosPermissionError,
    NotFoundError,
    OperationError,
    ValidationError,
)

try:
    device = client.device(1)
    device.update(name="New Name")
except NotFoundError:
    print("No device with that identifier")
except OperationError as e:
    for msg in e.messages:
        print(msg.kind, msg.field, msg.message)
except AuthenticationError:
    print("Check the client ID and secret")

KoiosError is the base class, so catch it when you only need to know that something failed. OperationError carries the server's own validation messages, including which field each one is about.

Python Packages Component Builder SDK