Manage Machines with Viam's Robot API

The robot API is the application programming interface that manages each of your machines running viam-server. Use the robot API to connect to your machine from within a supported Viam SDK, and send commands remotely.

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

Establish a connection

To interact with the robot 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 Code sample 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:

import asyncio

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


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
    robot = await connect()
    print('Resources:')
    print(robot.resource_names)
    await robot.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"
  "go.viam.com/utils/rpc"
)

func main() {
  logger := logging.NewLogger("client")
  robot, err := client.New(
      context.Background(),
      "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>",
    })),
  )
  if err != nil {
      logger.Fatal(err)
  }

  defer robot.Close(context.Background())
  logger.Info("Resources:")
  logger.Info(robot.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:

Add a timeout to DialOptions:

# Add the timeout argument to DialOptions:
dial_options = DialOptions(credentials=creds, timeout=10)

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

The robot API support the following selected methods. For the full list of methods, see the Viam Python SDK documentation or the Viam Go SDK documentation.

Method NameDescription
Options.with_api_keyCreate RobotClient connection options with an API key as credentials.
AtAddressCreate a RobotClient that is connected to the machine at the provided address.
WithChannelCreate a RobotClient that is connected to a machine over the given channel.
RefreshManually refresh the underlying parts of the machine.
GetOperationsGet the list of operations currently running on the machine.
CancelOperationCancel the specified operation on the machine.
BlockForOperationBlocks on the specified operation on the machine.
TransformPoseTransform a given source Pose from the original reference frame to a new destination reference frame.
DiscoverComponentsGet a list of discovered component configurations.
FrameSystemConfigGet the configuration of a machine’s frame system.
StatusGet the status of each of the resources on the machine.
CloseClose the connections and stop periodic tasks across the machine.
StopAllCancel all operations for the machine and stop its movement.
ResourceNamesGet a list of all the machine’s resources.
GetCloudMetadataReturns app-related information about the robot.

Options.with_api_key

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

Parameters:

  • api_key (str): An API key with access to the machine.
  • api_key_id (str): Your API key ID. Must be a valid UUID.

Returns:

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

robot = 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.

Parameters:

  • address (str): Address of the machine. For example, IP address or URL.
  • options (Options): Options for connecting and refreshing.

Returns:

  • RobotClient: The robot client.
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
    robot = await connect()

For more information, see the Python 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.

Parameters:

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

Returns:

  • RobotClient: The robot client.
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())

robot = await connect_with_channel()

For more information, see the Python SDK Docs.

Refresh

Manually refresh the underlying parts of this machine.

Parameters:

  • None

Returns:

  • None
await robot.refresh()

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.
err := robot.Refresh(ctx)

For more information, see the Go SDK Docs.

GetOperations

Get the list of operations currently running on the machine.

Parameters:

  • None

Returns:

operations = await robot.get_operations()

For more information, see the Python SDK Docs.

CancelOperation

Cancel the specified operation on the machine.

Parameters:

Returns:

  • None
await robot.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.

Parameters:

Returns:

  • None
await robot.block_for_operation("INSERT OPERATION ID")

For more information, see the Python SDK Docs.

TransformPose

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

Parameters:

Returns:

pose = await robot.transform_pose(PoseInFrame(), "origin")

For more information, see the Python SDK Docs.

Parameters:

Returns:

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

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

For more information, see the Go SDK Docs.

DiscoverComponents

Get a list of discovered component configurations.

Parameters:

Returns:

# Define a new discovery query.
q = robot.DiscoveryQuery(subtype=acme.API, model="some model")

# Define a list of discovery queries.
qs = [q]

# Get component configurations with these queries.
component_configs = await robot.discover_components(qs)

For more information, see the Python SDK Docs

Parameters:

Returns:

  • ([]resource.Discovery): The search query qs and the corresponding list of discovered component configurations as an interface called Results. Results may be comprised of primitives, a list of primitives, maps with string keys (or at least can be decomposed into one), or lists of the forementioned type of maps.
  • (error): An error, if one occurred.
// Define a new discovery query.
q := resource.NewDiscoveryQuery(acme.API, resource.Model{Name: "some model"})

// Define a list of discovery queries.
qs := []resource.DiscoverQuery{q}

// Get component configurations with these queries.
component_configs, err := robot.DiscoverComponents(ctx.Background(), qs)

For more information, see the Go SDK Docs.

Parameters:

Returns:

// Define a new discovery query.
const q = new proto.DiscoveryQuery(acme.API, resource.Model{Name: "some model"})

// Define an array of discovery queries.
let qs:  proto.DiscoveryQuery[] = [q]

// Get the array of discovered component configurations.
const componentConfigs = await robot.discoverComponents(queries);

For more information, see the Typescript SDK Docs.

FrameSystemConfig

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

Parameters:

Returns:

# Get a list of each of the reference frames configured on the machine.
frame_system = await robot.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:

For more information, see the Go SDK Docs.

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

Parameters:

Returns:

For more information, see the TypeScript SDK Docs.

// Get the frame system configuration
console.log("FrameSytemConfig:", await robot.frameSystemConfig());

Status

Get the status of the resources on the machine. You can provide a list of ResourceNames for which you want statuses. If no names are passed in, the status of every resource configured on the machine is returned.

Parameters:

Returns:

  • (List[str]): A list containing the status of each resource.

For more information, see the Python SDK Docs.

# Get the status of the resources on the machine.
statuses = await robot.get_status()

Parameters:

  • ctx (Context): A Context carries a deadline, a cancellation signal, and other values across API boundaries.
  • resourceNames ([]resource.Name): An optional list of ResourceNames for components you want the status of. If no names are passed in, all resource statuses are returned.

Returns:

For more information, see the Go SDK Docs.

// Get the status of the resources on the machine.
status, err = robot.Status(ctx.Background())

Parameters:

  • resourceNames (commonApi.ResourceName[]): An optional array of ResourceNames for components you want the status of. If no names are passed in, all resource statuses are returned.

Returns:

For more information, see the TypeScript SDK Docs.

// Get the status of the resources on the machine.
const status = await robot.getStatus();

Close

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

Parameters:

  • None

Returns:

  • None

For more information, see the Python SDK Docs.

# Cleanly close the underlying connections and stop any periodic tasks.
await robot.close()

Parameters:

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

Returns:

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

For more information, see the Go SDK Docs.

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

Parameters:

  • None

Returns:

  • None

For more information, see the TypeScript SDK Docs.

// Cleanly close the underlying connections and stop any periodic tasks
await robot.disconnect();

StopAll

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

Parameters:

Returns:

  • None

For more information, see the Python SDK Docs.

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

Parameters:

Returns:

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

For more information, see the Go SDK Docs.

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

Parameters:

  • None

Returns:

  • None

For more information, see the TypeScript SDK Docs.

// Cancel all current and outstanding operations for the machine and stop all actuators and movement.
await robot.stopAll();

ResourceNames

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

Parameters:

  • None

Returns:

For more information, see the Python SDK Docs

resource_names = robot.resource_names

Parameters:

  • None

Returns:

resource_names := robot.ResourceNames()

For more information, see the Go SDK Docs.

Parameters:

  • None

Returns:

// Get a list of all resources on the machine.
const resource_names = await robot.resourceNames();

For more information, see the Typescript SDK Docs.

GetCloudMetadata

Returns app-related information about the robot.

Parameters:

  • None.

Returns:

For more information, see the Python SDK Docs.

metadata = robot.get_cloud_metadata()
print(metadata.robot_part_id)
print(metadata.primary_org_id)
print(metadata.location_id)

Parameters:

  • None.

Returns:

  • cloud.Metadata: App-related metadata containing the primary org id, location id, and robot part id for a robot running on the Viam app.

For more information, see the Go SDK Docs.

metadata := robot.GetCloudMetadata()
fmt.Println(metadata.RobotPartID)
fmt.Println(metadata.PrimaryOrgID)
fmt.Println(metadata.LocationID)

Parameters:

  • None.

Returns:

  • CloudMetadata: App-related metadata containing the primary org id, location id, and robot part id for a robot running on the Viam app.

For more information, see the Typescript SDK Docs.

// Get the metadata of the machine.
const metadata = await robot.getCloudMetadata();
console.log(metadata);

For the full list of robot API methods, see the Viam Python SDK documentation or the RDK (the Viam Go SDK) documentation.

Have questions, or want to meet other people working on robots? Join our Community Discord.

If you notice any issues with the documentation, feel free to file an issue or edit this file.