AsyncClient

The AsyncClient offered through the Omniverse Services framework exposes an interface allowing communication between distant endpoints, making it possible to execute queries hosted on remote servers, where dedicated features may be hosted in order to implement dedicated workflows.

This AsyncClient builds its interface on top of the OpenAPI specifications, which allows it to express typical HTTP server methods in a straightforward way, by facilitating the documentation of server endpoints defined by Services.

Example

To provide an illustrative example of the AsyncClient, let’s reuse the Viewport Capture Service sample already shared in another section of the documentation, in order to provide additional usage information on how the remote-hosted service can be queried.

An asynchronous client can be included in an Omniverse extension by first declaring the omni.services.client module within the extension’s TOML definition file, then importing the Python components and formatting the client along the declaration of the Service’s definition. Using the Viewport Capture example, we can observe that the Service exposes a single POST endpoint, which is tasked with receiving incoming requests responsible for capturing a screenshot of the viewport as it appears in the application’s window:

# [...]

# Using the `@router` annotation, we'll tag our `capture` function handler to document the responses and path of the
# API, once again using the OpenAPI specification format.
@router.post(
    path="/capture",
    summary="Capture a given USD stage",
    description="Capture a given USD stage as an image.",
    response_model=ViewportCaptureResponseModel,
)
async def capture(request: ViewportCaptureRequestModel,) -> ViewportCaptureResponseModel:
    # For now, let's just print incoming request to the log to confirm all components of our extension are properly
    # wired together:
    carb.log_warn(f"Received a request to capture an image of \"{request.usd_stage_path}\".")

    # Let's return a JSON response, indicating that the viewport capture operation failed to avoid misinterpreting the
    # current lack of image output as a failure:
    return ViewportCaptureResponseModel(
        success=False,
        captured_image_path=None,
        error_message="Image not yet captured.",
    )

To trigger the remote execution of this endpoint, for example in a situation where an integration requires triggering the viewport capture of a remote server, the following integration can be developed in order to trigger execution of the Service defined by this API:

from omni.services.client import AsyncClient
from omni.services.core import main
from omni.services.example.viewport.capture.services.capture import ViewportCaptureRequestModel

client = AsyncClient(uri="local://", app=main.get_app())
request = ViewportCaptureRequestModel(usd_stage_path="omniverse://path/to/usd/stage.usd")
response = await client.capture.post(**request.to_dict())

Note

Notice that in order to reach the endpoint defined by a Service, the AsyncClient emits a request against the name of the action defined by API. Using the preceding example, the capture API ends up being targeted by the client and scoped using the .post(...) HTTP verb, as the Service’s router defines the action according to this standard option in order to support supplying issuing queries against the API.

In a similar fashion, PUT requests can be issued by formatting client requests using the .put(...) endpoint, or methods implementing the DELETE keyword can be issued by addressing a request to the name of the service suffixed by the .delete(...) keyword, as is also the case for the GET or PATCH methods which can be implemented by Services.

Naming convention

The name of the API used in the AsyncClient to identify the endpoint to target is based on the path associated to the endpoint, as it appears in the OpenAPI specification generated for the Service. As Python syntax prohibits using special characters such as dashes (-) in property names, it is worth noting that an endpoint containing dashes to separate words in order to expose pretty URLs will be replaced by underscore characters (_) in the equivalent Python attribute name.

For example, considering an endpoint defined with the "/capture-application-screenshot" path:

# [...]

# Example of an endpoint definition using dash characters (`-`) to separate words, in order to make them
# human-friendly:
@router.get(
    path="/capture-application-screenshot",
    summary="Capture a screenshot of the application",
    description="Capture a screenshot of the application in its current state.",
    response_model=ApplicationCaptureResponseModel,
)
async def application_capture(request: ApplicationCaptureRequestModel,) ->ApplicationCaptureResponseModel:
    return ApplicationCaptureResponseModel(
        # [...]
    )

This endpoint could be called using the following naming convention on an AsyncClient, replacing the dash characters (-) in its path with underscore characters (_):

# [...]

# Example usage of an `AsyncClient` to call the remote endpoint defined above:
client = AsyncClient(uri="[...]")
response = await client.capture_application_screenshot.get(
    # [...]
)

Including additional Headers

In some circumstances, a Service may require providing additional HTTP Headers along with parameters expected by the endpoint. Such requirements may include authorization or authentication features to provide along with the request, in order to confirm the identity or permission associated to a request. In such cases, the __headers__ parameter can be emitted along with the AsyncClient, as illustrated in the following example:

from typing import Any, Dict

from omni.services.client import AsyncClient
from omni.services.core import main


async def get_farm_queue_settings(farm_queue_server_url: str) -> Dict[str, Any]:
    """
    Return the settings from the selected Omniverse Farm Queue.

    Args:
       farm_queue_server_url(str): Server URL of the selected Omniverse Farm Queue.

    Returns:
       Dict[str, Any]: Settings exposed by the selected Omniverse Farm Queue.

    """
    try:
        client = AsyncClient(uri=farm_queue_server_url)
        settings_response = await client.queue.settings.get(__headers__=await _get_headers())
        return settings_response["settings"]
    except:
        return {}

async def _get_headers(self) -> Dict[str, str]:
    """
    Return the headers to send for each request against the Omniverse Farm API.

    Args:
       None

    Returns:
       Dict[str, str]: The headers to send along with each request made against the Omniverse Farm API.

    """
    headers = {
        "Accept": "application/json",
        "Content-Type": "application/json",
        "Authorization": "Bearer <secret>", # NOTE: Provide the `<secret>` key to authorize against the target Farm Queue.
    }
    return headers