# Omniverse Client Library — Python API Reference

Module: `omni.client` (pybind11 bindings over the C API).

## Table of Contents

1. [Calling Conventions](#calling-conventions)
2. [Core Classes](#core-classes)
3. [Enumerations](#enumerations)
4. [Lifecycle](#lifecycle)
5. [File Operations](#file-operations)
6. [Subscriptions](#subscriptions)
7. [Copy and Move](#copy-and-move)
8. [Checkpoints](#checkpoints)
9. [ACLs and Locking](#acls-and-locking)
10. [Authentication](#authentication)
11. [Connection Monitoring](#connection-monitoring)
12. [Live Objects](#live-objects)
13. [Live Sessions](#live-sessions)
14. [User and Group Management](#user-and-group-management)
15. [URL Utilities](#url-utilities)
16. [Cloud Storage Configuration](#cloud-storage-configuration)
17. [Bookmarks](#bookmarks)
18. [Miscellaneous](#miscellaneous)

---

## Calling Conventions

Every major operation has three variants:

```python
# Blocking — waits for completion, returns result tuple
result, entries = omni.client.list(url)

# Async — returns asyncio.Future
result, entries = await omni.client.list_async(url)

# Callback — returns Request object for manual control
request = omni.client.list_with_callback(url, callback)
```

The blocking and async variants are auto-generated wrappers around the callback versions.

## Core Classes

### Request
Manages async operation lifecycle.
```python
request = omni.client.stat_with_callback(url, callback)
request.id          # Unique request identifier
request.wait()      # Block until complete
request.stop()      # Cancel the operation
request.is_finished()  # Non-blocking check
# Also a context manager: with omni.client.stat_with_callback(...) as req:
```

### Registration
Manages callback subscriptions. Drop or set to `None` to unregister.
```python
reg = omni.client.register_connection_status_callback(callback)
reg.unregister()  # or: reg = None
# Also a context manager
```

### Content
Binary content buffer with Python buffer protocol support.
```python
result, version, content = omni.client.read_file(url)
data = memoryview(content).tobytes()  # Zero-copy access
length = len(content)
byte = content[0]
```

### Url
Parsed URL components.
```python
parsed = omni.client.break_url("omniverse://server:3009/Projects/file.usd?branch=main")
parsed.scheme    # "omniverse"
parsed.host      # "server"
parsed.port      # "3009"
parsed.path      # "/Projects/file.usd"
parsed.query     # "branch=main"
parsed.fragment  # None
parsed.is_raw    # False
parsed.user      # None
```

### ListEntry
File/folder metadata.
```python
result, entries = omni.client.list(url)
for e in entries:
    e.relative_path   # str: item name
    e.access          # AccessFlags bitmask
    e.flags           # ItemFlags bitmask
    e.size            # int: bytes
    e.modified_time   # datetime
    e.created_time    # datetime
    e.modified_by     # str: username
    e.version         # str: provider-specific
    e.hash            # str: content hash
    e.comment         # str: checkpoint comment
    e.deleted_time    # datetime (if soft-deleted)
    e.deleted_by      # str
    e.locked_by       # str or None
```

### ServerInfo
```python
result, info = omni.client.get_server_info(url)
info.version              # str
info.username             # str
info.auth_token           # str
info.connection_id        # str
info.cache_enabled        # bool
info.omniobjects_enabled  # bool
info.checkpoints_enabled  # bool
info.undelete_enabled     # bool
```

### WriteFileExInfo
```python
result, info = omni.client.write_file_ex(url, data)
info.version            # str
info.hash               # str
info.request_id         # str
info.checkpoint_query   # str
```

### AclEntry
```python
entry = omni.client.AclEntry()
entry.identity  # str: user or group name
entry.flags     # AccessFlags
```

### LiveSessionInfo
```python
result, sessions = omni.client.live_list_sessions(base_url)
for s in sessions:
    s.session_url        # str
    s.session_name       # str
    s.session_owner      # str
    s.modified_timestamp # int (ns since epoch)
```

### LiveSessionEvent
```python
event.event_type     # LiveSessionEventType
event.connection_id  # str
event.user_name      # str (JOINED only)
event.app_name       # str (JOINED only)
event.message_data   # bytes (MESSAGE only)
event.message_size   # int (MESSAGE only)
```

### AuthDeviceFlowParams
```python
params.server      # str
params.url         # str: user visits this
params.code        # str: user enters this
params.expiration  # int: seconds
```

## Enumerations

### Result
```python
omni.client.Result.OK
omni.client.Result.OK_LATEST              # Live: have latest version
omni.client.Result.OK_NOT_YET_FOUND       # Subscribe: monitoring
omni.client.Result.ERROR
omni.client.Result.ERROR_CONNECTION
omni.client.Result.ERROR_NOT_SUPPORTED
omni.client.Result.ERROR_ACCESS_DENIED
omni.client.Result.ERROR_NOT_FOUND
omni.client.Result.ERROR_ALREADY_EXISTS   # Copy/move destination exists
omni.client.Result.ERROR_SOURCE_IS_DEST
omni.client.Result.ERROR_ACCESS_LOST
omni.client.Result.ERROR_LOCKED
omni.client.Result.ERROR_BAD_REQUEST
omni.client.Result.ERROR_FOLDER_NOT_EMPTY
omni.client.Result.ERROR_WRONG_TYPE
omni.client.Result.ERROR_UNSUPPORTED_VERSION
```

### AccessFlags (combinable with |)
```python
omni.client.AccessFlags.READ
omni.client.AccessFlags.WRITE
omni.client.AccessFlags.ADMIN
```

### ItemFlags (combinable with |)
```python
omni.client.ItemFlags.READABLE_FILE
omni.client.ItemFlags.WRITEABLE_FILE
omni.client.ItemFlags.CAN_HAVE_CHILDREN     # Folder-like
omni.client.ItemFlags.DOES_NOT_HAVE_CHILDREN
omni.client.ItemFlags.IS_MOUNT
omni.client.ItemFlags.IS_INSIDE_MOUNT
omni.client.ItemFlags.CAN_LIVE_UPDATE
omni.client.ItemFlags.IS_OMNI_OBJECT
omni.client.ItemFlags.IS_CHANNEL            # Deprecated
omni.client.ItemFlags.IS_CHECKPOINTED
omni.client.ItemFlags.IS_DELETED
```

### CopyBehavior
```python
omni.client.CopyBehavior.ERROR_IF_EXISTS   # Default
omni.client.CopyBehavior.OVERWRITE
```

### ListIncludeOption
```python
omni.client.ListIncludeOption.NO_DELETED_FILES       # Default
omni.client.ListIncludeOption.INCLUDE_DELETED_FILES
omni.client.ListIncludeOption.ONLY_DELETED_FILES
```

### ConnectionStatus
```python
omni.client.ConnectionStatus.CONNECTING
omni.client.ConnectionStatus.CONNECTED
omni.client.ConnectionStatus.CONNECT_ERROR
omni.client.ConnectionStatus.DISCONNECTED
omni.client.ConnectionStatus.SIGNED_OUT
omni.client.ConnectionStatus.AUTH_ABORT
omni.client.ConnectionStatus.AUTH_CANCELLED
omni.client.ConnectionStatus.AUTH_ERROR
omni.client.ConnectionStatus.AUTH_FAILED
omni.client.ConnectionStatus.SERVER_INCOMPATIBLE
omni.client.ConnectionStatus.INVALID_HOST
```

### FileStatus
```python
omni.client.FileStatus.READING
omni.client.FileStatus.WRITING
omni.client.FileStatus.COPYING
omni.client.FileStatus.MOVING
omni.client.FileStatus.DELETING
omni.client.FileStatus.OBLITERATING
omni.client.FileStatus.LISTING
omni.client.FileStatus.STATING
```

### LogLevel
```python
omni.client.LogLevel.DEBUG
omni.client.LogLevel.VERBOSE
omni.client.LogLevel.INFO
omni.client.LogLevel.WARNING
omni.client.LogLevel.ERROR
```

### LiveUpdateType
```python
omni.client.LiveUpdateType.REMOTE   # From remote client
omni.client.LiveUpdateType.LOCAL    # Server ack of local update
omni.client.LiveUpdateType.MORE     # Queued update ready
```

### LiveSessionEventType
```python
omni.client.LiveSessionEventType.CONNECTION  # Your connection_id
omni.client.LiveSessionEventType.JOINED      # Peer joined
omni.client.LiveSessionEventType.LEFT        # Peer left
omni.client.LiveSessionEventType.MESSAGE     # Message received
```

## Lifecycle

```python
omni.client.get_version() -> str
omni.client.initialize(version=DEFAULT) -> bool
omni.client.shutdown() -> None
omni.client.set_product_info(name, version=None, extra=None)
omni.client.set_log_level(LogLevel)
omni.client.set_log_callback(callback)  # callback(str) or None to clear
omni.client.set_hang_detection_time_ms(timeout)
omni.client.write_chrome_trace()
```

## File Operations

### Stat
```python
stat(url, include_deleted_option=NO_DELETED_FILES) -> (Result, ListEntry)
stat_async(url, ...) -> (Result, ListEntry)
stat_with_callback(url, callback, ...) -> Request
# callback: (Result, ListEntry) -> None
```

### List
```python
list(url, include_deleted_option=NO_DELETED_FILES) -> (Result, List[ListEntry])
list_async(url, ...) -> (Result, List[ListEntry])
list_with_callback(url, callback, ...) -> Request
# callback: (Result, List[ListEntry]) -> None
```

### Read File
```python
read_file(url) -> (Result, str_version, Content)
read_file_async(url) -> (Result, str_version, Content)
read_file_with_callback(url, callback) -> Request
# callback: (Result, str_version, Content) -> None
# Use memoryview(content).tobytes() to get bytes
```

### Write File
```python
write_file(url, content: bytes, message="") -> Result
write_file_async(url, content, message="") -> Result
write_file_with_callback(url, content, callback, message="") -> Request

write_file_ex(url, content, message="", skip_checkpoint=False) -> (Result, WriteFileExInfo)
write_file_ex_async(url, content, message="", skip_checkpoint=False) -> (Result, WriteFileExInfo)
write_file_ex_with_callback(url, content, callback, message="", skip_checkpoint=False) -> Request
```

### Create Folder
```python
create_folder(url) -> Result
create_folder_async(url) -> Result
create_folder_with_callback(url, callback) -> Request
```

### Delete
```python
delete(url) -> Result                              # Recursive
delete_single(url) -> Result                       # File or empty folder only
undelete(url) -> Result                            # Restore soft-deleted
obliterate(url, obliterate_checkpoints: bool) -> Result  # Permanent
# All have _async and _with_callback variants
```

### Get Local File
```python
get_local_file(url, download=True) -> (Result, str_local_path)
# For Hub cache: call request.stop() when done to release the file
```

## Subscriptions

```python
# Stat subscribe: initial stat + ongoing updates
stat_subscribe_with_callback(url, callback, include_deleted_option=...) -> Request
# callback: (Result, ListEvent, ListEntry) -> None

# List subscribe: initial list + change events
list_subscribe_with_callback(url, list_callback, subscribe_callback, include_deleted_option=...) -> Request
# list_callback: (Result, List[ListEntry]) -> None
# subscribe_callback: (Result, ListEvent, ListEntry) -> None
```

`ListEvent` values: `UNKNOWN`, `CREATED`, `UPDATED`, `DELETED`, `METADATA`, `LOCKED`, `UNLOCKED`, `OBLITERATED`.

## Copy and Move

```python
copy(src, dst, behavior=ERROR_IF_EXISTS, message="") -> Result
copy_file(src, dst, behavior=ERROR_IF_EXISTS, message="") -> Result
copy_folder(src, dst, behavior=ERROR_IF_EXISTS, message="") -> Result

move(src, dst, behavior=ERROR_IF_EXISTS, message="") -> (Result, bool_copied)
move_file(src, dst, behavior=ERROR_IF_EXISTS, message="") -> (Result, bool_copied)
move_folder(src, dst, behavior=ERROR_IF_EXISTS, message="") -> (Result, bool_copied)
# copied=True means cross-server copy+delete; False means server-side rename
# All have _async and _with_callback variants
```

## Checkpoints

```python
create_checkpoint(url, comment, force=False) -> (Result, str_query)
list_checkpoints(url) -> (Result, List[ListEntry])
# ListEntry.relative_path contains the checkpoint query string
# All have _async and _with_callback variants

# Parse/build queries
get_branch_and_checkpoint_from_query(query) -> (str_branch, int_checkpoint)
make_query_from_branch_and_checkpoint(branch, checkpoint) -> str
```

## ACLs and Locking

```python
get_acls(url) -> (Result, List[AclEntry])
set_acls(url, acls: List[AclEntry]) -> Result

lock(url) -> Result      # Auto-unlocked on disconnect
unlock(url) -> Result
# All have _async and _with_callback variants
```

## Authentication

### Auth Callback
```python
def my_auth(prefix: str):
    """Return credentials for a URL prefix."""
    return ("username", "password")  # Tuple[str, str] for user/pass
    return "bearer_token"            # str for token
    return omni.client.AUTH_ABORT    # True to abort
    return None                      # None to try next method

reg = omni.client.register_authentication_callback(my_auth)
# Drop reg or set to None to unregister
```

### Device Flow
```python
def device_flow(auth_handle: int, params: AuthDeviceFlowParams | None):
    if params:
        print(f"Visit {params.url} and enter code {params.code}")
    else:
        print("Auth complete")

reg = omni.client.register_device_flow_auth_callback(device_flow)
```

### Storage Auth (OmniStorage)
```python
def storage_auth(auth_handle: int, server: str) -> bool:
    token = get_token_somehow(server)
    omni.client.set_storage_access_token(auth_handle, token)
    return True  # True = handling auth; False = fallback

reg = omni.client.register_storage_auth_callback(storage_auth)
```

**Important**: `register_storage_auth_callback` returns a `Registration` object. The `auth_handle` integer passed to the callback is different — use `auth_handle` with `set_storage_access_token`, keep the `Registration` object alive to maintain the callback.

### Other Auth Functions
```python
authentication_cancel(auth_handle: int)
sign_out(url)         # Disconnect + clear token
reconnect(url)        # Force retry
refresh_auth_token(url) -> (Result, str_token)
```

## Connection Monitoring

```python
def on_status(url: str, status: ConnectionStatus):
    if status == omni.client.ConnectionStatus.CONNECTED:
        print(f"Connected to {url}")

reg = omni.client.register_connection_status_callback(on_status)

def on_file_status(url: str, status: FileStatus, percent: int):
    print(f"{status}: {url} ({percent}%)")

reg = omni.client.register_file_status_callback(on_file_status)
```

## Live Objects

For Nucleus omni-object live collaboration. Requires `omniverse://` URLs.

```python
# Create live object
live_create(url, content: bytes) -> (Result, int_objectId, int_sequenceNum)

# Read live object (subscribe to updates)
live_read_with_callback(url, callback, have_object_id=0, have_sequence_num=0) -> Request
# callback: (Result, int_objectId, int_sequenceNum, Content) -> None

# Update live object
live_update(url, object_id: int, content: bytes) -> (Result, int_sequenceNum, int_requestId)

# Register for update notifications
reg = omni.client.live_register_queued_callback(callback)
# callback: (LiveUpdateType, Result, str_url, int_objectId, int_sequenceNum, int_serverTime) -> None

# Process updates — MUST be called from main thread
omni.client.live_process()
omni.client.live_process_up_to(url, server_time: int)
omni.client.live_get_latest_server_time(url) -> int
omni.client.live_wait_for_pending_updates()

# Send message in a joined channel
live_send_message(join_request_id: int, content: bytes) -> Result
```

## Live Sessions

Higher-level collaboration for session management and user presence.

```python
# List sessions for a layer
live_list_sessions(base_layer_url) -> (Result, List[LiveSessionInfo])

# Create session
live_create_session(base_layer_url, session_name="") -> (Result, str_sessionUrl)

# Join session (callback-only, long-lived)
def on_event(result: Result, event: LiveSessionEvent):
    if event.event_type == LiveSessionEventType.CONNECTION:
        my_id = event.connection_id
    elif event.event_type == LiveSessionEventType.JOINED:
        print(f"{event.user_name} joined via {event.app_name}")
    elif event.event_type == LiveSessionEventType.LEFT:
        print(f"User left: {event.connection_id}")
    elif event.event_type == LiveSessionEventType.MESSAGE:
        print(f"Message: {event.message_data}")

request = omni.client.live_join_session_with_callback(session_url, "MyApp", on_event)
# First event is CONNECTION with your connection_id

# Send message to all session members
omni.client.live_send_message(request.id, b"Hello!")

# Leave session
request.stop()
```

## User and Group Management

```python
get_groups(url) -> (Result, List[str])
get_group_users(url, group) -> (Result, List[str])
create_group(url, group) -> Result
rename_group(url, group, new_group) -> Result
remove_group(url, group) -> (Result, int_changeCount)
get_users(url) -> (Result, List[str])
get_user_groups(url, user) -> (Result, List[str])
add_user_to_group(url, user, group) -> Result
remove_user_from_group(url, user, group) -> Result
# All have _async and _with_callback variants
```

## URL Utilities

```python
break_url(url) -> Url
break_url_reference(url) -> Url                      # RFC 3986
make_url(scheme=None, user=None, host=None, port=None,
         path=None, query=None, fragment=None) -> str
make_file_url(path) -> str
normalize_url(url) -> str
combine_urls(base_url, other_url) -> str
combine_with_base_url(other_url) -> str
make_relative_url(base_url, other_url) -> str
make_printable(url) -> str

set_alias(alias, value)  # value=None to remove

# Base URL stack (thread-local)
push_base_url(url)
pop_base_url(url)
get_base_url() -> str | None
```

## Cloud Storage Configuration

```python
# S3
set_s3_configuration(
    url, bucket=None, region=None,
    accessKeyId=None, secretAccessKey=None, sessionToken=None,
    cloudfrontUrl=None, cloudfrontForList=False, writeConfig=False,
    canonicalHost=None
) -> Result

# Azure
set_azure_sas_token(host, container, sasToken, writeConfig=False) -> Result

# Generic HTTP header
set_http_header(key, value)  # value=None to remove

# Clear cached list/stat results
clear_list_cache(url) -> bool

# Storage registration
register_storage_from_discovery(discovery_url) -> (Result, List[str])
register_storage_direct(url, headers=None) -> (Result, List[str])
# Both have _async and _with_callback variants
```

## Bookmarks

```python
add_bookmark(name, url)
remove_bookmark(name)
list_bookmarks_with_callback(callback) -> Request
# callback: (Dict[str, str]) -> None
```

## Miscellaneous

```python
get_server_info(url) -> (Result, ServerInfo)
get_hub_version() -> (Result, str)
get_hub_http_uri() -> (Result, str)

set_retries(timeout, retry_delay, max_retries) -> (int, int, int)  # Returns previous
enable_nucleus_cache_bypass(bypass_on: bool) -> bool

# Resolve (search paths)
resolve(url, search_urls: List[str]) -> (Result, ListEntry, str_resolvedUrl)
resolve_subscribe_with_callback(url, search_urls, callback, subscribe_callback) -> Request

add_default_search_path(path)
remove_default_search_path(path)
get_default_search_paths() -> List[str]

# Tracing
trace_start() -> bool
trace_stop()
```
