Edit

Manage Machines with Viam's Machine Management API

The machine API allows you to connect to your machine from within a supported Viam SDK, retrieve status information, and send commands remotely.

The machine API is supported for use with the Viam Python SDK, the Viam Go SDK, and the Viam C++ SDK.

The machine API supports the following methods:

Method Name	Description
`GetOperations`	Get the list of operations currently running on the machine.
`GetMachineStatus`	Get status information about the machine.
`ResourceNames`	Get a list of all known resource names connected to this machine.
`CancelOperation`	Cancel the specified operation on the machine.
`BlockForOperation`	Blocks on the specified operation on the machine.
`FrameSystemConfig`	Get the configuration of the frame system of a given machine.
`TransformPose`	Transform a given source Pose from the original reference frame to a new destination reference frame.
`TransformPCD`	Transforms the pointcloud to the desired frame in the robot’s frame system.
`StopAll`	Cancel all current and outstanding operations for the machine and stop all actuators and movement.
`RestartModule`	Reload a module as if its config changed.
`Log`	Create a LogEntry object from the log to send to the RDK over gRPC.
`GetCloudMetadata`	Get app-related information about the robot.
`GetVersion`	Return version information about the machine.
`Options.with_api_key`	Create a `RobotClient.Options` using an API key as credentials.
`AtAddress`	Create a RobotClient that is connected to the machine at the provided address.
`WithChannel`	Create a RobotClient that is connected to a machine over the given channel.
`Refresh`	Manually refresh the underlying parts of this machine.
`Shutdown`	Shutdown shuts down the machine.
`Close`	Close the underlying connections and stop any periodic tasks across all constituent parts of the machine.

Establish a connection

To interact with the machine API with Viam’s SDKs, instantiate a RobotClient (gRPC client) and use that class for all interactions.

To find the API key, API key ID, and machine address, go to Viam app, select the machine you wish to connect to, and go to the CONNECT tab. Toggle Include API key, and then copy and paste the API key ID and the API key into your environment variables or directly into the code:

Python
Go

import asyncio

from viam.rpc.dial import DialOptions, Credentials
from viam.robot.client import RobotClient


async def connect() -> RobotClient:
    opts = RobotClient.Options.with_api_key(
            # Replace "<API-KEY>" (including brackets) with your API key
            api_key='<API-KEY>',
            # Replace "<API-KEY-ID>" (including brackets) with your
            # API key ID
            api_key_id='<API-KEY-ID>'
        )
    return await RobotClient.at_address(
      address='ADDRESS FROM CONNECT TAB OF VIAM APP',
      options=opts
    )


async def main():
    # Make a RobotClient
    machine = await connect()
    print('Resources:')
    print(machine.resource_names)
    await machine.close()

if __name__ == '__main__':
    asyncio.run(main())

You can use this code to connect to your machine and instantiate a RobotClient that you can then use with the robot API. As an example, this code uses the instantiated RobotClient to return the resources currently configured. Remember to always close the connection (using close()) when done.

package main

import (
  "context"

  "go.viam.com/rdk/logging"
  "go.viam.com/rdk/robot/client"
  "go.viam.com/rdk/utils"
)

func main() {
  logger := logging.NewLogger("client")
  machine, err := client.New(
      context.Background(),
      "ADDRESS FROM THE VIAM APP",
      logger,
      client.WithDialOptions(utils.WithEntityCredentials(
      // Replace "<API-KEY-ID>" (including brackets) with your machine's
      // API Key ID
      "<API-KEY-ID>",
      utils.Credentials{
          Type:    utils.CredentialsTypeAPIKey,
        // Replace "<API-KEY>" (including brackets) with your machine's API key
        Payload: "<API-KEY>",
    })),
  )
  if err != nil {
      logger.Fatal(err)
  }

  defer machine.Close(context.Background())
  logger.Info("Resources:")
  logger.Info(machine.ResourceNames())
}

You can use this code to connect to your machine and instantiate a robot client that you can then use with the Robot API. As an example, this code uses the instantiated robot client to return the configured resources. Remember to always close the connection (using Close()) when done.

Once you have instantiated the robot client, you can run any of the robot API methods against the robot object to communicate with your machine.

Configure a timeout

Because the robot API needs to be able to reliably connect to a deployed machine even over a weak or intermittent network, it supports a configurable timeout for connection and command execution.

To configure a timeout when using the robot API:

Python
Go

Add a timeout to DialOptions inside of Options:

# Add the timeout argument to DialOptions:
opts = RobotClient.Options(
    dial_options=DialOptions(timeout=10)).with_api_key(
      ...
    )

The example above shows a timeout of 10 seconds configured.

Import the time package, then add a timeout to your context using WithTimeout:

// Import the time package in addition to the other imports:
import (
  ...
  "time"
  ...
)

// Edit your main() to configure a timeoutContext, then pass this context to the dial invocation:
func main() {
  ctx := context.Background()
  timeoutContext, cancel := context.WithTimeout(ctx, 10*time.Second)
  defer cancel()
  machine, err := client.New(
      timeoutContext,
      "ADDRESS FROM THE VIAM APP",
      logger,
      client.WithDialOptions(rpc.WithEntityCredentials(
      // Replace "<API-KEY-ID>" (including brackets) with your machine's
      // API key ID
      "<API-KEY-ID>",
      rpc.Credentials{
          Type:    rpc.CredentialsTypeAPIKey,
        // Replace "<API-KEY>" (including brackets) with your machine's API key
        Payload: "<API-KEY>",
    })),
  )
}

The example above shows a timeout of 10 seconds configured.

API

GetOperations

Get the list of operations currently running on the machine.

Python

Parameters:

None.

Returns:

(List[viam.proto.robot.Operation]): The list of operations currently running on a given machine.

Example:

operations = await machine.get_operations()

For more information, see the Python SDK Docs.

GetMachineStatus

Get status information about the machine.

Python

Parameters:

None.

Returns:

(viam.proto.robot.GetMachineStatusResponse): current status of the resources (List[ResourceStatus]) and config of the machine.

Example:

machine_status = await machine.get_machine_status()
resource_statuses = machine_status.resources
config_status = machine_status.config

For more information, see the Python SDK Docs.

ResourceNames

Get a list of all known resource names connected to this machine.

Parameters:

None.

Returns:

([]resource.Name): A list of all known resource names.

Example:

resource_names := machine.ResourceNames()

For more information, see the Go SDK Docs.

CancelOperation

Cancel the specified operation on the machine.

Python

Parameters:

id (str) (required): ID of operation to cancel.

Returns:

None.

Example:

await machine.cancel_operation("INSERT OPERATION ID")

For more information, see the Python SDK Docs.

BlockForOperation

Blocks on the specified operation on the machine. This function will only return when the specific operation has finished or has been cancelled.

Python

Parameters:

id (str) (required): ID of operation to block on.

Returns:

None.

Example:

await machine.block_for_operation("INSERT OPERATION ID")

For more information, see the Python SDK Docs.

FrameSystemConfig

Get the configuration of the frame system of a given machine.

Python
Go

Parameters:

additional_transforms (List[viam.proto.common.Transform]) (optional): Any additional transforms.

Returns:

(List[viam.proto.robot.FrameSystemConfig]): The configuration of a given machine’s frame system.

Example:

# Get a list of each of the reference frames configured on the machine.
frame_system = await machine.get_frame_system_config()
print(f"frame system configuration: {frame_system}")

For more information, see the Python SDK Docs.

Parameters:

ctx (Context): A Context carries a deadline, a cancellation signal, and other values across API boundaries.

Returns:

(*framesystem.Config): The configuration of the given machine’s frame system.
(error): An error, if one occurred.

Example:

// Print the frame system configuration
frameSystem, err := machine.FrameSystemConfig(context.Background())
fmt.Println(frameSystem)

For more information, see the Go SDK Docs.

TransformPose

Transform a given source Pose from the original reference frame to a new destination reference frame.

Python
Go

Parameters:

query (viam.proto.common.PoseInFrame) (required): The pose that should be transformed.
destination (str) (required): The name of the reference frame to transform the given pose to.
additional_transforms (List[viam.proto.common.Transform]) (optional): Any additional transforms.

Returns:

(viam.proto.common.PoseInFrame): The pose and the reference frame for the new destination.

Example:

from viam.proto.common import Pose, PoseInFrame

pose = Pose(
    x=1.0,    # X coordinate in mm
    y=2.0,    # Y coordinate in mm
    z=3.0,    # Z coordinate in mm
    o_x=0.0,  # X component of orientation vector
    o_y=0.0,  # Y component of orientation vector
    o_z=0.0,  # Z component of orientation vector
    theta=0.0 # Orientation angle in degrees
)

pose_in_frame = PoseInFrame(
    reference_frame="world",
    pose=pose
)

transformed_pose = await machine.transform_pose(pose_in_frame, "world")

For more information, see the Python SDK Docs.

Parameters:

ctx (Context): A Context carries a deadline, a cancellation signal, and other values across API boundaries.
pose (*referenceframe.PoseInFrame): The pose that should be transformed.
dst (string): The name of the reference pose to transform the given pose to.
additionalTransforms ([]*referenceframe.LinkInFrame): Any additional transforms.

Returns:

(*referenceframe.PoseInFrame): Transformed pose in frame.
(error): An error, if one occurred.

Example:

import (
  "go.viam.com/rdk/referenceframe"
  "go.viam.com/rdk/spatialmath"
)

baseOrigin := referenceframe.NewPoseInFrame("test-base", spatialmath.NewZeroPose())
movementSensorToBase, err := machine.TransformPose(context.Background(), baseOrigin, "my-movement-sensor", nil)

For more information, see the Go SDK Docs.

TransformPCD

Transforms the pointcloud to the desired frame in the robot’s frame system. Do not move the robot between the generation of the initial pointcloud and the receipt of the transformed pointcloud, as doing so will make the transformations inaccurate.

Parameters:

ctx (Context): A Context carries a deadline, a cancellation signal, and other values across API boundaries.
srcpc (pointcloud.PointCloud): The source PointCloud to transform.
srcName (string): The name of the source point cloud to transform.
dstName (string): The name of the destination point cloud.

Returns:

(pointcloud.PointCloud): The transformed PointCloud.
(error): An error, if one occurred.

For more information, see the Go SDK Docs.

StopAll

Cancel all current and outstanding operations for the machine and stop all actuators and movement.

Python
Go

Parameters:

extra (Mapping[str, Any]) (required): Extra options to pass to the underlying RPC call.

Returns:

None.

Example:

# Cancel all current and outstanding operations for the machine and stop all actuators and movement.
await machine.stop_all()

For more information, see the Python SDK Docs.

Parameters:

ctx (Context): A Context carries a deadline, a cancellation signal, and other values across API boundaries.
extra (map[string]interface{}): Extra options to pass to the underlying RPC call.

Returns:

(error): An error, if one occurred.

Example:

// Cancel all current and outstanding operations for the machine and stop all actuators and movement.
err := machine.StopAll(context.Background(), nil)

For more information, see the Go SDK Docs.

RestartModule

Reload a module as if its config changed.

Parameters:

ctx (Context): A Context carries a deadline, a cancellation signal, and other values across API boundaries.
req (RestartModuleRequest)

Returns:

(error): An error, if one occurred.

For more information, see the Go SDK Docs.

Log

Create a LogEntry object from the log to send to the RDK over gRPC.

Python

Parameters:

name (str) (required): The logger’s name.
level (str) (required): The level of the log.
time (datetime.datetime) (required): The log creation time.
message (str) (required): The log message.
stack (str) (required): The stack information of the log.

Returns:

None.

For more information, see the Python SDK Docs.

GetCloudMetadata

Get app-related information about the robot.

Parameters:

None.

Returns:

(viam.proto.robot.GetCloudMetadataResponse): App-related metadata.

Example:

metadata = await machine.get_cloud_metadata()
print(metadata.machine_id)
print(metadata.machine_part_id)
print(metadata.primary_org_id)
print(metadata.location_id)

For more information, see the Python SDK Docs.

Parameters:

ctx (Context): A Context carries a deadline, a cancellation signal, and other values across API boundaries.

Returns:

(cloud.Metadata): App-related metadata containing the primary organization ID, location ID, and robot part ID for a machine running on the Viam app.
(error): An error, if one occurred.

Example:

metadata, err := machine.CloudMetadata(context.Background())
primary_org_id := metadata.PrimaryOrgID
location_id := metadata.LocationID
machine_id := metadata.MachineID
machine_part_id := metadata.MachinePartID

For more information, see the Go SDK Docs.

Parameters:

None.

Returns:

Future<CloudMetadata>

Example:

var metadata = await machine.getCloudMetadata();

For more information, see the Flutter SDK Docs.

GetVersion

Return version information about the machine.

Python
Go

Parameters:

None.

Returns:

(viam.proto.robot.GetVersionResponse): Machine version related information.

Example:

result = await machine.get_version()
print(result.platform)
print(result.version)
print(result.api_version)

For more information, see the Python SDK Docs.

Parameters:

ctx (Context): A Context carries a deadline, a cancellation signal, and other values across API boundaries.

Returns:

(VersionResponse)
(error): An error, if one occurred.

For more information, see the Go SDK Docs.

Options.with_api_key

Create a RobotClient.Options using an API key as credentials. Pass these options to AtAddress.

Python

Parameters:

api_key (str) (required): your API key.
api_key_id (str) (required): your API key ID. Must be a valid UUID.

Returns:

(typing_extensions.Self): the RobotClient.Options.

Raises:

(ValueError): Raised if the api_key_id is not a valid UUID.

Example:

# Replace "<API-KEY>" (including brackets) with your machine's API key
api_key = '<API-KEY>'
# Replace "<API-KEY-ID>" (including brackets) with your machine's API key ID
api_key_id = '<API-KEY-ID>'

opts = RobotClient.Options.with_api_key(api_key, api_key_id)

machine = await RobotClient.at_address('<ADDRESS-FROM-THE-VIAM-APP>', opts)

For more information, see the Python SDK Docs.

AtAddress

Create a RobotClient that is connected to the machine at the provided address.

Python
Flutter

Parameters:

address (str) (required): Address of the machine (IP address, URL, etc.).
options (Options) (required): Options for connecting and refreshing.

Returns:

(typing_extensions.Self)

Example:

async def connect():

    opts = RobotClient.Options.with_api_key(
        # Replace "<API-KEY>" (including brackets) with your machine's API key
        api_key='<API-KEY>',
        # Replace "<API-KEY-ID>" (including brackets) with your machine's API key ID
        api_key_id='<API-KEY-ID>'
    )
    return await RobotClient.at_address('ADDRESS FROM THE VIAM APP', opts)


async def main():
    # Make a RobotClient
    machine = await connect()

For more information, see the Python SDK Docs.

Parameters:

url String (required)
options RobotClientOptions (required)

Returns:

Future<RobotClient>

Example:

// Example usage; see your machine's CONNECT tab for your machine's address and API key.

Future<void> connectToViam() async {
  const host = '<YOUR ROBOT ADDRESS>.viam.cloud';
  // Replace "<API-KEY-ID>" (including brackets) with your machine's API key ID
  const apiKeyID = '<API-KEY-ID>';
  // Replace "<API-KEY>" (including brackets) with your machine's API key
  const apiKey = '<API-KEY>';

  final machine = await RobotClient.atAddress(
    host,
    RobotClientOptions.withApiKey(apiKeyID, apiKey),
  );
}

For more information, see the Flutter SDK Docs.

WithChannel

Create a RobotClient that is connected to a machine over the given channel. Any machines created using this method will NOT automatically close the channel upon exit.

Python

Parameters:

channel (grpclib.client.Channel | viam.rpc.dial.ViamChannel) (required): The channel that is connected to a machine, obtained by viam.rpc.dial.
options (Options) (required): Options for refreshing. Any connection options will be ignored.

Returns:

(typing_extensions.Self)

Example:

from viam.robot.client import RobotClient
from viam.rpc.dial import DialOptions, dial


async def connect_with_channel() -> RobotClient:
    async with await dial('ADDRESS', DialOptions()) as channel:
        return await RobotClient.with_channel(channel, RobotClient.Options())

machine = await connect_with_channel()

For more information, see the Python SDK Docs.

Refresh

Manually refresh the underlying parts of this machine.

Python
Flutter

Parameters:

None.

Returns:

None.

Example:

await machine.refresh()

For more information, see the Python SDK Docs.

Parameters:

None.

Returns:

Future<void>

Example:

await machine.refresh();

For more information, see the Flutter SDK Docs.

Shutdown

Shutdown shuts down the machine. Supported by viam-micro-server.

Python
Go

Parameters:

None.

Returns:

None.

Raises:

(GRPCError): Raised with DeadlineExceeded status if shutdown request times out, or if the machine server shuts down before having a chance to send a response. Raised with status Unavailable if server is unavailable, or if machine server is in the process of shutting down when response is ready.

Example:

await machine.shutdown()

For more information, see the Python SDK Docs.

Parameters:

ctx (Context): A Context carries a deadline, a cancellation signal, and other values across API boundaries.

Returns:

(error): An error, if one occurred.

Example:

// Shut down the robot.
err := machine.Shutdown(context.Background())

For more information, see the Go SDK Docs.

Close

Close the underlying connections and stop any periodic tasks across all constituent parts of the machine.

Parameters:

None.

Returns:

None.

Example:

await machine.close()

For more information, see the Python SDK Docs.

Parameters:

ctx (Context): A Context carries a deadline, a cancellation signal, and other values across API boundaries.

Returns:

(error): An error, if one occurred.

Example:

// Cleanly close the underlying connections and stop any periodic tasks,
err := machine.Close(context.Background())

For more information, see the Go SDK Docs.

Parameters:

None.

Returns:

Future<void>

Example:

await machine.close();

For more information, see the Flutter SDK Docs.

Was this page helpful?

Glad to hear it! If you have any other feedback please let us know:

We're sorry about that. To help us improve, please tell us what we can do better:

Thank you!