Nebius Python® SDK

The Nebius Python® SDK is a client library for Nebius AI Cloud services. It uses gRPC and supports all APIs in the Nebius API repository. 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.

See the API reference for all services and methods.

pip install nebius

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

pip install ./path/to/your/pysdk

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:

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:

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.

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.

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.

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.

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:

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.

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:

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:

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

Working examples are in src/nebius/examples. From the repository root, run an example as follows:

NEBIUS_IAM_TOKEN=$(nebius iam get-access-token) PYTHONPATH=src \
python -m nebius.examples.basic your-project-id

from nebius.sdk import SDK

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

This code initializes the SDK 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.

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

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, 2]

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

If you configured the Nebius AI Cloud CLI, you can initialize the SDK with its CLI configuration:

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:

from nebius.aio.cli_config import Config

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

See the Config documentation 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.

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.

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]

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

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]

Use SDK.whoami to test your credentials and connection. This method returns basic information about the authenticated profile:

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:

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:

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.

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:

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.

After you initialize and test the SDK, you can call service methods. The following sections assume that the SDK 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. 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:

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:

import asyncio

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

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

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 or the SDK.parent_id attribute. To use the CLI configuration without its parent ID, set no_parent_id=True.

Many core methods return a nebius.aio.Operation 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:

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:

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}")

Some operations supply a progress tracker with an estimated completion time, completed-work value, and step details. Use Operation.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:

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()

Use an OperationServiceClient to get or list operations.

The OperationServiceClient is in nebius.api.nebius.common.v1, 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().

This asynchronous example lists operations and gets one operation:

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 returns a complete Operation wrapper. Methods such as list and Compute list_operations_by_parent return an internal Operation representation. You cannot await or poll this internal representation as an Operation wrapper.

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.

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:

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.

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:

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.

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.

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 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:

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:

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}")

A request can raise different exceptions. Server exceptions derive from nebius.aio.service_error.RequestError. 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 from err.status.

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.

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

• A manually constructed X-ResetMask.
• A complete resource specification that your code got and changed.

The following sections show both methods.

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

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:

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.

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:

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.

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:

" ".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,
    ]
)

See the contributing guidelines to contribute.

This project is licensed under the MIT License. See the LICENSE file for details.

Copyright (c) 2025 Nebius B.V.