INTELLABEL Python SDK

Python client for the INTELLABEL platform API, built on httpx with sensible timeouts, automatic retries and typed errors. Ships a sync client (IntelLabel) and an async one (AsyncIntelLabel) with identical methods.

Install

pip install -r requirements.txt      # or: pip install -e .

Requires Python 3.9+.

Authentication

The SDK authenticates with an API key. Create one in the INTELLABEL dashboard (Settings → API Keys, Super Admin / Admin only). Keys look like ilk_<prefix>_<secret> and are shown only once at creation time. On the Explore plan a company can hold a single API key: delete it in the dashboard to create another. The other plans have no limit.

There are three ways to give the key to the SDK. Nothing else is needed: the server address is part of the deployment, not something you set.

Way	How	Best for
Log in once	intellabel login in a terminal, or intellabel.login() in Python. Checks the key, lets you pick a default workspace, saves both in ~/.intellabel/credentials.json. Change the workspace later with intellabel switch-workspace.	Laptops, notebooks
Environment variables / .env	API_KEY (or INTELLABEL_API_KEY) and, optionally, WORKSPACE_ID (or INTELLABEL_WORKSPACE_ID), in the shell environment or a .env file in the project folder.	CI, containers, servers
Constructor arguments	IntelLabel(api_key="ilk_...", workspace_id="...").	Keys loaded from your own secret store

When several are present the order is: constructor argument, then environment / .env, then the saved login (details in Where the key comes from).

API keys cannot be created, listed, deactivated or deleted through the SDK. The client refuses any /api-keys path with PermissionDeniedError before sending, and the backend independently rejects API keys on those routes (403). Manage keys from the dashboard.

Pick the scopes the key needs when you create it:

Scope	Allows
read	GET requests
write	POST / PUT / PATCH / DELETE requests
admin	Admin-only routes (also covers read + write)

A call outside the key's scopes raises PermissionDeniedError.

Log in from the terminal

Run intellabel login once (or python -m intellabel login). It asks for the key with hidden input, checks it against the backend, prints the result, lets you pick a default workspace, and saves both. After that, IntelLabel() works with no .env and no workspace_id.

$ intellabel login
INTELLABEL API key:
Checking the key against https://... ...

Workspaces available to this key:
  1. Production (c2b3cd06-...)
  2. Staging (3e6697d8-...)
Default workspace [1-2, Enter to skip]: 1

Success: the SDK is authenticated.
  Account:   Jane Doe (Admin), Acme
  API key:   CI key (ilk_97230dc9_********)
  Scopes:    read, write
  Expires:   05-11-2026 01:21 PM
  Workspace: Production (c2b3cd06-...)
  Saved to:  /home/jane/.intellabel/credentials.json

A rejected key prints the backend's reason and the prompt asks again, up to 3 times:

Error: Authentication failed: This API key has expired. Try again.

The backend reasons are Invalid API key., This API key is inactive., This API key has expired. and The account that created this API key no longer has access..

Command	What it does
intellabel login	Ask for the key (hidden), check it, choose a workspace, save. Options: --workspace-id ID (no workspace question), --no-save (only check), --api-key KEY (no prompt; the key ends up in your shell history)
intellabel whoami	Check the key the SDK would use right now and show the account, key name, scopes, expiry and where the key came from
intellabel switch-workspace [ID or NAME]	Change the saved default workspace. Checks the saved key, matches the id or name against the workspaces it may use, saves it. Without an argument it shows a numbered list. Only while logged in: after intellabel logout it fails with Not logged in
intellabel logout	Delete the saved key. A key in the environment or .env is not touched

Exit status is 0 on success, 1 when authentication fails and 2 for a usage error, so the commands can be used in scripts. Without a terminal (CI), the key is read from standard input: echo "$INTELLABEL_KEY" | intellabel login --workspace-id <uuid>.

The same from Python, for notebooks:

import intellabel

info = intellabel.login()            # hidden prompt, prints success or the error, saves
info.name, info.role, info.scopes    # who the key belongs to
intellabel.logout()

login() raises intellabel.LoginError after printing why it failed. With an existing client, client.whoami() returns the same AuthInfo (account, api_key, plus the properties name, role, company_name, scopes, masked_key, expires_at). It raises AuthenticationError with the backend's reason when the key is not accepted. It calls GET /api/auth/whoami, which any valid key can reach whatever its scopes.

workspaces, workspace_id and workspace_name on AuthInfo are filled by login() and switch_workspace().

Switch workspace

After logging in, the saved default workspace can be changed without logging in again:

import intellabel

info = intellabel.switch_workspace("Staging")      # by name, or by workspace id
info.workspace_id, info.workspace_name
intellabel.switch_workspace()                       # numbered list in a terminal; Enter keeps the current one

Or on a client that runs with the logged-in key:

with IntelLabel() as client:
    client.switch_workspace("3e6697d8-...")         # updates client.workspace_id and the saved default

Rules:

• Only a saved login can switch. After intellabel logout / intellabel.logout() there is no saved login and switch_workspace raises LoginError (Not logged in). A key from API_KEY, .env or api_key= is not a login, unless it is the very key intellabel login saved.
• The workspace must be one the key may use: for a Super Admin key the active workspaces of the company, for an Admin key the account's own workspace. Anything else raises LoginError (module function) or ValueError (client method), listing what is available. Two workspaces with the same name: use the id.
• A revoked or expired key raises LoginError (The saved key was rejected: ...) from the module function and AuthenticationError from the client method. The saved workspace is left as it was.

The key is saved in ~/.intellabel/credentials.json. Set INTELLABEL_CONFIG_DIR to use another folder. The file holds the key in plain text and only the current user can read it, as with ~/.netrc or the Hugging Face token file. The secret is never printed, only the masked key.

Where the key comes from

The clients take the first key they find:

1. the api_key= argument
2. API_KEY (or INTELLABEL_API_KEY) in the environment or .env
3. the key saved by intellabel login

The default workspace follows the same order (workspace_id=, then WORKSPACE_ID, then the saved workspace). The saved workspace is used whenever the active key is the one you logged in with, even if that same key also sits in .env. A different key never inherits it. client.api_key_source says which source the key came from. When no key is found, the ValueError points to intellabel login.

Quick start

from intellabel import IntelLabel

with IntelLabel() as client:                      # key from `intellabel login` or API_KEY in .env
    print(client.health())                        # {'app_status': 'WORKING', ...}
    page = client.projects.list(workspace_id="<workspace uuid>")
    print(page.total_count, [p.name for p in page])

Async, same API with await:

import asyncio
from intellabel import AsyncIntelLabel

async def main():
    async with AsyncIntelLabel() as client:
        print(await client.health())
        pages = await asyncio.gather(*(client.projects.list(ws) for ws in workspace_ids))

asyncio.run(main())

Pick the sync client for scripts, notebooks and CLIs. Pick the async client inside services that already run on asyncio or when fanning out many calls. Outside a with block call client.close() (sync) or await client.aclose() (async) when done.

Generic access for routes that don't have a resource module yet:

res = client.get("/workspaces")          # .get/.post/.put/.patch/.delete
print(res.output)                        # the `output` field of the response envelope

Configuration

.env file

A .env file is an alternative to intellabel login. Copy .env.example to .env in your project folder and fill it in:

API_KEY=ilk_<prefix>_<secret>
WORKSPACE_ID=workspace_uuid

Variable	Fallback name	Purpose
API_KEY	INTELLABEL_API_KEY	The ilk_... key from the dashboard (Settings → API Keys)
WORKSPACE_ID	INTELLABEL_WORKSPACE_ID	Optional. Default workspace UUID for resource calls; any call can still pass workspace_id=

The client loads the .env automatically: the nearest one from the working directory upwards, never one next to the installed SDK. A variable is used only when the matching constructor argument is omitted. Precedence is: explicit argument, then shell environment, then .env, then the key saved by intellabel login. Keep .env out of version control: the repo's .gitignore excludes .env and .env.* (except .env.example). Use dotenv_path="path/to/.env" to load a specific file or dotenv_path=False to skip loading.

You never set the server address. The package carries the production URL; operators can point a deployment elsewhere with the BASE environment variable, and base_url= exists only to reach a different INTELLABEL server from one script.

Constructor options

import httpx
from intellabel import IntelLabel

client = IntelLabel(
    api_key="ilk_...",                                  # or API_KEY in .env
    timeout=httpx.Timeout(30.0, connect=10.0),         # default; a float sets every phase
    max_retries=2,                                     # default
    default_headers={"X-Trace": "my-job"},
    verify=True,                                       # TLS verification, or a CA bundle path
)

Same options for AsyncIntelLabel. Every call also accepts a per-request timeout=. Pass http_client=httpx.Client(...) (or httpx.AsyncClient for the async client) to supply your own transport, proxies or mock; the SDK won't close a client it didn't create.

Retries

Condition	Retried?
Connection refused / connect timeout	Yes, any method (request never sent)
Read timeout	Only GET / HEAD / OPTIONS / PUT / DELETE
HTTP 429	Yes, honours Retry-After
HTTP 408 / 5xx	Only GET / HEAD / OPTIONS / PUT / DELETE

POST requests are never replayed once the server may have received them. Backoff is exponential with jitter, capped at 8 s.

Conventions

Every resource method exists on both clients with the same name and arguments. On IntelLabel it is called directly; on AsyncIntelLabel it is awaited, and generators become async for.

Conventions used throughout:

• workspace_id is optional everywhere. When omitted, the client default applies (IntelLabel(workspace_id=...) or WORKSPACE_ID in .env). With neither, a ValueError is raised before any request is sent.
• Argument mistakes the SDK can detect itself (empty ids, unknown statuses, bad dates, missing files) raise ValueError locally, with no request sent.
• Server rejections raise the APIError subclass for the HTTP status (see Errors). The common ones are listed per method.
• Pagination is 0-based. The backend clamps a page past the end to the last page instead of returning an empty page, so iterate stops on total_count, not on an empty result.
• Dates passed as filters accept datetime.date, datetime.datetime or a "YYYY-MM-DD" string and match one calendar day. Dates returned by the backend are strings in the company timezone, formatted "DD-MM-YYYY HH:MM AM/PM".
• Every model keeps unknown fields the backend may add later in .extra and exposes .to_dict().

---

Client — IntelLabel / AsyncIntelLabel

IntelLabel(api_key=None, *, base_url=None, timeout=..., max_retries=2, api_prefix="/api", default_headers=None, verify=True, http_client=None, dotenv_path=None, workspace_id=None)

Parameter	Type	Default	Description
api_key	str	API_KEY from env / .env	ilk_<prefix>_<secret>. Required.
base_url	str	packaged production URL	Server origin. Operators override it per deployment with the BASE environment variable; pass it only to reach a different INTELLABEL server.
timeout	float | httpx.Timeout	30 s total, 10 s connect	A float applies to every phase.
max_retries	int	2	Retry budget for transient failures, see Retries.
api_prefix	str	"/api"	Prefix every route is mounted under.
default_headers	Mapping[str, str]	None	Extra headers on every request.
verify	bool | str	True	TLS verification, or a CA bundle path.
http_client	httpx.Client / httpx.AsyncClient	None	Bring your own transport, proxies or mock. The SDK never closes a client it did not create.
dotenv_path	str | PathLike | False	nearest .env	A specific .env to load, or False to skip loading.
workspace_id	str	WORKSPACE_ID from env / .env	Default workspace for resource calls.

Raises ValueError when the key is missing or malformed, and TypeError when http_client is the wrong httpx class for the client.

Lifecycle

Method	Description
with IntelLabel() as client: / async with AsyncIntelLabel() as client:	Context managers that close the connection pool on exit.
close() / await aclose()	Explicit close when not using a context block. No-op for an injected http_client.

Properties and key rotation

Member	Returns	Description
base_url	str	Normalised origin without trailing slash.
workspace_id	str | None	Current default workspace; assignable at runtime.
api_key_prefix	str	The public <prefix> segment of the key. Safe to log.
masked_api_key	str	ilk_<prefix>_********. Used by repr().
set_api_key(api_key)	None	Swap the key, e.g. after rotation. ValueError on a malformed key.
switch_workspace(workspace)	str	Change the default workspace of a logged-in SDK, see Switch workspace. Updates workspace_id and the saved default; returns the new id. LoginError when this key is not the saved login.

Raw requests

request(method, path, *, params=None, json=None, data=None, files=None, headers=None, timeout=...) -> APIResponse
get(path, **kwargs) / post(path, **kwargs) / put(path, **kwargs) / patch(path, **kwargs) / delete(path, **kwargs)

Sends to <base_url><api_prefix><path>, adds the Authorization: Bearer header, drops None query parameters, applies the retry policy and unwraps the response envelope. Use these for routes without a resource module yet.

Returns APIResponse with .success, .status, .output and .extra. Raises an APIError subclass on error statuses, APIConnectionError / APITimeoutError on transport failures once retries are exhausted, and PermissionDeniedError locally for any /api-keys path (key management is dashboard-only).

health() -> dict

GET /api/health. Returns the backend's status dict (app_status, db_status, websocket_status, ping_status, response_time). Unauthenticated, so it proves connectivity, not that the key is valid.

download(path, dest, *, params=None, on_bytes=None, fallback_name="download", timeout=...) -> Path

Streams GET <path> to disk without loading it into memory. dest with a file extension (out.zip) is the file to write; any other dest is a folder, created if missing (relative to the current directory), and the file inside it keeps the server's Content-Disposition name; an existing file is never replaced (name (1).zip, name (2).zip, ...). An explicit file path is overwritten. on_bytes(written, total) reports progress. Error statuses raise the usual typed errors and leave no partial file.

---

Projects — client.projects

Usage

Every project call is scoped to a workspace. Pass workspace_id= per call, or set a default once with IntelLabel(workspace_id=...) or WORKSPACE_ID in .env.

from intellabel import IntelLabel, ProjectStatus

with IntelLabel(workspace_id="<workspace uuid>") as client:
    page = client.projects.list(status=ProjectStatus.IN_PROGRESS, page=0, limit=20)   # pages are 0-based
    print(page.total_count, [p.name for p in page])

    for project in client.projects.iterate(page_size=50):          # walks every page
        ...

    project = client.projects.create("Street signs", ProjectStatus.NOT_STARTED)
    project = client.projects.get(project.project_id)                 # None if in trash
    project = client.projects.set_status(project.project_id, ProjectStatus.IN_PROGRESS)
    project = client.projects.rename(project.project_id, "Street signs v2")
    project = client.projects.update(project.project_id, name="...", status="...")

With AsyncIntelLabel every call is awaited and iterate is an async generator (async for project in client.projects.iterate()).

Statuses: Not Started, In Progress, Quality Check, Completed, Skipped, Returned (constants on ProjectStatus). The API has no permanent delete; move_to_trash is the deletion the platform offers, and the dashboard trash can restore.

Methods

Backend router app/api/project_apis.py. All calls are workspace-scoped.

list(workspace_id=None, *, name=None, project_id=None, status=None, created_by=None, creation_date=None, last_updation_date=None, page=None, limit=None) -> ProjectList

GET /projects. Lists the workspace's projects, newest first, excluding trashed ones.

Parameter	Description
name, project_id, created_by	Fuzzy, case-insensitive matches (partial ratio above 70).
status	Exact match; one of ProjectStatus.ALL.
creation_date, last_updation_date	Match a calendar day.
page, limit	0-based page and page size. Must be given together; without them every match is returned.

Returns ProjectList(projects: list[Project], total_count: int). total_count is the number of matches before pagination. The object is iterable and supports len().

Raises ValueError for page without limit, page < 0, limit < 1, an unknown status or a bad date. Server errors: BadRequestError (no workspace), NotFoundError (unknown or trashed workspace), PermissionDeniedError (deactivated workspace or not a member).

iterate(workspace_id=None, *, page_size=50, **filters) -> Iterator[Project]

Generator over every matching project, fetching page_size per request from page 0 and stopping when total_count is reached. Accepts the same filters as list. async for on the async client. ValueError when page_size < 1.

get(project_id, workspace_id=None) -> Project | None

GET /projects/{project_id}. Returns the project, or None when it exists but is in the trash (the backend answers with an empty list for that case). Raises NotFoundError when the id does not exist in the workspace, ValueError for an empty id.

create(name, status="Not Started", workspace_id=None) -> Project

POST /projects. Names are trimmed and must be unique inside the workspace.

Parameter	Description
name	Required, non-empty.
status	One of ProjectStatus.ALL. Default "Not Started".

Raises ValueError for an empty name or unknown status; ConflictError on a duplicate name; PermissionDeniedError when the key lacks the write scope or the account is below Workforce Manager.

update(project_id, *, name=None, status=None, workspace_id=None) -> Project

PUT /projects/{project_id}. Changes the name and/or the status. The backend requires both fields on every update, so when only one is passed the current project is fetched first and the other field is sent back unchanged.

Raises ValueError when neither field is given, the name is empty, the status is unknown, or the project is in the trash; ConflictError on a duplicate name; NotFoundError for an unknown id.

rename(project_id, name, workspace_id=None) -> Project

Shortcut for update(project_id, name=name).

set_status(project_id, status, workspace_id=None) -> Project

Shortcut for update(project_id, status=status).

move_to_trash(project_id, workspace_id=None) -> str

POST /move-to-trash with {"project_id": ...}. Soft-deletes the project and its tasks: they leave every listing, get returns None, and the dashboard trash can restore them. The platform has no permanent delete; a trash-deletion worker purges trashed items later. Returns the server message ("Move to trash operation successful."). BadRequestError when already in trash.

Models

Project fields: id, project_id (8-character public id used in URLs), project_name, status, workspace_id, created_by, created_user_id, role, creation_date, last_updation_date, s3_key, in_trash, restored. Property name aliases project_name.

ProjectStatus constants: NOT_STARTED, IN_PROGRESS, QUALITY_CHECK, COMPLETED, SKIPPED, RETURNED; tuple ALL; validate(value) returns the value or raises ValueError. Matching is case-sensitive, like the backend.

---

Tasks — client.tasks

Usage

Tasks live inside a project, and every call needs the project id plus the workspace (default or per call). Dataset routes live on client.datasets.

from intellabel import IntelLabel, TaskStatus, StorageType, CloudProvider, CloudKey

with IntelLabel(workspace_id="<workspace uuid>") as client:
    page = client.tasks.list("a1b2c3d4", status=TaskStatus.IN_PROGRESS, page=0, limit=20)
    task = client.tasks.get("a1b2c3d4", "t1t1t1t1")              # None if absent
    refs = client.tasks.names(["a1b2c3d4"])                       # [TaskRef(task_id, task_name), ...]
    task = client.tasks.set_status("a1b2c3d4", task.task_id, TaskStatus.COMPLETED)
    client.tasks.move_to_trash("a1b2c3d4", task.task_id)

Uploading files

Creating a task opens an upload batch; the files follow in a second step and a worker processes them asynchronously. One call does the whole thing for local files:

task, progress = client.tasks.create_and_upload(
    "a1b2c3d4", "Batch A",
    files=["img/001.jpg", "img/002.png", "clips/a.mp4"],
    no_of_files_per_batch=50,                 # or no_of_batches=...
    video_fps={"a.mp4": 5},                   # required for videos (1-30), or one int for all
    show_progress=True,                       # terminal bar while sending, status line while processing
    timeout=600,
)

show_progress=True draws a byte-accurate bar for the transfer, then a Processing on server... line until the worker finishes. It uses tqdm when installed (pip install intellabel[progress]) and a plain stderr bar otherwise. Everything is computed on the client; nothing extra is asked of the backend. For your own UI use the callbacks instead: on_bytes(sent, total) per streamed chunk, on_chunk(files_sent, files_total, bytes_sent) per accepted request, on_progress(state) per server-side state change.

Or point it at a folder instead of listing files (recursive by default; non-media files are skipped):

task, progress = client.tasks.create_and_upload("a1b2c3d4", "Batch B", directory="datasets/batch-b", no_of_batches=4, video_fps=5)

Or step by step:

task = client.tasks.create("a1b2c3d4", "Batch A", total_files=len(paths), total_size=total_bytes, no_of_batches=4)
client.tasks.upload_files("a1b2c3d4", task.task_id, paths, task=task, show_progress=True)   # streams from disk
final = client.tasks.wait_for_upload("a1b2c3d4", task.task_id, timeout=600, show_progress=True)

• Large uploads. Files are streamed in several multipart requests that share the batch, the same way the dashboard does it: 50 files or 512 MB per request by default (files_per_request, bytes_per_request). The async client sends up to concurrency=2 requests at once. Tens of thousands of files or many gigabytes are fine; nothing is read into memory up front.
• Formats. Images jpg, jpeg, png, webp; videos mp4, avi, mov, mkv, wmv, flv, webm; a zip archive (extracted server-side) must be the only file of the call.
• Progress. The backend only exposes two states over its REST API: queued (files accepted, worker processing) and completed (worker finished and wrote total_files / corrupted_files on the task). progress() returns that as an UploadProgress; iter_progress() yields each change (async for on the async client); wait_for_upload() blocks until completion and raises UploadTimeoutError with the last state after timeout. Per-file percentages are only pushed on the dashboard's websocket, which API keys cannot open.

Importing from cloud storage

task = client.tasks.create("a1b2c3d4", "S3 batch", total_files=2, total_size=1_048_576, no_of_batches=1)
client.tasks.upload_from_cloud(
    "a1b2c3d4", task.task_id,
    keys=[CloudKey("imgs/a.jpg", size=524_288), CloudKey("imgs/b.jpg", size=524_288), CloudKey("more/", type="DIR")],
    cloud_storage="prod-bucket",             # display name configured in the workspace
    provider=CloudProvider.AWS_S3,           # AWS S3 only for now; Azure and GCS coming soon
    task=task, fps=30,
)
client.tasks.wait_for_upload("a1b2c3d4", task.task_id)

For S3 the backend verifies the key sizes and that their sum equals the total_size given at create.

TaskStatus has the same values as ProjectStatus. Task names are unique inside a project.

Methods

Backend router app/api/task_apis.py, excluding dataset routes. Every call takes the project_id the task belongs to plus the workspace.

list(project_id, workspace_id=None, *, task_id=None, name=None, status=None, created_by=None, creation_date=None, last_updation_date=None, page=None, limit=None) -> TaskList

GET /projects/{project_id}/tasks. Lists the project's tasks, newest first, excluding trashed ones. Filters behave as for projects (task_id, name, created_by fuzzy; status exact; dates by day; 0-based page with limit). Each task carries at most 5 preview files in total across images and videos.

Returns TaskList(tasks: list[Task], total_count: int), iterable with len().

Raises ValueError for an empty project id or bad filter arguments; NotFoundError when the project is not in the workspace; PermissionDeniedError for a worker account not assigned to any task of the project.

iterate(project_id, workspace_id=None, *, page_size=50, **filters) -> Iterator[Task]

Generator over every matching task, paged from 0 and stopping on total_count. Same filters as list.

get(project_id, task_id, workspace_id=None) -> Task | None

The backend has no single-task route, so this calls list with the task_id filter (which is fuzzy server-side) and keeps only the exact id. Returns None when the project has no such task. ValueError for an empty id.

names(project_ids=None, workspace_id=None) -> list[TaskRef]

GET /tasks. Returns TaskRef(task_id, task_name) for every non-trashed task in the workspace, or only in the given projects. project_ids may be a list or a single string. NotFoundError names any project id that does not exist ("Project(s) not found: ...").

create(project_id, name, *, total_files, status="Not Started", total_size=0, no_of_batches=None, no_of_files_per_batch=None, target_storage="local", cloud_display_name=None, cloud_provider=None, workspace_id=None) -> Task

POST /projects/{project_id}/validate/task. Creates the task row and opens its upload batch. The returned Task carries batch_id and progress_id, which the upload methods need. Files are uploaded afterwards with upload_files or upload_from_cloud.

Parameter	Description
name	Required; unique inside the project.
total_files	Number of files that will be uploaded. Non-negative integer.
status	One of TaskStatus.ALL. Default "Not Started".
total_size	Total size in bytes. Checked against the plan's remaining storage. For cloud imports it must equal the sum of the key sizes.
no_of_batches / no_of_files_per_batch	Exactly one of them, a positive integer: how the worker splits the files into datasets.
target_storage	StorageType.LOCAL (default) or StorageType.CLOUD, where the files are stored.
cloud_display_name, cloud_provider	Required with target_storage="cloud": the display name of a storage configured in the workspace and one of CloudProvider.ALL. Not allowed with local.

Raises ValueError for any inconsistent combination above; ConflictError on a duplicate name; BadRequestError when the plan's storage limit is reached; NotFoundError when the project, the cloud storage or the subscription cannot be found; PermissionDeniedError without the write scope.

update(project_id, task_id, *, name=None, status=None, workspace_id=None) -> Task

PUT /projects/{project_id}/tasks/{task_id}. Same semantics as the project update: both fields are required server-side, so a missing one is fetched from the current task first. ValueError when neither is given, the name is empty, the status is unknown, or the task is not found; ConflictError on a duplicate name.

rename(project_id, task_id, name, workspace_id=None) -> Task

Shortcut for update(..., name=name).

set_status(project_id, task_id, status, workspace_id=None) -> Task

Shortcut for update(..., status=status).

upload_files(project_id, task_id, files=None, *, directory=None, recursive=True, batch_id=None, progress_id=None, task=None, video_fps=None, files_per_request=50, bytes_per_request=512 MiB, on_chunk=None, on_bytes=None, show_progress=False, workspace_id=None) -> UploadBatch

POST /projects/{project_id}/tasks/{task_id}/upload/files, multipart. Streams local files into the task's upload batch and returns once the backend has queued every request for its worker. Give exactly one of files or directory; passing both, or neither, raises ValueError.

Parameter	Description
files	Sequence of file paths, or tuples (filename, data) / (filename, data, mime) where data is bytes or a binary file object. Paths are opened lazily and streamed; caller-owned file objects are left open.
directory	A folder to upload instead of files. Every file with a supported extension is included, sorted by path; other files are skipped. ValueError when the folder is missing, contains no supported files, or two files share a name (the backend identifies files by name).
recursive	Include sub-folders of directory. Default True.
batch_id, progress_id	From the created task. Either pass both, or pass task= instead.
task	The Task returned by create; supplies batch_id and progress_id.
video_fps	Required when any video is included. Either one int for every video or a {filename: fps} mapping. Range 1 to 30. Sent to the backend as its config form field.
files_per_request, bytes_per_request	Chunking limits. Files are split into several requests that share the batch, the way the dashboard uploads. Defaults 50 files / 512 MiB; the server accepts up to 5 GB per request.
on_chunk	callback(files_sent, files_total, bytes_sent) after each accepted request.
on_bytes	callback(bytes_sent, bytes_total) as httpx streams the files; fires many times per file.
show_progress	Draw a terminal progress bar for the bytes being sent (tqdm if installed, else a plain stderr bar). Set INTELLABEL_NO_TQDM=1 to force the plain bar.

Accepted formats by extension: images .jpg .jpeg .png .webp; videos .mp4 .avi .mov .mkv .wmv .flv .webm; archives .zip. The MIME type is derived from the extension in the way the backend expects (for example .mkv is sent as video/mkv). A zip archive must be the only file of the call, because the backend re-derives the batch total from the archive's contents.

Returns UploadBatch(queued: bool, batch_id: str).

Raises ValueError for an unsupported extension, a missing path, missing video_fps or an fps out of range, a zip mixed with other files, or missing batch ids. BadRequestError when the batch has already completed or expired ("uploading has already completed for this batch_id ..."). Files the SDK opened are always closed, also on errors.

On AsyncIntelLabel there is one extra keyword, concurrency=2: the maximum number of chunk requests in flight at once. on_chunk may be a coroutine function.

upload_from_cloud(project_id, task_id, keys, *, cloud_storage, provider, batch_id=None, progress_id=None, task=None, fps=30, workspace_id=None) -> UploadBatch

POST /projects/{project_id}/tasks/{task_id}/upload/cloud/files. Imports objects from a cloud storage configured in the workspace into the task's batch. The backend worker reads the objects itself; nothing passes through the SDK.

Parameter	Description
keys	Non-empty sequence of CloudKey objects, plain key strings (treated as FILE with size 0), or dicts with key, name, type, mime_type, size as the dashboard sends them.
cloud_storage	The storage's display name in the workspace.
provider	CloudProvider.AWS_S3 ("AWS S3"). AWS S3 is the only supported cloud provider today; Azure Blob Storage and Google Cloud Storage are coming soon.
batch_id, progress_id, task	As for upload_files.
fps	Frame rate applied to every imported video, 1 to 30. Default 30.

CloudKey(key, size=0, mime_type=None, name=None, type="FILE"): type="DIR" imports every media object under the prefix; size in bytes is verified against the real object for S3, and the sum of sizes must equal the total_size given at create. mime_type defaults from the extension.

Raises ValueError for empty keys, an unknown provider, an fps out of range, an invalid key type or size, or an unsupported extension. NotFoundError when the cloud storage is not configured ("Cloud Provider does not exists with the cloud storage name ..."); BadRequestError when S3 sizes do not match or the batch has completed.

progress(project_id, task_id, workspace_id=None) -> UploadProgress

Reads the task row and derives the upload state. The backend exposes exactly two states over its REST API: queued (files accepted, worker processing) and completed (the worker finalised the batch and wrote total_files and corrupted_files on the task). Per-file percentages are pushed only on the dashboard websocket, which API keys cannot open, and GET /get/progress is never updated by the upload worker, so neither is used.

Returns UploadProgress with project_id, task_id, done, total_files, uploaded_files, corrupted_files (both known only once done), preview_files (0 to 5), last_updation_date, and properties status ("queued" / "completed"), processed and progress_percentage (0 or 100). Raises ValueError when the task is not found.

iter_progress(project_id, task_id, *, interval=2.0, timeout=None, workspace_id=None) -> Iterator[UploadProgress]

Polls progress every interval seconds and yields each state change until done. async for on the async client. Raises ValueError for interval <= 0 or timeout <= 0, and UploadTimeoutError after timeout seconds (None waits forever); the error's .progress holds the last state seen.

wait_for_upload(project_id, task_id, *, interval=2.0, timeout=None, on_progress=None, show_progress=False, workspace_id=None) -> UploadProgress

Blocks until the worker finishes and returns the final state. on_progress(state) is called on every change; on the async client it may be a coroutine function. show_progress=True prints a Processing on server... status line meanwhile. Same errors as iter_progress.

create_and_upload(project_id, name, files=None, *, directory=None, recursive=True, status="Not Started", no_of_batches=None, no_of_files_per_batch=None, video_fps=None, files_per_request=50, bytes_per_request=512 MiB, on_chunk=None, on_bytes=None, show_progress=False, wait=True, interval=2.0, timeout=None, on_progress=None, workspace_id=None) -> tuple[Task, UploadProgress | None]

Convenience for local files. Takes exactly one of files or directory (as in upload_files), computes total_files and total_size from the resolved files, calls create, streams the files with upload_files, and, unless wait=False, blocks in wait_for_upload. With show_progress=True the byte bar is followed by a Processing on server... status line that updates on every state change until completion. Validation of videos and chunking happens before the task is created, so a bad argument never leaves an empty task behind. Returns the task and the final progress (None when not waiting). concurrency is available on the async client. Errors as for the three underlying methods.

move_to_trash(project_id, task_id, workspace_id=None) -> str

POST /move-to-trash with {"project_id", "task_id"}. Soft-deletes the task; it leaves list, get returns None, and the dashboard trash can restore it. Returns the server message.

Models and constants

Task fields: id, task_id, task_name, project_id, company_id, status, created_by, created_user_id, creation_date, last_updation_date, total_files, corrupted_files (null until the upload worker finalises), batch_id, progress_id, s3_key, target_storage, in_trash, restored, images, videos (preview lists, at most 5 together), labels, workflow (raw backend structures). Property name aliases task_name.

TaskRef(task_id, task_name), TaskList(tasks, total_count), UploadBatch(queued, batch_id), UploadProgress (see progress), CloudKey (see upload_from_cloud).

TaskStatus is the same object as ProjectStatus. StorageType.LOCAL / StorageType.CLOUD. CloudProvider.AWS_S3 is the supported cloud provider; Azure Blob Storage and Google Cloud Storage are coming soon.

---

Datasets — client.datasets

Usage

The upload worker splits a task's files into datasets (no_of_batches / no_of_files_per_batch at task creation). Each dataset sits in one workflow stage, has a status and a priority, and is worked on by at most one assigned worker at a time. Listings are per stage and pages are 1-based.

from intellabel import IntelLabel, Selection, DatasetStatus, FileType, ReviewTab

with IntelLabel(workspace_id="<workspace uuid>") as client:
    page = client.datasets.list("a1b2c3d4", "t1t1t1t1", "Annotator 1", dataset_status=DatasetStatus.NOT_STARTED)
    for ds in client.datasets.iterate("a1b2c3d4", "t1t1t1t1", "Annotator 1"):
        print(ds.dataset_name, ds.status, ds.priority, ds.assigned_user_id)

    detail = client.datasets.get("a1b2c3d4", "t1t1t1t1", ds.dataset_id, tab=ReviewTab.REJECTED)
    print(client.datasets.actions("a1b2c3d4", "t1t1t1t1", ds.dataset_id))      # what you may do now
    print(client.datasets.file_tabs("a1b2c3d4", "t1t1t1t1", ds.dataset_id))    # {"All": 20, "Rejected": 2, "Approved": 5}

    client.datasets.initiate("a1b2c3d4", "t1t1t1t1", ds.dataset_id, "usr_annotator_1")   # assign to a worker
    client.datasets.set_priority("a1b2c3d4", "t1t1t1t1", 90, [ds.dataset_id])
    client.datasets.change_stage("a1b2c3d4", "t1t1t1t1", "Annotator 1", "Review 1", Selection.all_except())
    client.datasets.approve_file("a1b2c3d4", "t1t1t1t1", ds.dataset_id, "img_123", FileType.IMAGE)
    client.datasets.move_to_trash("a1b2c3d4", "t1t1t1t1", "Review 1", [ds.dataset_id])

Selections. Bulk methods take datasets (or files) as either a list of ids, meaning exactly those, or Selection.all_except(exclude=[...]), meaning everything the matching listing shows minus the excluded ids. For release and set_priority the listing filters (stage, dataset_status, search, ...) passed as keyword arguments scope an all-selection; change_stage and move_to_trash are always scoped by current_stage.

Methods

Backend routes under /projects/{pid}/tasks/{tid}/datasets (app/api/task_apis.py). All calls take project_id and task_id plus the workspace. Stage arguments are stage names as shown in the workflow ("Annotator 1", "Review 1", ...).

list(project_id, task_id, stage, *, page=1, limit=20, workspace_id=None, **filters) -> DatasetList

GET ...?stage=. Datasets in one stage, highest priority first then newest. stage is required. Filters: dataset_id, dataset_name, dataset_status (one of DatasetStatus.ALL), priority (0-100), last_action_by, search. Returns DatasetList(datasets, pagination, project_id, project_name, task_id, task_name) with total_items / total_pages properties. Workers only see datasets of stages they are assigned to.

iterate(project_id, task_id, stage, *, page_size=50, workspace_id=None, **filters) -> Iterator[Dataset]

Generator over every dataset of the stage, paging from page 1 to total_pages.

get(project_id, task_id, dataset_id, *, page=None, limit=None, tab=None, workspace_id=None) -> DatasetDetail | None

GET .../{dataset_id}. The dataset with its files in images / videos (property files joins them) and the task's labels. tab narrows the files to ReviewTab.REJECTED or APPROVED; page and limit page the files. None when the dataset is in the trash; NotFoundError when unknown; PermissionDeniedError for a worker not assigned to it.

actions(project_id, task_id, dataset_id, workspace_id=None) -> DatasetActions

GET .../{dataset_id}/actions. Booleans submit, complete, reject, skip, file_approve, file_reject for the caller in the dataset's current stage.

events(project_id, task_id, dataset_id, *, page=1, limit=20, workspace_id=None) -> DatasetEvents

GET .../{dataset_id}/events. Event trail newest first (created, assigned, released, submitted, rejected, stage_changed, priority_changed, ...), each with actor, details, created_at. Workforce Manager and above.

file_tabs(project_id, task_id, dataset_id, workspace_id=None) -> dict

GET .../{dataset_id}/files/tabs. File counts per review tab; annotators see All and Rejected only.

initiate(project_id, task_id, dataset_id, user_id, workspace_id=None, *, stage=None) -> str

POST .../initiate[?stage=]. Assigns one dataset to one worker who is listed on the dataset's current stage. A dataset is initiated once: ConflictError when already initiated, BadRequestError when the task's workflow has not been saved yet or the worker is not on the stage, NotFoundError for an unknown user. Workers may only initiate for themselves.

initiate_many(project_id, task_id, stage, datasets, user_ids, *, workspace_id=None, **filters) -> str

POST .../initiate?stage=. Assigns several datasets of stage in one call, one user per dataset: user_ids[i] gets the i-th selected dataset, and the counts must match (BadRequestError from the backend otherwise; checked locally when datasets is a list of ids). A user id may repeat to give that user several datasets. With a list of ids the pairing follows the list; with Selection.all_except(...) it follows the stage listing order (priority, then newest first), narrowed by the list filters. Workforce Manager and above.

client.datasets.initiate_many("a1b2c3d4", "t1t1t1t1", "Annotator 1",
                              Selection.all_except(), ["384f9", "394ff", "dc466"])

release(project_id, task_id, datasets, workspace_id=None, *, stage=None, **filters) -> str

POST .../release[?stage=]. Un-assigns initiated datasets and marks them Released. Managers may release many at once; a worker only the one assigned to them. stage scopes an all-selection.

skip / submit / complete / reject (project_id, task_id, dataset_id, workspace_id=None) -> DatasetTransition

POST .../skip|submit|complete|reject with {"dataset_id"}. Only the dataset's assigned worker may act (supervisors may also reject). skip gives the dataset back in place; submit moves it to the next working stage; complete moves it into the complete stage when that is next; reject sends it down the reject edge. The result carries message, auto_assigned and next_dataset, the dataset the backend handed to the worker next, if any. BadRequestError when the dataset is not initiated or the move is not allowed from its stage.

change_stage(project_id, task_id, current_stage, target_stage, datasets, workspace_id=None, **filters) -> str

POST .../stage?current_stage=. Moves selected datasets of current_stage to target_stage; ValueError when both are equal. Workforce Manager and above.

set_priority(project_id, task_id, priority, datasets, workspace_id=None, **filters) -> str

POST .../priority. priority is an int from 0 to 100; higher sorts first.

move_to_trash(project_id, task_id, current_stage, datasets, workspace_id=None) -> str

POST .../trash?current_stage=. Trashes the datasets and every file in them; restorable from the dashboard trash. Returns "<n> TaskDataset(s) moved to trash.".

trash_files(project_id, task_id, dataset_id, files, *, tab=None, workspace_id=None) -> str

POST .../{dataset_id}/files/trash[?tab=]. files is a list of file ids or Selection.all_except, which tab scopes to the rejected or approved files.

approve_file / reject_file (project_id, task_id, dataset_id, file_id, file_type, workspace_id=None) -> str

POST .../{dataset_id}/files/approve?file_id=&file_type=. file_type is FileType.IMAGE or FileType.VIDEO. Approving clears a rejection and vice versa.

Models and constants

Dataset: dataset_id, dataset_name, status, stage, priority, total_files, in_trash, restored, created_by, assigned_to, last_action_by, created_at, updated_at; properties name, assigned_user_id. DatasetDetail adds project_id, project_name, task_id, task_name, images, videos, labels and property files. DatasetList, DatasetEvents (iterable, with pagination), DatasetEvent, DatasetActions, DatasetTransition, Selection. DatasetStatus.ALL: Not Started, Assigned, In Progress, Completed, Released, Submitted, Rejected, Skipped. ReviewTab: All, Rejected, Approved. FileType: Image, Video.

---

Labels — client.labels

Usage

Labels belong to a task and are unique by name inside it.

from intellabel import IntelLabel, LabelType

with IntelLabel(workspace_id="<workspace uuid>") as client:
    labels = client.labels.list("a1b2c3d4", "t1t1t1t1")                 # LabelList, iterable
    car = client.labels.create("a1b2c3d4", "t1t1t1t1", "Car", [LabelType.RECTANGLE, LabelType.POLYGON])
    client.labels.add_attribute("a1b2c3d4", "t1t1t1t1", car.label_id, ["red", "blue", "white"])
    car = client.labels.rename("a1b2c3d4", "t1t1t1t1", car.label_id, "Vehicle")
    client.labels.create_many("a1b2c3d4", "t1t1t1t1", [
        {"name": "Person", "label_types": ["rectangle"], "color_code": "#FF0000"},
        {"name": "Pose", "label_types": ["keypoint_skeleton"],
         "skeleton": {"keypoints": [{"point_id": 1, "name": "head", "x": 0.5, "y": 0.1}, {"point_id": 2, "name": "neck", "x": 0.5, "y": 0.3}],
                      "edges": [[1, 2]]}},
    ])
    client.labels.delete("a1b2c3d4", "t1t1t1t1", car.label_id)

Methods

Backend router app/api/label_apis.py. All calls take project_id and task_id plus the workspace. label_id is the integer id of the label (also exposed as label_id).

list(project_id, task_id, *, compact=False, workspace_id=None) -> LabelList

GET /project/{pid}/task/{tid}/labels. Returns LabelList(labels, task_name), newest first. compact=True sends the backend's exclude flag and returns the reduced form used by annotation tools: name, types, colour and a flat list of attribute tags.

list_all(project_ids=None, task_ids=None, workspace_id=None) -> list[Label]

GET /labels. Every label in the workspace, narrowed to the given projects and/or tasks (lists or single strings). Super Admin or Admin only; PermissionDeniedError otherwise.

get(project_id, task_id, label_id, workspace_id=None) -> Label

GET /project/{pid}/task/{tid}/label/{label_id}. NotFoundError for an unknown id.

random_color(project_id, task_id, workspace_id=None) -> str

GET .../get/random_hex_code. A #RRGGBB colour not yet used by the task's labels.

create(project_id, task_id, name, label_types, *, color_code=None, attributes=None, skeleton=None, workspace_id=None) -> Label

POST .../make_label.

Parameter	Description
name	Required; unique in the task (ConflictError on a duplicate).
label_types	One or more of LabelType.ALL: polygon, rectangle, polyline, dot, ellipse, circle, mask, keypoint_skeleton.
color_code	#RRGGBB. When omitted, random_color is called first.
attributes	Optional list of {"attribute_name", "attribute_type", "attribute_values"} dicts created with the label.
skeleton	Required for keypoint_skeleton: {"keypoints": [{"point_id": int, "name", "x", "y", "color_code"?}], "edges": [[from, to], ...], "reference": {"width", "height"}?}. Validated locally like the backend does (unique integer ids, both coordinates, edges between existing distinct points).

create_many(project_id, task_id, labels, workspace_id=None) -> list[Label]

One request creating several labels. Each item is a dict with name (or label_name), label_types, optional color_code, attributes, skeleton; colours are fetched for items without one.

update(project_id, task_id, label_id, *, name=None, label_types=None, color_code=None, attributes=None, skeleton=None, workspace_id=None) -> Label

PUT .../label/{label_id}. Pass any subset of fields. The backend requires name, types and colour on every update, so missing ones are read from the current label first. ConflictError on a duplicate name. Renaming or retyping a label also updates the annotations that use it.

rename(project_id, task_id, label_id, name, workspace_id=None) -> Label

Shortcut for update(..., name=name).

delete(project_id, task_id, label_id, workspace_id=None) -> str

DELETE .../label/{label_id}. Removes the label and strips it from existing annotations. Returns the server message.

add_attribute(project_id, task_id, label_id, values, *, attribute_type="checkbox", workspace_id=None) -> str

POST .../label/{label_id}/add/attribute. Adds an attribute group with the given tags. Returns the server message; fetch the label to see the new attribute_id.

update_attribute(project_id, task_id, label_id, attribute_id, values, *, attribute_type="checkbox", workspace_id=None) -> Label

PUT .../update/attribute/{attribute_id}. Replaces the group's tags. Returns the updated label.

delete_attribute(project_id, task_id, label_id, attribute_id, workspace_id=None) -> str

DELETE .../delete/attribute/{attribute_id}.

import_labels(project_id, task_id, data, workspace_id=None) -> str

POST .../make_labels. Bulk import of raw label definitions (label_name, label_types, color_code, attributes, skeleton), for example copied from another task's compact listing.

Models and constants

Label: id, label_id, task_id, project_id, label_name, label_types, color_code, skeleton, attributes (list of Attribute), created_by, company_id, creation_date; property name. Attribute: id, attribute_id, attribute_type, attribute_values ([{"id", "tag"}]), property tags. LabelType.ALL, AttributeType.CHECKBOX.

---

Workflows — client.workflows

Usage

A workflow is the stage graph of a task: a dataset stage, one or more annotation stages, review stages and a complete stage, with edges between them and worker user ids on each stage's user_data. Every task receives the default layout once its files are uploaded.

from intellabel import IntelLabel, StageType, default_layout

with IntelLabel(workspace_id="<workspace uuid>") as client:
    wf = client.workflows.get("a1b2c3d4", "t1t1t1t1")
    print([s["name"] for s in wf.stages], wf.user_ids("Annotator 1"))

    annotators = client.workflows.workers(stage=StageType.ANNOTATION)        # who can be assigned
    wf = client.workflows.assign("a1b2c3d4", "t1t1t1t1", "Annotator 1", [annotators[0].user_id])
    wf = client.workflows.assign("a1b2c3d4", "t1t1t1t1", "Review 1", ["usr_qa_1"], replace=True)

    stages, edges = default_layout()                                           # build a custom graph
    stages[1]["user_data"] = ["usr_annotator_1", "usr_annotator_2"]
    wf = client.workflows.create("a1b2c3d4", "t1t1t1t1", stages, edges)

    counts = client.workflows.stages("a1b2c3d4", "t1t1t1t1")                  # datasets per stage
    users = client.workflows.stage_users("a1b2c3d4", "t1t1t1t1", stage="Review 1")

    progress = client.workflows.progress("a1b2c3d4", "t1t1t1t1")              # live file-based progress
    print(f"{progress.progress_percentage}% approved, {progress.rejected_files} rejected")
    for stage in progress.stages:
        print(stage.name, stage.files, f"{stage.share_percentage}% of files", f"{stage.approved_percentage}% approved")

    for snapshot in client.workflows.watch_progress("a1b2c3d4", "t1t1t1t1", interval=10):
        print(snapshot.progress_percentage)                                    # each change, until all approved

Methods

Backend router app/api/workflows.py. All calls take project_id and task_id plus the workspace unless noted.

get(project_id, task_id, workspace_id=None) -> Workflow

GET /projects/{pid}/tasks/{tid}/workflow. Stage user_data entries come back expanded to {user_id, first_name, last_name} and the dataset stage carries total_files and a file preview. A task without a workflow receives the default layout on this read, so the call succeeds for any existing task.

create(project_id, task_id, stages, edges, *, workflow_id=None, workspace_id=None) -> Workflow

POST /projects/{pid}/tasks/{tid}/workflow. Creates the workflow or replaces the existing one.

Parameter	Description
stages	Non-empty list of dicts with unique id, a type in StageType.ALL (dataset_stage, annotation_stage, review_stage, complete_stage), name, position and user_data (worker user id strings; expanded user dicts from get are accepted and reduced to ids).
edges	List of dicts with source and target stage ids (plus id, type, style, sourceHandle as in default_layout). Must reference existing stages.
workflow_id	The backend requires one in every payload. Pass the existing id when replacing a workflow; when omitted the SDK mints a new 6-character id, as the backend does.

The backend validates assignments: annotators (and annotator/QA hybrids) on annotation stages, QAs on review stages, every id must exist in the workspace (NotFoundError otherwise, BadRequestError for a wrong role). Saving moves all of the task's datasets back to the entry stage.

update(...)

Alias of create; the backend upserts on POST.

assign(project_id, task_id, stage, user_ids, *, replace=False, workspace_id=None) -> Workflow

Convenience: fetches the workflow, adds user_ids to the stage named or identified by stage (replace=True overwrites the stage's list), and saves. ValueError for an unknown stage or a non-assignable stage type.

delete(project_id, task_id, workspace_id=None) -> str

DELETE /projects/{pid}/tasks/{tid}/workflow. Returns "Workflow has been deleted.". The next read recreates the default layout with empty assignments, so this effectively resets the workflow.

stages(project_id, task_id, workspace_id=None, **filters) -> dict

GET .../workflow/stages. Dataset counts per stage plus total_datasets and workflow_id. Filters: stage, dataset_id, dataset_name, dataset_status, priority, last_action_by, search. Workers only see the stages they are assigned to.

progress(project_id, task_id, workspace_id=None) -> WorkflowProgress

GET .../workflow/progress. Live, file-based progress of the task through its workflow, computed on each call. The task's files are counted per dataset from their approved / rejected flags, and the datasets are grouped by the stage they sit in now. Workforce Manager and above.

summary holds the task totals:

Key	Meaning
total_datasets, completed_datasets	Live datasets; completed = status Completed or in a complete stage
total_files, approved_files, rejected_files, pending_files	File counts. A file that is approved is never counted as rejected. Pending = neither.
files_without_dataset	Files not in any live dataset (tasks created before datasets existed)
progress_percentage	Approved files / files. This is the percentage the analytics dashboard shows, so it equals client.analytics.progress(...) for the same task.
rejected_percentage, pending_percentage	Shares of all files
complete_stage_percentage	Files sitting in a complete stage / files
dataset_completion_percentage	Completed datasets / datasets

stages lists the workflow's stages in flow order (the dataset pool stage is left out, as in stages()), each a StageProgress with datasets, assigned_datasets, files, approved_files, rejected_files, pending_files, dataset_status (datasets per status) and four percentages: share_percentage (this stage's files / the task's files) and approved_percentage, rejected_percentage, pending_percentage (shares of this stage's own files). A dataset whose stage is no longer part of the workflow is counted in outside_workflow (otherwise None). Per-stage dataset counts equal those of stages().

watch_progress(project_id, task_id, *, interval=5.0, timeout=None, until_complete=True, workspace_id=None) -> Iterator[WorkflowProgress]

Polls progress every interval seconds and yields the first snapshot and then each one whose numbers changed. It stops once every file is approved (until_complete=False keeps watching) or, without raising, after timeout seconds. On the async client use async for.

stage_users(project_id, task_id, stage=None, workspace_id=None) -> list[StageUser]

GET .../workflow/users. Supervisors (Super Admin, Admin, Workforce Managers) plus the workers of the given stage. BadRequestError for an unknown stage name.

workers(stage=None, workspace_id=None) -> list[Worker]

GET /get/workers. Workers of the workspace who can be placed on workflows. stage must be annotation_stage or review_stage and narrows to the matching worker kind. Workforce Manager and above.

Models and helpers

Workflow: id, workflow_id, project_id, task_id, task_name, company_id, workflow_creator_id, created_at, updated_at, stages, edges; helpers stage(name_or_id), stages_of_type(type), user_ids(name_or_id). StageUser(user_id, first_name, last_name, role), Worker(id, user_id, first_name, last_name, role, checked). default_layout() returns the default (stages, edges) to start a custom graph from. WorkflowProgress: project_id, task_id, task_name, workflow_id, generated_at (UTC ISO time of the snapshot), summary, stages, outside_workflow; properties progress_percentage, total_files, approved_files, rejected_files, pending_files, is_complete; helper stage(name_or_id). StageProgress as described under progress.

---

Custom models — client.models

Usage

A workspace's custom models are add-on models linked from Hugging Face or Roboflow, and deployed models (your own .pt weights hosted by INTELLABEL, Growth / Enterprise plans). The built-in SAM2 is editor-only and not part of client.models.

from intellabel import IntelLabel, ModelTask, ModelFramework, Selection

with IntelLabel(workspace_id="<workspace uuid>") as client:
    # add-on models: provider is inferred from the link
    client.models.add("https://huggingface.co/facebook/detr-resnet-50", "hf_xxx")
    client.models.add("https://app.roboflow.com/my-ws/cars/model/3", "<roboflow api key>")

    detr = client.models.get("facebook/detr-resnet-50")
    print(detr.task, detr.labels)                      # object-detection ['person', 'car', ...]

    # one file: returns the predicted shapes, nothing is saved
    shapes = client.models.predict("a1b2c3d4", "t1t1t1t1", "img_123", detr, {"car": "Car", "person": "Person"},
                                   threshold=0.5)

    # many datasets: queued; the worker saves the annotations
    client.models.predict_bulk("a1b2c3d4", "t1t1t1t1", detr, {"car": "Car"}, Selection.all_except(),
                               stage="Annotator 1")

    # deploy your own weights, then wait until the model is listed
    job = client.models.deploy("runs/best.pt", model_name="cars-yolo", model_type="yolov8n",
                               task=ModelTask.OBJECT_DETECTION, framework=ModelFramework.ULTRALYTICS,
                               show_progress=True)
    model = client.models.wait_for_model("cars-yolo")

Label mapping. Predictions take label_mapping = {model_class: task_label_name}. Every class the model can return must be a key; values are label names of the task. Detection and segmentation need a non-empty mapping; bulk prediction always does.

Background jobs. predict_bulk and deploy return once the backend has queued the job. The workers report progress and errors only over the dashboard websocket, which the SDK can't join, so a failed deployment surfaces as DeployTimeoutError from wait_for_model (the reason is in the dashboard notifications).

Add-on model links (Hugging Face and Roboflow)

An add-on model is a model that already lives on Hugging Face or Roboflow. INTELLABEL stores its link and an access token, and every prediction calls the provider's hosted inference. This section is about add-on models only. Deployed models (your own .pt weights uploaded through POST /api/deploy/model) don't use a link.

The backend checks the link's shape before it contacts the provider. A link in the wrong shape is rejected with 400 and nothing is saved.

Roboflow: the link must include the model ID

Roboflow identifies a trained model by its model ID, <project>/<version>, for example hard-hat-sample-x1y2z/3. The version number is part of the model ID, so the link must end with it. A link to the project, the dataset or the workspace is rejected.

https://app.roboflow.com/acme-safety/hard-hat-sample-x1y2z/model/3
        \______________/ \_________/ \___________________/      |
             host         workspace       project slug       version

model ID           = hard-hat-sample-x1y2z/3
name in INTELLABEL = acme-safety/hard-hat-sample-x1y2z/3

Part	Example	Notes
Host	app.roboflow.com or universe.roboflow.com	app for models in your own workspace, universe for public Universe models. No other host is accepted.
Workspace	acme-safety	The Roboflow workspace slug, not the INTELLABEL workspace.
Project	hard-hat-sample-x1y2z	The project slug. Universe slugs often end in a random suffix; keep it.
Version	3	The trained version. Digits only.

Accepted forms:

Link	Use for
https://app.roboflow.com/<workspace>/<project>/model/<version>	Private model (recommended form)
https://app.roboflow.com/<workspace>/<project>/models/<model_name>/<version>	Private model, model page link. <model_name> is ignored.
https://app.roboflow.com/<workspace>/<project>/<version>	Private model, version page link
https://universe.roboflow.com/<workspace>/<project>/model/<version>	Public Universe model (recommended form)
https://universe.roboflow.com/<workspace>/<project>/<version>	Public Universe model

Rejected examples:

Link	Why
https://universe.roboflow.com/acme-safety/hard-hat-sample-x1y2z	No version, so no model ID
https://universe.roboflow.com/acme-safety/hard-hat-sample-x1y2z/dataset/3	Dataset tab. Open the Model tab instead, which gives .../model/3.
https://serverless.roboflow.com/hard-hat-sample-x1y2z/3, https://detect.roboflow.com/...	Inference URLs. Paste the app or Universe link; the backend builds the inference URL itself.
acme-safety/hard-hat-sample-x1y2z-3-rfdetr-t1	Newer training-specific model ID. Use the project slug and version number from the legacy <project>/<version> ID instead.

What the backend does with the link:

1. Reads https://api.roboflow.com/<workspace>/<project>/<version> with the key to get the project type, the classes and whether a trained model exists.
2. Accepts the project types object-detection, instance-segmentation and classification. Any other type is rejected with 400.
3. Saves the model as <workspace>/<project>/<version> and sends predictions to https://serverless.roboflow.com/<project>/<version>.
4. If the version has no trained model, the add still succeeds, but inference is false and predictions are refused.
5. If a version has several trainings, <project>/<version> runs the first successful one (Roboflow's legacy alias). The <model_name> in a /models/<model_name>/<version> link doesn't pick another training.

Which key to use:

• Private model: the Private API Key of the Roboflow workspace that owns the project. A key from another workspace gets 403.
• Public Universe model: any valid Roboflow Private API Key, normally your own. You don't need the owner's key. A 403 means the project isn't public.
• Don't use the Publishable key.
• Inference runs on Roboflow's hosted API under your key, so Roboflow counts the usage against that key's workspace. Commercial use of Universe models may need a paid Roboflow plan.

Roboflow: getting the link

Public model (Roboflow Universe)

1. Go to universe.roboflow.com and search for a model. Pick a project that has a trained model; such projects show a Model tab or a "Try This Model" button.
2. Open the project and click Model in the left sidebar, not Dataset or Images.
3. Choose the version in the version selector. The page shows the model ID as <project>/<version>.
4. Copy the URL from the address bar. It should end in /model/<version>, for example https://universe.roboflow.com/acme-safety/hard-hat-sample-x1y2z/model/3. If it ends at the project slug, append /model/<version> using the number from the model ID.
5. Get a key: sign in at app.roboflow.com (a free account works), open Settings → API Keys and copy the Private API Key.

Private model (your own workspace)

1. Sign in at app.roboflow.com and switch to the workspace that owns the project. The workspace slug is the first path segment of every URL in it (app.roboflow.com/<workspace>/...).
2. Open the project, then Models (Versions in older layouts), and open a version whose training has finished.
3. Copy the model ID shown on that page, <project>/<version>. The deploy code snippets show it too, as model_id="<project>/<version>".
4. Build the link from the workspace slug and the model ID: https://app.roboflow.com/<workspace>/<project>/model/<version>.
5. Get the key from the same workspace: Settings → API Keys → Private API Key.

Hugging Face: link structure

https://huggingface.co/facebook/detr-resnet-50
        \____________/ \______/ \____________/
             host        owner       repo

name in INTELLABEL = facebook/detr-resnet-50   (the repo ID)

Part	Example	Rules
Scheme and host	https://huggingface.co	https only. No www., and not the short hf.co host.
Owner	facebook	User or organization name: letters, digits, _ and -.
Repo	detr-resnet-50	Letters, digits, _, - and ..

The link is exactly https://huggingface.co/<owner>/<repo>. A trailing / is fine; anything else after the repo name is rejected.

Link	Why it's rejected
https://huggingface.co/facebook/detr-resnet-50/tree/main	Extra path (files tab)
https://huggingface.co/facebook/detr-resnet-50/blob/main/config.json	Link to a file, not the repo
https://huggingface.co/facebook/detr-resnet-50?library=transformers	Query string
facebook/detr-resnet-50	Repo ID only. Prefix it with https://huggingface.co/.
https://huggingface.co/spaces/<owner>/<space>, https://huggingface.co/datasets/<owner>/<name>	Spaces and datasets aren't models (400 or 404)

What the backend does with the link:

1. Reads the repo's model info with the token. A missing repo, or one the token can't see, fails with 404 or 401.
2. The repo's pipeline tag (the task shown under the model name) must be object-detection, image-segmentation or image-classification. Other tasks are rejected with 400.
3. The repo must have a model config (config.json), as Transformers models do. Labels come from it.
4. Predictions run on Hugging Face's hosted inference. At prediction time the backend asks Hugging Face for the model's inference status and runs only when it is warm, meaning a Hugging Face Inference Provider serves it. Otherwise the request is refused with "Inference can't be done at this moment".

The token must start with hf_ (20 to 100 characters after it) and is required for public models too.

Hugging Face: getting the link

Public model

1. Go to huggingface.co/models. Under Tasks, pick Object Detection, Image Segmentation or Image Classification. To list only models that can run, also filter Inference Providers → HF Inference.
2. Open the model. Check that the task tag under its name is one of the three above, and that the Inference Providers panel on the right lists it as available.
3. Copy the link. Either copy the address bar URL and cut everything after the repo name, or click the copy icon next to the model name (it copies <owner>/<repo>) and prefix https://huggingface.co/.
4. If the model is gated (the page asks you to accept its conditions), accept them with the same account the token belongs to.
5. Get a token: avatar → Settings → Access Tokens → Create new token. A Read token works. For a Fine-grained token, enable "Read access to contents of all public gated repos you can access". Copy the hf_... value; Hugging Face shows it only once.

Private model

1. Sign in with an account that owns the repo or belongs to the organization that does.
2. Open your profile or the organization page, go to Models and open the repo (marked Private).
3. Copy the link the same way as for a public model. It has the same shape, https://huggingface.co/<owner>/<repo>.
4. Create the token on that account. A Read token works. For a Fine-grained token, add the repo (or the organization) under repository permissions with read access to its contents. A token from an account without access fails with 401 or 404.
5. Hugging Face Inference Providers usually don't serve private repos, so the add can succeed while predictions are refused (see the inference check above). To run your own private weights, use a deployed model instead.

Adding the model from the SDK

with IntelLabel(workspace_id="<workspace uuid>") as client:
    # Roboflow, private: key of the workspace that owns the project
    client.models.add("https://app.roboflow.com/acme-safety/hard-hat-sample-x1y2z/model/3", "<private api key>")

    # Roboflow, public Universe model: your own key
    client.models.add("https://universe.roboflow.com/acme-safety/hard-hat-sample-x1y2z/model/3", "<private api key>")

    # Hugging Face, public or private: hf_ token with read access to the repo
    client.models.add("https://huggingface.co/facebook/detr-resnet-50", "hf_xxx")

The provider is inferred from the host. A link the backend rejects raises BadRequestError (400), a bad key AuthenticationError or PermissionDeniedError, and a model that is already added ConflictError.

Methods

Backend routes in app/api/custom_model.py, all under /api.

list(workspace_id=None) -> list[CustomModel]

GET /get/models. The workspace's add-on and deployed models. The built-in SAM2, which the backend also lists, is left out: it is editor-only and not a custom model.

get(model_name, workspace_id=None) -> CustomModel | None

Exact-name lookup over list() (there is no single-model route).

add(link, access_token, *, provider=None, workspace_id=None) -> str

POST /add/model. Add-on models only. Hugging Face: https://huggingface.co/<org>/<repo> with an hf_... token; the model is named <org>/<repo> and must be an object-detection, image-segmentation or image-classification pipeline. Roboflow: the link must end with the model ID (<project>/<version>), as in https://app.roboflow.com/<ws>/<project>/model/<version> or https://universe.roboflow.com/<ws>/<project>/model/<version>, with a Private API Key; the model is named <ws>/<project>/<version>. Link formats, key rules and how to find the link are in Add-on model links. ConflictError when already added. Workforce Manager and above.

delete(model_name, workspace_id=None) -> str

DELETE /delete/models. A deployed model's hosted weights are removed too.

update_access_token(model_name, access_token, provider, workspace_id=None) -> str

PUT /update/access_token. The backend takes the token as a query parameter, so it can show up in server access logs.

predict(project_id, task_id, file_id, model, label_mapping, *, provider=None, task_type=None, file_type="Image", threshold=None, timeout=120, workspace_id=None) -> list[dict]

POST /{pid}/{tid}/predict/model. Runs the model on one file and returns shapes (tool_type, tool_values, label_data, score, ...). Nothing is persisted. model is a CustomModel (provider and task are read from it) or a name with provider and task_type. Uses prediction credits.

predict_bulk(project_id, task_id, model, label_mapping, datasets, *, provider=None, task_type=None, threshold=None, clean_annotations=False, stage=None, workspace_id=None, **filters) -> ModelJob

POST /{pid}/{tid}/predict/images. Auto-annotates every image of the selected datasets (videos are skipped). datasets is a list of ids or Selection.all_except(...), narrowed by stage and the dataset list filters. clean_annotations=True first deletes those images' existing annotations.

delete_task_annotations(project_id, task_id, workspace_id=None) -> str

DELETE /projects/{pid}/tasks/{tid}/delete/task/annotations. Deletes every annotation of the task. Admin and above. Not reversible.

deploy(file, *, model_name, model_type, task, framework, input_size=640, confidence_threshold=0.25, iou_threshold=0.45, chunk_size=20 MB, on_bytes=None, show_progress=False, workspace_id=None) -> ModelJob

POST /deploy/model/upload-chunk (once, or per 20 MB chunk for larger files, like the dashboard; failed chunks are retried) then POST /deploy/model. file is a path or (filename, bytes | seekable file object) and must be a .pt. model_type is a free-form architecture name ("yolov8n"); task is a ModelTask; framework a ModelFramework; thresholds are in (0, 1]. Checks model_name against the workspace's models before uploading. PermissionDeniedError when the plan has no deployments or its cap is reached.

wait_for_model(model_name, *, interval=10, timeout=900, on_poll=None, workspace_id=None) -> CustomModel

Polls list() until the model appears. DeployTimeoutError after timeout seconds.

Models and constants

CustomModel: model_name, model_provider, model_type, model_link, inference, labels, data, has_separate_endpoint, model_key, added_by, created_at, id; properties name, provider, task, is_deployed. ModelJob(status, message, job_id). ModelProvider (HUGGINGFACE, ROBOFLOW), ModelTask (OBJECT_DETECTION, IMAGE_SEGMENTATION, IMAGE_CLASSIFICATION), ModelFramework (TRANSFORMERS, ULTRALYTICS, PYTORCH, TENSORFLOW, JAX, ONNX).

---

Cloud storages — client.cloud_storages

Usage

A cloud storage is a named bucket connection, referred to by display_name (for example by tasks.upload_from_cloud). Display names are unique across the company. Only AWS S3 with an access key pair is implemented by the backend; other providers are rejected locally because the backend answers them with a 200 that does nothing.

from intellabel import IntelLabel

with IntelLabel(workspace_id="<workspace uuid>") as client:
    print(client.cloud_storages.aws_regions())
    client.cloud_storages.add("raw-images", bucket_name="my-bucket", region="ap-south-1",
                              access_key_id="AKIA...", secret_access_key="...", description="Raw captures")

    for s in client.cloud_storages.list():
        print(s.display_name, s.bucket_name, s.region)

    page = client.cloud_storages.contents("raw-images", prefix="2026/09")
    images = [k for k in page.files if k.mime_type and k.mime_type.startswith("image/")]
    # CloudKey items go straight into an upload
    client.tasks.upload_from_cloud("a1b2c3d4", "t1t1t1t1", images, cloud_storage="raw-images", provider="AWS S3")

Methods

Backend routes in app/api/cloud_services.py, all under /api. Reads never return credentials (provider_data["authentication_data"] is None).

list(workspace_id=None) -> list[CloudStorage]

GET /get/storages. Newest first. Any role.

get(display_name, workspace_id=None) -> CloudStorage | None

GET /get/storage. None when the workspace has no storage by that name.

add(display_name, *, bucket_name, region, access_key_id, secret_access_key, description=None, prefix=None, endpoint_url=None, provider="AWS S3", workspace_id=None) -> str

POST /add/storage. The backend validates the region, the keys (sts:GetCallerIdentity) and bucket access (s3:HeadBucket) before saving: BadRequestError for a malformed key, NotFoundError for an unknown region or bucket, PermissionDeniedError when the keys can't reach the bucket, ConflictError when the name is taken. Workforce Manager and above.

update(display_name, *, bucket_name, region, access_key_id, secret_access_key, new_display_name=None, description=None, endpoint_url=None, provider="AWS S3", workspace_id=None) -> str

PUT /update/storage?display_name=. Replaces the settings; the bucket can't change and the keys are re-validated, so pass them again. description and endpoint_url are overwritten (omitting them clears them).

delete(display_name, *, provider="AWS S3", workspace_id=None) -> str

DELETE /delete/storage. Removes the configuration; the bucket is untouched.

contents(display_name, *, prefix=None, continuation_token=None, max_keys=100, provider="AWS S3", workspace_id=None) -> CloudContents

GET /cloud_storage/{display_name}/content. One level of the bucket under prefix: items are CloudKeys (type "DIR" with the summed size below it, or "FILE"), plus total_size, next (continuation token) and is_truncated; properties files and dirs. Workforce Manager and above.

iter_contents(display_name, *, prefix=None, page_size=100, provider="AWS S3", workspace_id=None) -> Iterator[CloudKey]

Every item at prefix, following continuation tokens. Not recursive.

aws_regions(workspace_id=None) -> list[str]

GET /get/aws_regions. Region names accepted for region, sorted.

Models

CloudStorage: id, display_name, description, provider, prefix, provider_data, workspace_id, company_id, created_by, user_id, role, created_at, updated_at; properties name, bucket_name, region. CloudContents(items, total_size, next, is_truncated).

---

Annotations — client.annotations

Usage

Export, download and import of a task's annotations. Both export and import run as backend jobs.

from intellabel import IntelLabel, AnnotationFormat

with IntelLabel(workspace_id="<workspace uuid>") as client:
    job = client.annotations.export("a1b2c3d4", "t1t1t1t1", AnnotationFormat.YOLO_DETECTION,
                                    file_name="cars", include_images=True)
    export = client.annotations.wait_for_export(job.job_id)          # polls the export history
    path = client.annotations.download_export(export, "exports/", show_progress=True)

    client.annotations.import_annotations("a1b2c3d4", "t9t9t9t9", path, AnnotationFormat.YOLO_DETECTION)

Downloads. A finished export into local storage gets a token-signed link (.../api/download/annotation?token=...). The SDK sends only the token to your client's base_url, so it works even when the server builds links with an internal host name. Links and files expire (server settings): an expired link raises PermissionDeniedError. Exports whose task targets a cloud bucket are written to that bucket instead and have nothing to download.

Background jobs. Workers report progress and errors only over the dashboard websocket. A failed export keeps status="PENDING", so wait_for_export ends with ExportTimeoutError; an import's outcome is visible in the task, not through the API.

Methods

Backend routes in app/api/annotations.py, all under /api. Formats: AnnotationFormat.ALL (COCO_JSON, COCO_KEYPOINTS, PASCAL_VOC, IMAGE_MASK_BINARY, YOLO_DETECTION, YOLO_DETECTION_TRACK, YOLO_ORIENTED_BOUNDING_BOXES, YOLO_POSE, YOLO_SEGMENTATION, YOLO_CLASSIFICATION, TAG_EXPORT).

export(project_id, task_id, export_format, *, file_name=None, include_images=False, include_videos=False, workspace_id=None) -> AnnotationJob

POST /projects/{pid}/tasks/{tid}/export/annotation. Queues the export and returns AnnotationJob(job_id, queued, job_status, message, ...). file_name names the export folder in the zip (default "<task_id>-export"); include_images / include_videos add the media.

list_exports(*, project_id=None, task_id=None, job_id=None, workspace_id=None) -> list[Export]

GET /exports. The workspace's export history, newest first, optionally filtered. Workforce Manager and above.

get_export(job_id, workspace_id=None) -> Export | None

The history row of one export job.

wait_for_export(job_id, *, interval=3, timeout=600, on_poll=None, workspace_id=None) -> Export

Polls until the row is SUCCESS. ExportTimeoutError (with .export, the last row) after timeout seconds.

download_export(export, dest=".", *, on_bytes=None, show_progress=False) -> Path

GET /download/annotation?token=. Streams the zip to dest: a folder (default ".", created if missing, relative to the current directory) where the server's file name is used, or a file path such as "cars.zip". Downloading again into the same folder keeps the earlier files: the new one is saved as name (1).zip, name (2).zip, ... An explicit file path is overwritten. export is an Export or its download URL. ValueError for an export that is not ready or went to cloud storage.

import_annotations(project_id, task_id, file, import_format, *, chunk_size=10 MB, on_bytes=None, show_progress=False, workspace_id=None) -> AnnotationJob

POST .../import/annotation/zip-chunk (once, or per 10 MB chunk for larger files, like the dashboard; failed chunks are retried) then POST .../import/annotation?import_format=. file is a .zip path or (filename, bytes | seekable file object). Returns the queued job (status, message).

Models

Export: id, job_id, project_id, task_id, task_name, export_format, file_name, status, path, target_storage, progress_id, workspace_id, created_at, updated_at, expiry_time; properties is_ready, storage_type, download_url. AnnotationJob(job_id, status, message, queued, worker_status, job_status). ExportStatus (PENDING, SUCCESS).

---

Analytics — client.analytics

Usage

Every figure is the one the INTELLABEL analytics dashboard shows. Durations are in milliseconds. Time comes from the annotation tool's timer, which records each annotation session per file and user.

import datetime as dt
from intellabel import IntelLabel, AnalyticsSection

with IntelLabel(workspace_id="<workspace uuid>") as client:
    # the main dashboard: workspace, one project, one project's tasks, or one user
    overview = client.analytics.overview()
    print(overview.mode, overview.summary["total_projects"], overview.total_time_spent_ms)
    project_view = client.analytics.overview("a1b2c3d4", start=dt.date(2026, 10, 1))
    task_view = client.analytics.overview("a1b2c3d4", ["t1t1t1t1", "t2t2t2t2"])
    user_view = client.analytics.overview(user_id="usr_annotator_1")

    # file-based progress per task (approved files / files)
    for row in client.analytics.progress("a1b2c3d4"):
        print(row["task_name"], row["files_completed"], "/", row["total_files"], row["progress_percentage"])

    # annotation timers: totals per user, task and file, plus a daily series
    timers = client.analytics.timers("a1b2c3d4", start="2026-10-01", end="2026-10-09")
    print(timers.summary["total_time_ms"], timers.summary["avg_session_ms"])
    for user in timers.users:
        print(user.name, user.time_ms, user.idle_ms, user.sessions, user.files)
    slowest = timers.files[0]                       # files are sorted by time, slowest first
    print(slowest.file_name, slowest.time_ms, slowest.users)

    # the timer of one file, per user (what the annotation tool shows)
    entries = client.analytics.file_timer("a1b2c3d4", "t1t1t1t1", "62ff2944", "Image")

    # the dashboard's time charts: daily time, idle vs active per hour, time per tool
    time = client.analytics.annotation_time("a1b2c3d4", user_id="usr_annotator_1")
    print(time.over_time["change_pct"], time.idle_by_hour["total_idle_ms"], time.by_tool)

    # annotations per class and time per task for chosen tasks (admin-scoped key)
    report = client.analytics.project("a1b2c3d4", ["t1t1t1t1"], label_names=["car"])
    print(report.label_counts("t1t1t1t1"), report.total_time_taken_per_task)

    # analytics access of Workforce Managers
    client.analytics.set_access("usr_wfm_1", {AnalyticsSection.ANNOTATION_TIME: True, AnalyticsSection.TEAM_WORKLOAD: True})
    print(client.analytics.get_access("usr_wfm_1").enabled_sections)

Live progress of a task through its workflow stages is client.workflows.progress (see Workflows).

Access and ranges

• Super Admins and Admins see all analytics. A Workforce Manager needs at least one section switched on by an admin (set_access) and receives only the sections switched on. timers needs the annotation_time section. An API key acts as its creator, so in practice a key is a Super Admin's or an Admin's. project and set_access also need the key's admin scope.
• start and end accept a datetime.date, a datetime.datetime or an ISO-8601 string ("2026-10-01", "2026-10-01T09:00:00Z"). A date covers the whole day and an end date includes that day. Naive datetimes are read in the company timezone and aware ones are exact. Either bound can be left out. An unparsable value raises ValueError locally. A start after the end raises BadRequestError.
• project_ids and task_ids take one id or a list of ids. Projects must belong to the workspace (NotFoundError otherwise). With no projects given, every project of the workspace is included.
• Analytics tables are paginated the backend's way: *_page is 1-based and *_limit=0 (the default) returns every row. The page info is returned with each table.

Methods

Backend routes in app/api/analytics.py (plus the file timer in app/api/annotations.py), all under /api.

overview(project_ids=None, task_ids=None, *, user_id=None, start=None, end=None, workspace_id=None, **pagination) -> AnalyticsOverview

GET /analytics/overview, the main analytics API that the dashboard calls. The filters choose the view, which mode names:

Filters	mode	Main sections
none, or several projects	"overview"	summary, project_progress, project_status_distribution, tasks, team_workload
one project	"project"	project, summary, task_progress, task_status_distribution, tasks, team_workload
one project + task_ids	"task"	project, tasks_info, summary, task_progress, file_status_distribution, tasks, team_workload
user_id	"user"	user, summary, assigned_work, file_status_distribution, productivity, review_overview
workspace without projects or tasks	"empty"	none

Every view also has annotations_over_time, annotation_time_over_time, idle_by_hour, annotation_distribution (by_tool, by_class) and auto_annotation. A section the caller may not see is None. The daily charts cover the range, or the last 7 days without one. Pagination keywords: progress_page / progress_limit (the progress table of the view), workload_*, unassigned_*, tool_*, class_*, auto_user_*. An unknown keyword raises ValueError. Annotators and QAs may only request their own user_id (PermissionDeniedError otherwise).

progress(project_ids=None, task_ids=None, *, user_id=None, start=None, end=None, workspace_id=None) -> list[dict]

The view's progress table with every row: one row per project for the workspace (total_tasks, tasks_completed, progress_percentage, annotation_time_ms, status), one row per task for a project (total_files, files_completed, progress_percentage, annotation_time_ms, status), or one row per assigned task for a user. progress_percentage is approved files / files.

timers(project_ids=None, task_ids=None, *, user_id=None, file_id=None, start=None, end=None, workspace_id=None, **pagination) -> AnnotationTimers

GET /analytics/timers. Annotation timer analytics of the selection:

• summary: total_time_ms, total_idle_ms, sessions, files, users, tasks, projects, avg_session_ms, avg_time_per_file_ms.
• users (TimerUser), tasks (TimerTask) and files (TimerFile, with a per-user split in users), each sorted by time, longest first.
• over_time: points: [{date, duration_ms, sessions}] per company-local day over the range (last 7 days without one), with total_ms and sessions.

Without a range, the stored per-file totals are summed. These are the same totals as the dashboard's total_time_spent_ms and the per-task annotation_time_ms. With a range, only timer sessions that started inside it count. Idle time comes from the idle events the annotation tool reports. user_id and file_id narrow to one user or one file. Pagination keywords: user_page / user_limit, task_*, file_*. Page info is in pagination["users" | "tasks" | "files"].

annotation_time(project_ids=None, task_ids=None, *, user_id=None, start=None, end=None, workspace_id=None) -> AnnotationTime

The time sections of overview for the same filters: total_time_ms, over_time (daily duration_ms with previous_total_ms and change_pct against the period before), idle_by_hour (points: [{hour, idle_ms, active_ms, days}], 24 hours) and by_tool ({tool: total_time_ms}).

file_timer(project_id, task_id, file_id, file_type="Image", workspace_id=None) -> list[FileTimerEntry]

GET /projects/{pid}/tasks/{tid}/get/timestamp. The timer of one file: one entry per user with total_duration_ms, name and updated_at. file_type is "Image" or "Video". Workers can read only the files of datasets assigned to them.

project(project_id, task_ids, *, label_names=None, start=None, end=None, workspace_id=None) -> ProjectAnalytics

GET /project/{pid}/analytics. For the given tasks of one project: total_annotations, total_annotations_per_class (per task: task_name and label_annotations), total_time_taken and total_time_taken_per_task (ms). label_names limits the counts to those classes. task_ids is required. Needs the admin scope.

users(workspace_id=None) -> list[AnalyticsUser]

GET /analytics/users. The users that can be picked for overview(user_id=...): everyone except Workforce Managers for a Super Admin, and Annotators and QAs (plus the caller) for an Admin.

get_access(user_id, workspace_id=None) -> AnalyticsPermissions

GET /analytics/access. The section toggles of a Workforce Manager.

set_access(user_id, permissions, workspace_id=None) -> AnalyticsPermissions

POST /analytics/access. Switches sections on or off ({section: bool}). Sections left out keep their state. Sections are checked locally against AnalyticsSection.ALL. The user must be a Workforce Manager (BadRequestError otherwise). Needs the admin scope. grant_all(user_id) and revoke_all(user_id) switch every section on or off.

Models and constants

AnalyticsOverview: one optional field per section listed above, with properties mode, is_empty, total_time_spent_ms and the helper progress_rows(). AnnotationTimers(timezone, range, summary, users, tasks, files, over_time, pagination); properties and helpers total_time_ms, user(account_id), task(task_id), file(file_id). TimerUser(account_id, name, role, time_ms, idle_ms, sessions, files, tasks, avg_session_ms). TimerTask(project_id, project_name, task_id, task_name, time_ms, sessions, files, users, avg_time_per_file_ms). TimerFile(file_id, file_type, file_name, project_id, task_id, time_ms, sessions, users). AnnotationTime(total_time_ms, over_time, idle_by_hour, by_tool). FileTimerEntry(user_id, name, total_duration_ms, updated_at). ProjectAnalytics with total_time_ms and label_counts(task_id). AnalyticsUser(id, role, full_name). AnalyticsPermissions(user_id, full_name, role, permissions) with enabled_sections and has_access. AnalyticsSection: the 18 section names (KPI_OVERVIEW, PROJECT_PROGRESS, TASK_PROGRESS, TEAM_WORKLOAD, ANNOTATION_TIME, IDLE_BY_HOUR, ...) and ALL.

---

Errors

from intellabel import IntelLabel, AuthenticationError, PermissionDeniedError, NotFoundError, APIError, APITimeoutError

try:
    client.get("/projects/123")
except AuthenticationError as e:      # 401: invalid / inactive / expired key
    ...
except PermissionDeniedError as e:    # 403: missing scope or role
    ...
except NotFoundError as e:            # 404
    ...
except APITimeoutError:               # transport timeout after retries
    ...
except APIError as e:                 # anything else from the server
    print(e.status_code, e.detail)

All inherit from intellabel.IntelLabelError.

Exception	When
APIConnectionError	The request never reached the server after retries.
APITimeoutError	Connect, read, write or pool timeout after retries. Subclass of APIConnectionError.
APIError	Any non-success HTTP status. Attributes status_code, detail, body, response.
BadRequestError (400)	Malformed input the SDK could not catch, missing workspace, storage limit, batch already completed.
AuthenticationError (401)	Invalid, inactive or expired API key, or the owning account lost access.
PermissionDeniedError (403)	Missing scope, insufficient role, deactivated or foreign workspace, or a /api-keys path (raised locally).
NotFoundError (404)	Unknown project, task, workspace or cloud storage.
ConflictError (409)	Duplicate project or task name.
UnprocessableEntityError (422)	Request-validation failure on the server.
RateLimitError (429)	Too many requests; retried automatically with Retry-After.
ServerError (5xx)	Backend failure.
UploadTimeoutError	wait_for_upload / iter_progress gave up; .progress is the last state.
ExportTimeoutError	annotations.wait_for_export gave up (or the export failed); .export is the last row.
DeployTimeoutError	models.wait_for_model gave up (or the deployment failed); .model_name is set.

Response envelope

The backend wraps successful responses as {"success": true, "status": 200, "output": ...}. client.request() and the verb helpers return an APIResponse with .success, .status, .output and .extra (any additional top-level fields). Resource modules return typed objects built from output.

Development

pip install -r requirements.txt
cp .env.example .env        # fill in API_KEY and WORKSPACE_ID
pytest                      # unit tests (mocked transport) + live tests
pytest tests/test_live.py   # live tests only

All tests read API_KEY from .env. Unit tests replace the network with httpx.MockTransport; live tests call the server and are skipped automatically when it is not reachable. Live tests move everything they create to the trash when they finish.

Project layout

intellabel/
  _client.py            _BaseClient (auth, retries, envelope) + IntelLabel (sync) + AsyncIntelLabel
  _auth.py              login() / logout(), saved credentials file, AuthInfo (whoami)
  cli.py, __main__.py   `intellabel login | whoami | logout` command line
  _types.py             APIResponse + dataclass base
  _progress.py          terminal progress bar / status line / byte-counting file proxy
  exceptions.py         error hierarchy
  resources/
    _base.py            Request spec, SyncResource / AsyncResource, shared validators
    projects.py         _ProjectsSpec (build + parse, no I/O), Projects, AsyncProjects
    tasks.py            _TasksSpec, Tasks, AsyncTasks   <- app/api/task_apis.py (+ uploads, progress; no datasets)
    datasets.py         _DatasetsSpec, Datasets, AsyncDatasets   <- task_apis.py dataset routes
    labels.py           _LabelsSpec, Labels, AsyncLabels   <- app/api/label_apis.py
    workflows.py        _WorkflowsSpec, Workflows, AsyncWorkflows   <- app/api/workflows.py
    custom_models.py    _ModelsSpec, Models, AsyncModels   <- app/api/custom_model.py
    cloud_storages.py   _CloudStoragesSpec, CloudStorages, AsyncCloudStorages   <- app/api/cloud_services.py
    annotations.py      _AnnotationsSpec, Annotations, AsyncAnnotations   <- app/api/annotations.py (export/import/download)
    analytics.py        _AnalyticsSpec, Analytics, AsyncAnalytics   <- app/api/analytics.py (+ the file timer route)
    _chunked.py         chunked single-file upload shared by models.deploy and annotations import
tests/
  conftest.py           API_KEY from .env; `mode` fixture runs each test on both clients
  test_client.py        core clients (mocked)
  test_live.py          core clients (live)
  test_auth.py          login / logout / whoami, credentials file, key resolution, CLI (mocked + live)
  test_projects.py      projects (mocked + live)
  test_tasks.py         tasks (mocked + live)
  test_datasets.py      datasets (mocked + live)
  test_labels.py        labels (mocked + live)
  test_workflows.py     workflows (mocked + live)
  test_models.py        custom models (mocked + live read-only)
  test_cloud_storages.py  cloud storages (mocked + live read-only)
  test_annotations.py   annotation export / download / import (mocked + live round trip)
  test_analytics.py     analytics, timers (mocked + live read-only cross-checks with workflow progress)

Each backend router gets one module under resources/: a spec class that builds requests and parses responses once, plus a sync and an async wrapper of a few lines each. One test file covers both wrappers. Next up: workspaces, trash.