Command execution

Note

This page builds upon the Service configuration discussed in Terminal & SFTP/SCP. No additional Service configuration is required for command execution.

Client operation

Single device, single command

The simplest use case for this feature is to execute a command on a device and retrieve the output:

# Assign a device from the service inventory to a variable
>>> csr1 = service.inventory['csr1']

# Execute a single command on that device
>>> response = csr1.exec("show ip int brief")

# Display the output of the command executed
>>> response.data | print

The Device.exec() call returns an object that is an instance of radkit_client.sync.exec.SingleExecResponse, which represents the response that is returned from the Service for this command on the device.

>>> csr1.exec("show ip int brie")
[PROCESSING] SingleExecResponse(device_name='localhost', command='pwd')
--------------  -----------------
client_id       radkit-user@example.com
service_id      xxxx-yyyy-zzzz
device          csr1
sudo            False
data            None
errors
--------------  -----------------

The response object starts with a status of PROCESSING, and will transition to either SUCCESS, FAILURE or PARTIAL_SUCCESS.

• SUCCESS if the command was run successfully (output is available);

• FAILURE if the processing of the request failed completely;

• PARTIAL_SUCCESS if multiple commands are performed, but only some of them succeeded.

The RADKit Client is an asynchronous environment: while the request is in PROCESSING state, the CLI is not blocked; you can get on with other work (including performing more requests). Once the response is received, the object simply changes state in the background.

If however you prefer to synchronously wait for the request to complete, use .wait() to pause execution of your script, or suspend the REPL prompt, until the request has completed. Once completed, the data attribute will contain the response data.

>>> response = csr1.exec("show ip int brie")

# Wait for the request to complete (status will move to SUCCESS)
>>> response.wait()
[PROCESSING] SingleExecResponse(device_name='csr1', command='show ip int brie')
--------------  --------------------------------------------------------------------------------
client_id       radkit-user@example.com
service_id      xxxx-yyyy-zzzz
device          csr1
device_uuid     ...
command         show ip int brie
data            (pending)
--------------  --------------------------------------------------------------------------------

>>> response
[SUCCESS] SingleExecResponse(device_name='csr1', command='show ip int brie')
--------------  --------------------------------------------------------------------------------
client_id       radkit-user@example.com
service_id      xxxx-yyyy-zzzz
device          csr1
device_uuid     ...
command         show ip int brie
data            csr1#show ip int brie\nInterface              IP-Address      OK? Metho...
--------------  --------------------------------------------------------------------------------

Note

The .wait() method can take an optional parameter that specifies how many seconds to wait for. After that time, if the request still hasn’t completed, a TimeoutError exception will be raised. Note that the request is not affected, and it will still eventually complete (successfully or unsuccessfully) as if the .wait() and the timeout had not happened.

>>> response = csr1.exec("ping 10.1.1.1 repeat 10")
>>> response.wait(5)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "<frozen radkit_client.sync.request>", line 1030, in wait
File "<frozen radkit_client.sync.request>", line 355, in wait
File "/Users/radkit-user/.pyenv/versions/3.9.12/lib/python3.9/concurrent/futures/_base.py", line 448, in result
    raise TimeoutError()
concurrent.futures._base.TimeoutError

This is different from the timeout parameter to .exec() explained below, which affects the actual command execution.

The request attribute contains the underlying SimpleRequest object that represents the RPC request that was sent to the service. This SimpleRequest object has attributes like sent_timestamp.

>>> response.request
[SUCCESS] SimpleRequest(status='SUCCESS', rpc_name='exec-commands')
--------------------  -----------------------------------------------------------
sent_timestamp        2025-03-21 14:15:07
request_type          Exec command: 'pwd' on 1ffbf28c-6181-4df1-9e29-7277019c7b6d
client_id             None
service_id            None
result                None
forwarder
e2ee_used             True
compression_used      zstd
h2_multiplexing_used  True
--------------------  -----------------------------------------------------------

Once the request has completed, the actual command output is found under the .data property of the SingleExecResponse object. This is a single multi-line string.

>>> response.data
'csr1#ping 10.1.1.1 repeat 10\nType escape sequence to abort.\nSending 10, 100-byte ICMP Echos to 10.1.1.1, timeout is 2 seconds:\n..........\nSuccess rate is 0 percent (0/10)\ncsr1#'

This output can be easily processed using standard Python commands, or with the Client’s Pipe commands:

# Print the output as lines
>>> response.data | print
csr1#ping 10.1.1.1 repeat 10
Type escape sequence to abort.
Sending 10, 100-byte ICMP Echos to 10.1.1.1, timeout is 2 seconds:
..........
Success rate is 0 percent (0/10)
csr1#

# Print lines that match a regex
>>> response.data | regrep("Success rate") | print
Success rate is 0 percent (0/10)

Elevated privileges with sudo

If the platform/device allows it (currently we only support this on devices of type Linux), you can also execute privileged commands by adding the sudo=True parameter to the exec command. It provides the configured Terminal password to the sudo command:

>>> service.inventory['utm-linux'].exec("whoami", sudo=True).wait().data | print
sudo -k -p xsuzedhood: -- whoami
xsuzedhood:
root
ubuntu2@ubuntu-Standard-PC-Q35-ICH9-2009:~$

The -k parameter is added in order to always ask for a password, and -p is added to use a custom, one-time password prompt that is different for each command (xsuzedhood in our example).

Running multiple commands .exec(["whoami", "ls"], sudo=True) will execute all the commands successively using sudo.

Note

Executing commands chained with a semicolon will only execute the first command with sudo, the second will be run as the logged-in user. For instance:

# exec("whoami; whoami", sudo=True)
# yields the output of: "sudo whoami; whoami"
root
user

Command timeout

The Device.exec() call accepts an optional parameter called timeout which tells the Service how long (in seconds) it should wait for a command to complete. If the command does not complete within the specified time, the terminal connection is reset by the Service, an error response is returned to the Client, and the request goes to FAILURE state:

>>> response = csr1.exec("ping 10.1.1.1 repeat 10", timeout=5).wait()
18:06:44.168Z ERROR | command execution failed [device_name='csr1' commands=['ping 10.1.1.1 repeat 10'] error='Device action failed: Performing action failed: Timeout exception while performing commands']

>>> response
[FAILURE] SingleExecResponse(device_name='csr1', command='ping 10.1.1.1 repeat 10')
--------------  -------------------------------------------------------------------------------------------
client_id       radkit-user@example.com
service_id      xxxx-yyyy-zzzz
device          csr1
device_uuid     ...
command         show ip int brie
data            (failed)
errors          Device action failed: Performing action failed: Timeout exception while performing commands
--------------  -------------------------------------------------------------------------------------------

In this example, the response object shows that the command execution failed (FAILURE) and the errors attribute gives the reason for the failure (Timeout exception while performing commands).

>>> response.errors
['command execution failed: Device action failed: Performing action failed: Timeout exception while performing commands']

If one tries to access the command output, an exception is raised:

>>> response.data
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "<frozen radkit_client.sync.command>", line 299, in data
File "/Users/radkit-user/.pyenv/versions/3.9.12/lib/python3.9/concurrent/futures/_base.py", line 446, in result
    return self.__get_result()
File "/Users/radkit-user/.pyenv/versions/3.9.12/lib/python3.9/concurrent/futures/_base.py", line 391, in __get_result
    raise self._exception
File "/Users/radkit-user/.pyenv/versions/radkit-1_0/lib/python3.9/site-packages/anyio/from_thread.py", line 187, in _call_func
    retval = await retval
File "<frozen radkit_client.sync.event_loop_thread>", line 120, in coro
File "<frozen radkit_client.sync.command>", line 297, in in_loop
File "<frozen radkit_client.sync.event_loop_thread>", line 64, in result
radkit_client.async_.exec.ExecError: command execution failed: Device action failed: Performing action failed: Timeout exception while performing commands

command execution failed: Device action failed: Performing action failed: Timeout exception while performing commands

Note

This exec timeout is different from the timeout parameter to .wait() explained previously, which has no effect on the actual command execution.

Reset the session

By default the same terminal session will be cached and reused for successive calls to Device.exec, for faster response times and reduced load on the device (no need for a full SSH handshake every time).

The optional parameter reset_before=True can be used to force clearing the existing cached terminal connection, if it exists, before the execution of the command. This guarantees that the command will run on a freshly established terminal connection, with a prompt in a deterministic state:

>>> response = csr1.exec("ping 10.1.1.1 repeat 10", reset_before=True).wait()

Another option is reset_after=True which disables the caching of the connection. This can be used to ensure that the terminal connection state will not be shared with other Device.exec requests.

>>> response = csr1.exec("ping 10.1.1.1 repeat 10", reset_after=True).wait()

Single device, multiple commands

Multiple commands can be run by passing a list of commands (list[str]) instead of a single one (str). These commands will be executed sequentially in one sitting.

>>> response = csr1.exec(["show clock", "show ntp association"])
>>>
>>> response.wait()
[SUCCESS] <ExecResponse_OneDevice_ByCommand {2 entries}>
status    service_id    device     command                sudo    data
--------  ------------  ---------  ---------------------  ------  ----
SUCCESS   None          csr1       show clock            False    ...
SUCCESS   None          csr1       show ntp association  False    ...
--------------  -----------------------------------------------------

Note how response no longer is a single result, but a dict of results, one per command.

Warning

If the same command is specified multiple times, it will also be executed multiple times, however, in order to address the specific outputs individually, the response.by_command[index] notation needs to be used.

Individual command outputs are accessed by subscripting response with the command string:

>>> response['show clock']
[SUCCESS] SingleExecResponse(device_name='csr1', command='show clock')

-----------  --------------------------------------------------------
client_id    radkit-user@example.com
service_id   xxxx-yyyy-zzzz
device       csr1
device_uuid  b816846e-21b9-48ae-8efc-6a53af57c3b8
command      show clock
sudo         False
data         csr1#show clock\n14:43:32.937 EDT Thu Aug 11 2022\ncsr1#
errors
-----------  --------------------------------------------------------

>>> response['show clock'].data | print
csr1#show clock
14:43:32.937 EDT Thu Aug 11 2022
csr1#

>>> response['show ntp association'].data | print
csr1#show ntp association

address         ref clock       st   when   poll reach  delay  offset   disp
*~10.66.141.51    .GNSS.           1    195   1024   377 207.52  -3.862  1.011
~4.4.4.4         .TIME.          16      -     64     0  0.000   0.000 15937.
* sys.peer, # selected, + candidate, - outlyer, x falseticker, ~ configured
csr1#

Note

If a timeout parameter is provided in an .exec() statement with multiple commands, the same timeout applies to every command independently. Currently if a single command in the set times out, subsequent commands in the list will be skipped, all commands are considered to have timed out and no output is returned (see below). If more granularity is required, please use separate .exec() statements.

>>> response = csr1.exec(["show clock", "ping 10.1.1.1 repeat 10","show ntp association"], timeout=5)

>>> response
{
    'show clock': <radkit_client.sync.command.ExecSingleCommandResult object {command='show clock', status='FAILURE'} at 0x117d98d60>,
    'ping 10.1.1.1 repeat 10': <radkit_client.sync.command.ExecSingleCommandResult object {command='ping 10.1.1.1 repeat 10', status='FAILURE'} at 0x117d98220>,
    'show ntp association': <radkit_client.sync.command.ExecSingleCommandResult object {command='show ntp association', status='FAILURE'} at 0x117d98970>
}

Multiple devices, single command

To run a command on multiple devices, use DeviceDict.exec() on all or a subset of the Service inventory (see Multiple devices for details).

>>> csrs = service.inventory.filter("name", "csr")

>>> csrs
<radkit_client.sync.device.DeviceDict object at 0x118bb9e50>

name    host           device_type    Terminal    Netconf    Swagger    HTTP    description    failed
------  -------------  -------------  ----------  ---------  ---------  ------  -------------  --------
csr1    172.18.124.37  IOS_XE         True        True       False      False   csr1           True
csr2    172.18.124.38  IOS_XE         True        True       False      False                  False
csr3    172.18.124.39  IOS_XE         True        True       False      False                  False
csr4    172.18.124.55  IOS_XE         True        True       False      False                  False
csr5    172.18.124.42  IOS_XE         True        True       False      False                  False

>>> response = csrs.exec("show ver")

Note how response is now a mapping of device name to result object. The result for each command is accessed by subscripting the response with the name of the device:

>>> response
[SUCCESS] <ExecResponse_ByDevice_ToSingle {5 entries}>
status    client_id       device    command   sudo  data
--------  --------------  --------  --------- ----  -------------------------------------
SUCCESS   xxxx-yyyy-zzzz  csr1      show ver  False  csr1#show ver\nCisco IOS XE...
SUCCESS   xxxx-yyyy-zzzz  csr2      show ver  False  csr2#show ver\nCisco IOS XE...
SUCCESS   xxxx-yyyy-zzzz  csr3      show ver  False  csr3#show ver\nCisco IOS XE...
SUCCESS   xxxx-yyyy-zzzz  csr4      show ver  False  csr4#show ver\nCisco IOS XE...
SUCCESS   xxxx-yyyy-zzzz  csr5      show ver  False  csr5#show ver\nCisco IOS XE...

>>> response['csr1']
[SUCCESS] SingleExecResponse(device_name='csr1', command='show ver')
-----------  ----------------------------------------------------------------------------------
client_id    radkit-user@example.com
service_id   xxxx-yyyy-zzzz
device       csr1
device_uuid  b816846e-21b9-48ae-8efc-6a53af57c3b8
command      show ver
sudo         False
data         csr1#show ver\nCisco IOS XE Software, Version 17.06.02\nCisco IOS Software [Ben...
errors
-----------  ----------------------------------------------------------------------------------

>>> response['csr2'].data
'csr2#show ver\nCisco IOS XE Software...'

It is then possible to process the data for multiple devices at a time using standard Python syntax such as list/dict comprehensions as well as Client Pipe commands, for example:

>>> { dev: (res.data | regrep('^System image')) for dev, res in response.items() }
{
    'csr4': 'System image file is "bootflash:c8000v-universalk9.BLD_POLARIS_DEV_LATEST_20220721_184356.SSA.bin"',
    'csr2': 'System image file is "bootflash:c8000v-universalk9.17.07.01a.SPA.bin"',
    'csr1': 'System image file is "bootflash:c8000v-universalk9.17.06.02.SPA.bin"',
    'csr3': 'System image file is "bootflash:c8000v-universalk9.17.06.02.SPA.bin"',
    'csr5': 'System image file is "bootflash:c8000v-universalk9.17.08.01a.SPA.bin"'
}

Partial results and errors

A multi-device request is executed in parallel for all the devices in the set. If an error occurs for some devices, the response status is set to PARTIAL_SUCCESS, and the .status property of the individual entries shows which commands/devices failed. Accessing the .data attribute of the failed entries will result in an ExecError which contains the error message as well.

>>> devices = service.inventory.subset(['csr1', 'asa-1'])

>>> response = devices.exec("show ver").wait()
20:06:48.541Z ERROR | command execution failed [device_name='asa-1' commands=['show ver'] error='Device action failed: Permission error while preparing connection']

>>> response
[PARTIAL_SUCCESS] <ExecResponse_ByDevice_ToSingle {2 entries}>
status    service_id      device    command    sudo  data
--------  --------------  --------  ---------  ----  -----------------------------------------------------------------------------------
FAILURE   xxxx-yyyy-zzzz  asa-1     show ver   False  (failed)
SUCCESS   xxxx-yyyy-zzzz  csr1      show ver   False  csr1#show ver\nCisco IOS XE Software, Version 17.06.02\nCisco IOS Software [Ben...

Multiple devices, multiple commands

When running a list of commands on multiple devices, you simply have to subscript .result twice, first with the device name, then with the command string:

>>> csrs
<radkit_client.sync.device.DeviceDict object at 0x105c63520>

name    host           device_type    Terminal    Netconf    Swagger    HTTP    description    failed
------  -------------  -------------  ----------  ---------  ---------  ------  -------------  --------
csr1    172.18.124.37  IOS_XE         True        True       False      False   csr1           False
csr2    172.18.124.38  IOS_XE         True        True       False      False                  False

>>> response = csrs.exec(['show clock', 'show debug']).wait()

>>> response
[SUCCESS] <ExecResponse_ByDevice_ByCommand {4 entries}>
status    service_id      device    command     sudo  data
--------  --------------  --------  ----------  ----  ----
SUCCESS   xxxx-yyyy-zzzz  csr1      show clock  False  ...
SUCCESS   xxxx-yyyy-zzzz  csr1      show debug  False  ...
SUCCESS   xxxx-yyyy-zzzz  csr2      show clock  False  ...
SUCCESS   xxxx-yyyy-zzzz  csr2      show debug  False  ...

>>> response['csr1']
[SUCCESS] <ExecResponse_OneDevice_ByCommand {2 entries}>
status    service_id      device    command     sudo  data
--------  --------------  --------  ----------  ----  ----
SUCCESS   xxxx-yyyy-zzzz  csr1      show clock  False  csr1#show clock\n10:37:00.618 EDT Fri Aug 12 2022\ncsr1#
SUCCESS   xxxx-yyyy-zzzz  csr1      show debug  False  csr1#show debug\nIOSXE Conditional Debug Configs:\n\nConditional Debug Global St...

>>> response['csr1']['show clock']
[SUCCESS] SingleExecResponse(device_name='csr1', command='show clock')
-----------  --------------------------------------------------------
client_id    radkit-user@example.com
service_id   xxxx-yyyy-zzzz
device       csr1
device_uuid  b816846e-21b9-48ae-8efc-6a53af57c3b8
command      show clock
data         csr1#show clock\n10:37:00.618 EDT Fri Aug 12 2022\ncsr1#
errors
-----------  --------------------------------------------------------

>>> response['csr1']['show clock'].data
'csr1#show clock\n10:37:00.618 EDT Fri Aug 12 2022\ncsr1#'

Indexing and filtering a response object

The responses in an exec response object can be filtered or addressed by device name, device type, command name or index.

For this, there are .by_device[...], .by_device_name, .by_command[...] and .by_index[...] methods available in all the response classes. This allows for filtering the results by any of these fields or simply indexing by a sequence number. These methods can be chained, for instance:

filtered_response = response.by_device['csr1'].by_status['SUCCESS']

A response object can also be filtered according to a custom condition by passing a callable to filter(). This callable will receive ExecRecord instances and should return a boolean True for records that should be included:

from radkit_client.sync import ExecStatus, ExecRecord

def my_success_filter(entry: ExecRecord[str]) -> bool:
    return entry.status == ExecStatus.SUCCESS)

response = response.filter(my_success_filter)

Or in short, using one lambda expression:

response = response.filter(lambda entry: entry.status == ExecStatus.SUCCESS)

Or, by using .by_status[...]:

response = response.by_status["SUCCESS"]

Note

ExecRecord is a generic class. The [str] refers to the type of its data attribute. This is the output for the given command. When map() is called (see next section), the output can be mapped to a different output, and so, the type can be different.

Transforming (parsing) a response object

Response objects have a map() method that allows for transforming the data column of all their entries, for instance by parsing the response string into a data structure.

The returned object is always an ExecResponse_ByIndex instance, where the data column now contains the parsed data. The original data remains accessible through the raw_data column. ExecResponse_ByIndex is a generic type, where the type between the square brackets refers to the type of that data column.

The map function should accept ExecRecord which contains:

• all information about the individual entry, like a device type, which can be used for lookup up the right parser function;

• the data attribute that represents the data that needs to be parsed.

from radkit_client.sync import ExecStatus, ExecRecord, ExecResponse_ByIndex
from typing import Any

def parse_response(entry: ExecRecord[str]) -> dict[str, Any]:
    return {
        'version': re.findall('[0-9]+', entry.data)[0]
        'all_text': entry.data,
    }

parsed_response: ExecResponse_ByIndex[dict[str, Any]] = response.map(parse_response)

for response in parsed_response.values():
    print("Parsed response: ", response.data)

The map() method is chainable and can be combined with filter(). Both map() and filter() can be called before the exec response was returned from the service, and the returned view will be dynamically populated while the responses arrive.

When the map function raises an exception, the corresponding records will automatically have a status FAILED. This can also be used as a way to mark invalid outputs:

from radkit_client.sync import ExecStatus, ExecRecord, ExecError, ExecResponse_ByIndex
from typing import Any

def map_func(entry: ExecRecord[str]) -> str:
    if 'ERROR' in entry.data:
        return ExecError("Invalid output.")
    return entry.data

mapped_response: ExecResponse_ByIndex[str] = response.map(map_func)

# Print success entries only:
for response in mapped_response.by_status["SUCCESS"]:
    print("Parsed response: ", response.data)

Sorting a response object

ExecResponseBase objects have several sort methods that allow for sorting the final response view on any of the corresponding columns:

• sort_by_command()

• sort_by_device()

• sort_by_device_type()

• sort_by_status()

The returned object is a ExecResponse_ByIndex object itself.

In order to sort on two columns, the sort_by methods can be chained. The latter sort_by column in compared first, and if values are identical, the previous sort_by column is taken into account.

An example of filtering by status, then sorting by command, then taking the first two results:

response.by_status['SUCCESS'].sort_by_command[:2]