# Nebius Python® SDK

The **Nebius Python® SDK** is a client library for
[Nebius AI Cloud](https://nebius.com) services. It uses gRPC and supports all
APIs in the [Nebius API repository](https://github.com/nebius/api). Use the SDK
to authenticate, manage resources, and communicate with Nebius services.

> **Note**: "Python" and the Python logos are trademarks or registered trademarks of the Python Software Foundation, used by Nebius B.V. with permission from the Foundation.

### Full documentation and reference

See the [API reference](https://nebius.github.io/pysdk/apiReference.html) for
all services and methods.

### Installation

```bash
pip install nebius
```

If you have a ZIP archive or a Git checkout, install the SDK from its directory:

```bash
pip install ./path/to/your/pysdk
```

### Migration from 0.4.x to 0.5.x

Version 0.5.0 runs all SDK asynchronous work on one internal event loop per SDK
instance. By default, each SDK instance owns this loop and its daemon thread.
It also owns a daemon executor with a maximum of two worker threads. You can
set the maximum with `executor_max_workers`.

Some 0.4.x workflows do not work in 0.5.0. Use these replacements:

| 0.4.x workflow | 0.5.x behavior | Required change |
| --- | --- | --- |
| Treat `Channel.bg_task()` as an `asyncio.Task`. | The method returns a reusable `CrossLoopAwaitable`. | Await it directly. Before you pass it to `asyncio.wait()`, wrap it with `asyncio.ensure_future()`. |
| Call `run_sync()` or `sync_close()` from an asynchronous call stack. | The SDK rejects the call from every active caller loop. | Await the SDK handle. You can also move the complete synchronous call to `asyncio.to_thread()`. |
| Give the SDK a stopped `event_loop`. | A supplied loop must already run. The caller continues to own it. | Start the loop in its owner thread. Keep it running until SDK close is complete. |

#### Runtime ownership and shutdown

You can give an already-running `event_loop` to `SDK`. The SDK does not stop
or reconfigure this loop during `SDK.close()`. The caller must keep the loop
responsive until SDK close is complete.

Set `loop_exception_handler` to a synchronous asyncio exception handler. Do not
use an `async def` function. A synchronous wrapper must also return `None`
instead of a coroutine or another awaitable. The SDK rejects directly
recognizable async functions. If a synchronous handler returns any other value,
the SDK closes a newly returned, unstarted native coroutine and reports both
the original context and the contract violation through asyncio's default
exception handler. It does not change a suspended coroutine, returned Future,
Task, or opaque awaitable because the handler might not own that work. The SDK
cannot know whether an invalid handler processed the original context, so
default reporting can duplicate a diagnostic that the handler already emitted.
The loop calls the handler on its
thread, and the handler must return promptly. A blocking handler stops all work
on that loop. Request and operation failures still propagate through their
returned awaitables.

With an SDK-owned loop, the handler remains installed until the loop closes.
On a supplied loop, the handler receives diagnostics from SDK work and other
loop users. It replaces the loop's current handler and remains installed after
SDK close. It starts receiving diagnostics after all other SDK initialization
succeeds. A loop has one exception handler. After construction succeeds, a
later loop assignment replaces that handler. If you construct the SDK from
another thread, the constructor waits for the supplied loop to install the
handler for up to 30 seconds. Keep the loop responsive during SDK construction.
Construction raises `RuntimeError` if the supplied loop stops or does not
install the handler before this time limit.
If construction fails after the SDK starts to use a caller-supplied loop, the
SDK starts cleanup before it propagates the error. Cleanup can continue after
the constructor returns.

The event loop stores an SDK forwarding callable for the handler.
`loop.get_exception_handler()` does not have to return the same callable that
you passed to the SDK. Handler installation is the final SDK initialization
action. If an asynchronous `BaseException` arrives after the loop accepts the
handler but before the constructor returns, the handler can remain installed
even though the constructor did not return an SDK object.

The custom handler replaces asyncio's current handler. It does not
automatically preserve default reporting. To keep asyncio's default reporting,
call the default handler explicitly:

```python
def report_loop_exception(loop, context):
    record_redacted_fields(context)
    loop.default_exception_handler(context)


sdk = SDK(loop_exception_handler=report_loop_exception)
```

An exception context can contain sensitive data and tasks, futures, or handles
owned by the event loop. Read loop-owned objects only on that loop. Copy and
redact the required immutable fields before another thread processes them. Do
not log or export the complete context without checking its contents. The loop
retains the handler after SDK close. The handler can therefore retain objects
that it captures until another handler replaces it or the loop closes.

The caller also owns the supplied loop's default executor. Do not fill this
executor with synchronous SDK waits. SDK work or custom extensions can need
the same executor. Await SDK handles, or use independent executors with enough
capacity.

An owned executor starts worker threads only when it receives work. Shutdown
waits for submitted executor work. A function that does not return can delay
shutdown without a time limit. The function must stop cooperatively because
Python cannot safely stop a running thread.

Fork the process before you create SDK or gRPC objects. Create separate SDK
objects after each child process starts. The SDK rejects an object that a
child process inherits from its parent. Event-loop threads, gRPC state, and
locks cannot move safely to the child.

Do not call synchronous helpers from an asynchronous call stack. Await the SDK
handle instead. For a synchronous low-level pool method, use
`asyncio.to_thread()`. Application threads can call synchronous helpers at the
same time if the SDK loop stays responsive. Do not call these helpers from an
SDK executor worker.

When SDK work calls `close()` on the internal loop, the code immediately after
`await close()` can finish. Shutdown cancels that code if it waits again.

#### Custom extensions and data

The SDK loop runs custom resolvers, interceptors, authorization providers,
token bearers, and asynchronous metrics callbacks. The SDK fixes request
options before submission. Custom callback and credential objects must be
thread-safe and loop-neutral.

Do not store application-loop objects in a custom callback or credential
object. These objects include asyncio locks, events, queues, tasks, and client
sessions. Thread-local application state does not move to the SDK thread. Use
a `ContextVar` when the SDK work needs application context.

Use one stateful authorization provider or token bearer for one SDK. Do not
share it between SDK instances that use different loops. You can share an
immutable implementation only if it supports concurrent calls and independent
`close()` calls. The SDK cannot detect hidden loop ownership or mutable state.

The SDK closes a rejected native coroutine. It sends cleanup for a rejected
`asyncio.Future` to the Future's owner loop. The SDK does not call `close()` on
an unknown custom awaitable from a cancellation or shutdown thread. Such an
awaitable must provide thread-safe cleanup for work that does not start.

When a custom transport belongs to a different caller-owned event loop, the
SDK schedules cleanup on that loop and does not wait for it during shutdown.
Keep that loop running and able to process callbacks until cleanup finishes.

Supported mutable request messages become fixed when the request wrapper is
created. This rule keeps generated reset-mask metadata consistent with the
message. Set metadata, timeouts, credentials, and authentication options
before the first await or `.wait()` call. Submission fixes these options.

`Channel.run_async()` is a new optional custom-channel method. The built-in
channel uses it to schedule work immediately and return a cross-loop handle.
The minimum custom-channel protocol did not change. A legacy custom channel
can omit this method and use its local-awaitable fallback.

The fallback does not make a legacy channel cross-loop. If an operation or
stream first uses a contended asyncio lock on one caller loop, continue to use
that object on the same loop. Implement `run_async()` to support cross-loop
scheduling.

#### RPC and stream behavior

Low-level `Channel.unary_unary()` calls copy metadata and supported request
messages when the call wrapper is created. The SDK serializes a custom request
immediately on the thread that constructs the call when a serializer exists.
The serializer must be thread-safe, loop-neutral, and return promptly. Without
a serializer, the custom value must be immutable or safe to share between
threads.

After a native unary call ends, its metadata and status awaitables must finish
promptly. Shutdown drains these awaitables to preserve the RPC result. A
custom transport that leaves an accessor pending can delay final shutdown.
`sync_close(timeout=...)` limits the caller's wait, but it cannot stop custom
transport code safely.

Channel cleanup is best effort for compatibility. The SDK logs each resource
cleanup failure after it tries to close all resources. These failures do not
make `close()` fail. In contrast, `close()` and `sync_close()` report failures
from runtime finalization.

Generated operation-service factories remain synchronous. You can call them
from asynchronous application code. They resolve the source service on the
SDK loop when the first operation RPC starts. They then keep that address for
the operation-service adapter. This behavior keeps all polls for one operation
on one endpoint.

Direct synchronous resolver and pool helpers still reject active event loops.
Use a generated client. You can also move an explicit synchronous call to an
application thread.

The SDK loop consumes a caller-supplied asynchronous iterator for a client
stream. The iterator must be loop-neutral. The SDK does not support an
iterator with state that belongs to another event loop.

The SDK bridges an explicit `asyncio.Future` through its owner loop. The owner
loop must stay running until the bridge copies the result. This rule also
applies when the Future is already complete. If the owner loop stops, the
bridge fails without reading the Future from another thread.

A unary streaming request and its authentication options become fixed when
the SDK creates the stream wrapper. Each client-streaming `write()` copies a
supported request message before SDK-loop dispatch. An unknown custom value
is not copied and must be safe to share between threads.

Low-level pool methods still expose `AddressChannel.channel` for custom
transport compatibility. This field is a native `grpc.aio.Channel` that
belongs to the SDK loop. Do not call it from an external loop. Use a generated
client or `Channel.unary_unary()` for cross-loop calls.

Each `AddressChannel` wrapper represents one pool lease. Release the wrapper
only once. A later checkout can return a new wrapper for the same native
channel. Custom wrappers can override `_new_lease()` to copy required wrapper
state.

`get_authorization_provider()` still returns the configured provider object.
Generated requests dispatch it on the SDK loop. Code that calls the provider
directly is responsible for correct loop and thread use.

#### Cross-loop handles

`Channel.bg_task()` now returns a cross-loop SDK handle. It does not return an
`asyncio.Task`. Await the handle directly. Wrap it with
`asyncio.ensure_future()` before you pass it to `asyncio.wait()`. Task-only
inspection, naming, and callback-removal methods are not available.

Pending `result()` and `exception()` calls can block an application thread.
They reject active event loops and SDK executor workers. Asynchronous code
must await the handle. A handle cannot await itself.

All direct waiters share cancellation. The handle can report cancellation
before an asynchronous finalizer is complete. The `cancel(msg)` method accepts
and ignores the message. After cancellation, `result()` and `exception()`
raise `concurrent.futures.CancelledError`.

Completion callbacks run on the event loop that registered them. This loop
must run until it delivers the callback. Registration on a stopped or closed
loop raises `RuntimeError`. If the loop stops later, the SDK logs a warning.
It does not move the callback to another thread.

### Migration from 0.3.x to 0.4.x

Version 0.4.0 replaces the provider-backed generated API layer. The new layer
contains SDK-owned message, enum, reflection, and service-client
implementations. Imports from public package facades and client calls continue
to work:

```python
from nebius.api.nebius.compute.v1 import Instance, InstanceServiceClient
```

The SDK no longer contains generated `*_pb2.py`, `*_pb2.pyi`, and
`*_pb2_grpc.py` modules. Change code that imported these modules directly.
Also change code that depended on `google.protobuf.message.Message` or used the
global protobuf descriptor pool. Use the package-facade exports and the SDK
registry and reflection APIs instead. The SDK now supplies its required Google
protobuf and RPC types in `nebius.api.google.protobuf` and
`nebius.api.google.rpc`.

### Migration from 0.2.x to 0.3.x

Version 0.3.0 contains these changes to authorization:

- Authorization options moved to a direct request argument.
- The SDK removed `nebius.aio.authorization.options.options_to_metadata`.
- The SDK removed unused metadata cleanup.

`<= 0.2.74`:
```python
from nebius.aio.authorization.options import options_to_metadata
service.request(
    req,
    metadata=({'your':'metadata'}).update(options_to_metadata(
        {
            OPTION_RENEW_REQUIRED: "true",
            OPTION_RENEW_SYNCHRONOUS: "true",
            OPTION_RENEW_REQUEST_TIMEOUT: ".9",
        }
    ))
)
```

`>= 0.3.0`:
```python
service.request(
    req,
    metadata={'your':'metadata'},
    auth_options={
        OPTION_RENEW_REQUIRED: "true",
        OPTION_RENEW_SYNCHRONOUS: "true",
        OPTION_RENEW_REQUEST_TIMEOUT: ".9",
    }
)
```

### Example

Working examples are in `src/nebius/examples`. From the repository root, run
an example as follows:
```bash
NEBIUS_IAM_TOKEN=$(nebius iam get-access-token) PYTHONPATH=src \
python -m nebius.examples.basic your-project-id
```

### How-to

#### Initialize

```python
from nebius.sdk import SDK

sdk = SDK(user_agent_prefix="example-application/1.0")
```

This code initializes the
[SDK](https://nebius.github.io/pysdk/nebius.sdk.SDK.html) with an IAM token
from the `NEBIUS_IAM_TOKEN` environment variable. Replace
`example-application/1.0` with the name and version of your application.

The following sections show the supported credential sources.

##### Initialize with an IAM token

You can give the token directly to the SDK. You can also read it from a
different environment variable:

```python
import os
from nebius.sdk import SDK
from nebius.aio.token.static import Bearer, EnvBearer  # [1]
from nebius.aio.token.token import Token  # [2]

sdk = SDK(
    credentials=os.environ.get("NEBIUS_IAM_TOKEN", ""),
    user_agent_prefix="example-application/1.0",
)
# or
sdk = SDK(
    credentials=Bearer(os.environ.get("NEBIUS_IAM_TOKEN", "")),
    user_agent_prefix="example-application/1.0",
)
# or
sdk = SDK(
    credentials=EnvBearer("NEBIUS_IAM_TOKEN"),
    user_agent_prefix="example-application/1.0",
)
# or
sdk = SDK(
    credentials=Bearer(Token(os.environ.get("NEBIUS_IAM_TOKEN", ""))),
    user_agent_prefix="example-application/1.0",
)
```
[[1](https://nebius.github.io/pysdk/nebius.aio.token.static.html), [2](https://nebius.github.io/pysdk/nebius.aio.token.token.html)]

Each example gets the same token through a different credential interface.

##### Initialize using CLI config

If you configured the [Nebius AI Cloud CLI](https://docs.nebius.com/cli), you
can initialize the SDK with its
[CLI configuration](https://nebius.github.io/pysdk/nebius.aio.cli_config.Config.html):

```python
from nebius.sdk import SDK
from nebius.aio.cli_config import Config

sdk = SDK(
    config_reader=Config(),
    user_agent_prefix="example-application/1.0",
)
```

The SDK also reads the domain from the configured endpoint if you do not set
the domain explicitly.

The SDK uses the token in `NEBIUS_IAM_TOKEN` if this variable is set. It uses
`NEBIUS_PROFILE` to select a profile in the same way as the CLI. Set
`Config(no_env=True)` to ignore these environment variables.

The configuration reader can also get the default parent ID:

```python
from nebius.aio.cli_config import Config

print(f"My default parent ID: {Config().parent_id}")
```

See the
[`Config` documentation](https://nebius.github.io/pysdk/nebius.aio.cli_config.Config.html)
for settings such as the file path, profile name, metrics, and environment
variable names. `Config(metrics=...)` receives configuration and authentication
callbacks. `Config(auth_metrics=...)` receives only authentication callbacks
for credentials from the configuration reader. If you attach callbacks later
with `Config.set_metrics(...)`, the reader replays the last configuration-load
event.

##### Initialize with the private key file

You can authorize with a service account and its private key. In the following
example, specify the service account ID and the public key ID for your private
key. Also change the path to your `private_key.pem` file.

```python
from nebius.sdk import SDK
from nebius.base.service_account.pk_file import Reader as PKReader  # [1]

sdk = SDK(
    credentials=PKReader(
        filename="location/of/your/private_key.pem",
        public_key_id="public-key-id",
        service_account_id="your-service-account-id",
    ),
    user_agent_prefix="example-application/1.0",
)
# or without importing PKReader
sdk = SDK(
    service_account_private_key_file_name="location/of/your/private_key.pem",
    service_account_public_key_id="public-key-id",
    service_account_id="your-service-account-id",
    user_agent_prefix="example-application/1.0",
)
```
[[1](https://nebius.github.io/pysdk/nebius.base.service_account.pk_file.Reader.html)]

##### Initialize with a credentials file

You can also use one credentials file that contains the private key and all
required IDs:

```python
from nebius.sdk import SDK
from nebius.base.service_account.credentials_file import Reader as CredentialsReader  # [1]

sdk = SDK(
    credentials=CredentialsReader(
        filename="location/of/your/credentials.json",
    ),
    user_agent_prefix="example-application/1.0",
)
# or without importing CredentialsReader
sdk = SDK(
    credentials_file_name="location/of/your/credentials.json",
    user_agent_prefix="example-application/1.0",
)
```
[[1](https://nebius.github.io/pysdk/nebius.base.service_account.credentials_file.Reader.html)]

#### Test the SDK

Use
[`SDK.whoami`](https://nebius.github.io/pysdk/nebius.sdk.SDK.html#whoami) to
test your credentials and connection. This method returns basic information
about the authenticated profile:

```python
import asyncio

async def my_call():
    async with sdk:
        print(await sdk.whoami())

asyncio.run(my_call())
```

Close the SDK to stop and collect all coroutines and tasks. Use `async with`,
as in the previous example, or call `sdk.close()` explicitly:

```python
import asyncio

async def my_call():
    try:
        print(await sdk.whoami())
        # Other calls to SDK
    finally:
        await sdk.close()

asyncio.run(my_call())
```

Each SDK instance uses one internal asyncio event loop. By default, the SDK
starts a daemon thread for this loop. It also creates a private daemon executor
with a maximum of two workers. The executor starts workers only when the SDK
submits work.

The internal loop runs requests, authentication, token renewal, streaming,
asynchronous metric callbacks, and cleanup. You can await an SDK awaitable from
any external asyncio loop. Reuse SDK instances. Do not create an SDK instance
for each request because each default instance owns a loop thread and an
executor.

Use an asynchronous context when possible. If no asynchronous event loop is
running, you can use the SDK synchronously:

```python
try:
    print(sdk.whoami().wait())
    # Other calls to SDK
finally:
    sdk.sync_close()
```

Do not call synchronous helpers from asynchronous code. Await the SDK handle
instead. The 0.5 migration section describes the other loop and thread rules.

If you do not close the SDK, unterminated tasks can cause errors.

#### Test token renewal

The SDK renews tokens when you use a service account, a credentials file, or
another renewable credential source. By default, it writes renewal errors to
the log. To receive these errors as request errors, give renewal options to the
request:


```python
import asyncio

from nebius.aio.token.renewable import (
    OPTION_RENEW_REQUEST_TIMEOUT,
    OPTION_RENEW_REQUIRED,
    OPTION_RENEW_SYNCHRONOUS,
)

async def my_call():
    try:
        await sdk.whoami(
            auth_options={
                OPTION_RENEW_REQUIRED: "true",
                OPTION_RENEW_SYNCHRONOUS: "true",
                OPTION_RENEW_REQUEST_TIMEOUT: ".9",
            }
        )
    except Exception as err:
        print(f"Token renewal failed: {err=}")
    finally:
        await sdk.close()

asyncio.run(my_call())
```

You can give these options to any request.

The `auth_timeout` value limits the total time for authentication and renewal.
The default value is 15 minutes. You can change it for one call. For example,
use `await sdk.whoami(auth_timeout=600.0)`. If you set `auth_timeout=None`,
authentication can wait indefinitely when renewal cannot finish.

#### Call a method

After you initialize and test the SDK, you can call service methods. The
following sections assume that the
[SDK](https://nebius.github.io/pysdk/nebius.sdk.SDK.html) is in the `sdk`
variable. The examples do not show how to close the SDK.

All service API classes are in submodules of `nebius.api.nebius`. See the
[API reference](https://nebius.github.io/pysdk/apiReference.html). The
`nebius.api.nebius` package also contains direct message, enum, reflection,
and service-client classes.

This example gets a bucket from Object Storage by its ID:

```python
import asyncio

from nebius.api.nebius.storage.v1 import GetBucketRequest
from nebius.api.nebius.storage.v1 import BucketServiceClient

async def my_call():
    service = BucketServiceClient(sdk)
    return await service.get(GetBucketRequest(
        id="some-bucket-id",
    ))

asyncio.run(my_call())
```

The following example makes the same call synchronously:

```python
import asyncio

from nebius.api.nebius.storage.v1 import BucketServiceClient, GetBucketRequest

service = BucketServiceClient(sdk)
result = service.get(GetBucketRequest(
    id="some-bucket-id",
)).wait()
```

##### Parent ID

Some requests contain `parent_id`. The SDK sets this field automatically in
these cases:

- The `parent_id` field is empty for the `list` and `get_by_name` methods.
- The `metadata.parent_id` field is empty for all methods except `update`.

The SDK sets `parent_id` only if initialization supplied a value. The value can
come from the
[CLI `Config`](https://nebius.github.io/pysdk/nebius.aio.cli_config.Config.html)
or the
[`SDK.parent_id`](https://nebius.github.io/pysdk/nebius.sdk.SDK.html)
attribute. To use the CLI configuration without its parent ID, set
`no_parent_id=True`.

##### Operations

Many core methods return a
[`nebius.aio.Operation`](https://nebius.github.io/pysdk/nebius.aio.operation.Operation.html)
object for a long-running asynchronous operation. For example,
`BucketServiceClient.create` returns this object. The wrapper supplies methods
that monitor the operation. You can await the wrapper until the operation is
complete.

The following example uses asynchronous calls:

```python
from nebius.api.nebius.storage.v1 import BucketServiceClient, CreateBucketRequest

service = BucketServiceClient(sdk)
operation = await service.create(CreateBucketRequest(
    # Set the required fields.
))
await operation.wait()
print(f"New bucket ID: {operation.resource_id}")
```

The following example uses synchronous calls:

```python
from nebius.api.nebius.storage.v1 import BucketServiceClient, CreateBucketRequest

service = BucketServiceClient(sdk)
operation = service.create(CreateBucketRequest(
    # Set the required fields.
)).wait()
operation.wait_sync()
print(f"New bucket ID: {operation.resource_id}")
```

##### Progress tracker

Some operations supply a progress tracker with an estimated completion time,
completed-work value, and step details. Use
[`Operation.progress_tracker`](https://nebius.github.io/pysdk/nebius.aio.operation.Operation.html#progress_tracker)
to get it. This method returns `None` for v1alpha1 operations and operations
without progress details.

This example polls an operation and shows progress on one line:

```python
from asyncio import sleep
from datetime import datetime
from nebius.base.protos.well_known import local_timezone

while not operation.done():
    await operation.update()
    tracker = operation.progress_tracker()
    parts = [f"waiting for operation {operation.id} to complete:"]

    if tracker:
        work = tracker.work_fraction()
        if work is not None:
            parts.append(f"{work:.0%}")

        desc = tracker.description()
        if desc:
            parts.append(desc)

        started = tracker.started_at()
        if started is not None:
            elapsed = datetime.now(local_timezone) - started
            parts.append(f"{elapsed}")

        eta = tracker.estimated_finished_at()
        if eta is not None:
            parts.append(f"eta {eta}")

    print(" ".join(parts), end="\r", flush=True)
    await sleep(1)

print()
```

##### Operations service

Use an
[`OperationServiceClient`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.OperationServiceClient.html)
to get or list operations.

The
[`OperationServiceClient`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.OperationServiceClient.html)
is in
[`nebius.api.nebius.common.v1`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.html),
but it does not work in the same way as other services. Get the applicable
operation service from the source service. Call
[`service.operation_service()`](https://nebius.github.io/pysdk/nebius.aio.client.ClientWithOperations.html#operation_service).

This asynchronous example lists operations and gets one operation:
```python
from nebius.api.nebius.common.v1 import GetOperationRequest, ListOperationsRequest
from nebius.api.nebius.storage.v1 import BucketServiceClient

service = BucketServiceClient(sdk)
op_service = service.operation_service()

resp = await op_service.list(ListOperationsRequest(resource_id="your-bucket-id"))
op_id = resp.operations[0].id  # These elements are not Operation wrappers.
real_operation = await op_service.get(GetOperationRequest(id=op_id))

# The get method returns an Operation wrapper that you can await.
await real_operation.wait()
```

> **Note:** Only
> [`get`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.OperationServiceClient.html#get)
> returns a complete
> [`Operation`](https://nebius.github.io/pysdk/nebius.aio.operation.Operation.html)
> wrapper. Methods such as
> [`list`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.OperationServiceClient.html#list)
> and Compute
> [`list_operations_by_parent`](https://nebius.github.io/pysdk/nebius.api.nebius.compute.v1.DiskServiceClient.html#list_operations_by_parent)
> return an internal
> [`Operation`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.Operation.html)
> representation. You cannot await or poll this internal representation as an
> `Operation` wrapper.

##### Timeouts and retries

SDK requests have an internal retry layer and two request timeouts:

- Overall request timeout: Limits SDK-loop dispatch plus native request and
  retry work. Authentication uses its independent clock and does not consume
  this request-only budget.
- Per-retry timeout: Limits each retry attempt. If you do not set this value,
  it uses the overall request timeout.

By default, the overall request timeout is 60 seconds and the per-retry timeout
is 20 seconds (`60 / 3`). To disable timeouts for one call, set
`timeout=None`. For example, use `service.get(req, timeout=None)`. Without a
timeout, a request can wait indefinitely.

Network errors, resource exhaustion, quota errors, and service errors can stop
a retry loop. Timeouts are not the only possible cause. If an attempt can wait
on a slow resource, set a shorter per-retry timeout. The attempt then fails
sooner, and the retry loop can continue or return the error.

Operations add an operation-level timeout. This timeout limits the complete
operation lifecycle. The names of timeouts for each operation update request
have the `poll_` prefix.

###### Authentication timeout (auth_timeout)

The independent `auth_timeout` value limits SDK-loop dispatch and the complete
authentication flow. Authentication includes token acquisition or renewal and
can retry before and during the RPC. This timeout includes internal
authentication retries and the request inside the authentication loop.

- Default: 15 minutes (900 seconds)
- Scope: The authentication loop and its request
- Behavior:
    - The credential provider controls authentication retries. For example, it
      can retry after a temporary network error or an `UNAUTHENTICATED`
      response. `auth_timeout` limits all these retries.
    - For synchronous calls such as `.wait()`, `auth_timeout` limits the total
      wait time.
- Per-call change: Give `auth_timeout` to a service method. For example:

```python
response = await service.get(req, auth_timeout=300.0)  # 5 minutes
```

- To disable the authentication deadline, set `auth_timeout=None`.
  Authentication can then wait indefinitely if it does not succeed.

Notes:

- `auth_timeout` starts when the request is submitted, before it waits for the
  SDK loop. It limits the authorized request `timeout`, which limits each
  `per_retry_timeout`.
- The request clock charges SDK-loop queueing once, pauses while credentials
  are acquired or renewed, and resumes for native RPC and retry work. Streaming
  calls use the same independent-clock rule during initial authentication.
- Individual token-exchange attempts can have shorter deadlines.
  `auth_timeout` limits their total time.

##### Keepalive and metrics

By default, the SDK enables gRPC keepalive settings that are compatible with
the Nebius SDK for Go. You can change them with the
`NEBIUS_GRPC_KEEPALIVE_TIME`, `NEBIUS_GRPC_KEEPALIVE_TIMEOUT`, and
`NEBIUS_GRPC_KEEPALIVE_PERMIT_WITHOUT_STREAM` environment variables. To
disable SDK keepalive, set `keepalive=False`. You can also give explicit
options:

```python
from nebius.aio.keepalive import KeepaliveOptions
from nebius.sdk import SDK

sdk = SDK(
    keepalive=KeepaliveOptions(time_ms=20_000, timeout_ms=10_000),
    user_agent_prefix="example-application/1.0",
)
```

Metrics are optional and use callbacks. Give `metrics` to `SDK` or `Config` to
receive configuration and authentication events. Give `auth_metrics` to
receive only authentication events. Callback names can use snake_case or
camelCase.

Synchronous callbacks work in all contexts. The SDK schedules asynchronous
callbacks when an event loop is running. It waits for them when synchronous
code emits an event. `callback_timeout_seconds` limits awaitable callback
results. Its default value is 1 second, and the SDK adjusts invalid values to
its limits.

This timeout uses cooperative cancellation. A callback can still block the
event loop if it ignores cancellation, blocks execution, or does not await
frequently. Keep metric callbacks fast and nonblocking.

```python
from nebius.aio.cli_config import Config
from nebius.aio.metrics import Metrics
from nebius.sdk import SDK


def config_load(metric):
    print(metric.source, metric.result, metric.duration_seconds)


def token_acquire(metric):
    print(metric.provider, metric.result, metric.attempt)


sdk = SDK(
    config_reader=Config(
        metrics=Metrics(
            config_load=config_load,
            token_acquire=token_acquire,
            callback_timeout_seconds=0.5,
        )
    ),
    user_agent_prefix="example-application/1.0",
)
```

The SDK ignores metric callback failures. Thus, instrumentation does not affect
SDK requests. The SDK emits token-lifetime metrics only for expiration
timestamps that contain time-zone information.


##### Retrieve additional metadata

You can get request information in addition to the result. This information
can help the Nebius support team investigate a problem. Service methods return
[`Request`](https://nebius.github.io/pysdk/nebius.aio.request.Request.html)
objects instead of basic coroutines. A `Request` object supplies information
about its request.

The following example gets the request ID and trace ID. Errors usually contain
these IDs, but you can also get them for a successful request:

```python
request = service.get(req)  # Do not await the request yet.

# Await these values in any order or at the same time.
response = await request
request_id = await request.request_id()
trace_id = await request.trace_id()

log.info(f"Server answered: {response}; Request ID: {request_id} and Trace ID: {trace_id}")
```

Use the synchronous methods in a synchronous context:

```python
request = service.get(req)  # Do not wait for the request yet.

# Call these methods in any order. The first call starts the request and waits.
response = request.wait()
request_id = request.request_id_sync()
trace_id = request.trace_id_sync()

log.info(f"Server answered: {response}; Request ID: {request_id} and Trace ID: {trace_id}")
```

##### Parse errors

A request can raise different exceptions. Server exceptions derive from
[`nebius.aio.service_error.RequestError`](https://nebius.github.io/pysdk/nebius.aio.service_error.RequestError.html).
This error contains the request status and available information from the
server.

Print `RequestError` to see the information as text. To get structured
information, read
[`nebius.aio.service_error.RequestStatusExtended`](https://nebius.github.io/pysdk/nebius.aio.service_error.RequestStatusExtended.html)
from `err.status`.

```python
from nebius.aio.service_error import RequestError

try:
    response = await service.get(req)
except RequestError as err:
    log.exception(f"Caught request error {err}")
```

Save the request ID and trace ID if you must contact Nebius support.

### Calling `update` methods on resources

For a resource `update` method, send one of these inputs:

- A manually constructed
  [`X-ResetMask`](https://nebius.github.io/pysdk/nebius.base.fieldmask.Mask.html).
- A complete resource specification that your code got and changed.

The following sections show both methods.

#### Sending a full specification

This example gets a bucket specification, doubles its size limit, and sends the
complete specification:

```python
from nebius.api.nebius.storage.v1 import UpdateBucketRequest

bucket = await service.get(req)
bucket.spec.max_size_bytes *= 2  # Example of the change
operation = await service.update(
    UpdateBucketRequest(
        metadata=bucket.metadata,
        spec=bucket.spec,
    ),
)
```

This operation checks the resource version. If another client changes the
resource concurrently, the server rejects one request. To omit the resource
version check, set `metadata.resource_version` to **0**:

```python
from nebius.api.nebius.storage.v1 import UpdateBucketRequest

bucket = await service.get(req)
bucket.spec.max_size_bytes *= 2  # Example of the change
bucket.metadata.resource_version = 0  # Skip the check and replace the resource.
operation = await service.update(
    UpdateBucketRequest(
        metadata=bucket.metadata,
        spec=bucket.spec,
    ),
)
```

This request **fully replaces** the bucket specification. It overwrites
changes from concurrent updates.


#### Updating with manually set `X-ResetMask`

You can replace values without first requesting the complete specification.
To set a value to its Protocol Buffers default, set `X-ResetMask` manually in
the metadata. The update does not overwrite an unset field or a default field
that is not in the mask.

Here is an example of resetting the limit on the bucket:

```python
from nebius.api.nebius.storage.v1 import UpdateBucketRequest
from nebius.api.nebius.common.v1 import ResourceMetadata
from nebius.base.metadata import Metadata

md = Metadata()
md["X-ResetMask"] = "spec.max_size_bytes"
operation = await service.update(
    UpdateBucketRequest(
        metadata=ResourceMetadata(
            id="some-bucket-id",  # Required to identify the resource
        )
    ),
    metadata=md,
)
```

This example resets only `max_size_bytes` and removes the bucket limit. It does
not unset or change other fields.

> **Note:** Nebius field masks have more detail than Google field masks. The
> two mask types are not compatible. See the Nebius API documentation for more
> information about masks.

> **Note:** Read the API documentation before you use manually set masks to
> change lists and maps.

### User-agent

Set `user_agent_prefix` in each SDK constructor. Use a value that identifies
your application and version, such as `example-application/1.0`. The SDK sends
this value to the server as part of the user-agent.

You can also add the `grpc.primary_user_agent` option to SDK `options` or
`address_options`. The SDK combines user-agent parts in approximately this
order. It omits parts that are not set:
```python
" ".join(
    [
        *primary_user_agents_from_options,
        *primary_user_agents_from_address_options,
        user_agent_prefix,
        pysdk_user_agent,
        grpc_user_agent,  # gRPC adds this value.
        *secondary_user_agents_from_options,
        *secondary_user_agents_from_address_options,
    ]
)
```

### Contributing

See the [contributing guidelines](CONTRIBUTING.md) to contribute.

### License

This project is licensed under the MIT License. See the [LICENSE](LICENSE) file for details.

Copyright (c) 2025 Nebius B.V.