Command Runner Guide

Introduction

Command runner is the feature in Catalyst Center that allows you to execute commands on the devices managed by Catalyst Center.

At the moment, command runner supports Read-only commands and not configuration ones.

Goal

The goals of this guide are:

Execute a show version command.
Execute a show ip interface brief command.
Retrieve the results.

Command Runner workflow

Endpoints and methods used

POST /dna/system/api/v1/auth/token
GET /dna/intent/api/v1/network-device
POST /dna/intent/api/v1/network-device-poller/cli/read-request
GET /dna/intent/api/v1/task/{task_id}
GET /dna/intent/api/v1/file/{file_Id}

Prerequisites

For this guide, we recommend that the developer becomes familiar with authenticating and obtaining the device ID of the target devices.

Authentication and Authorization
Devices

Environment

This guide was developed using:

Command Runner API

The command runner API consists of two endpoints: The first endpoint retrieves the accepted keywords for the CLI command, and the second sends the commands.

For this guide, we execute a command runner call. The following steps ensure a successful API execution:

Authenticate against the Catalyst Center API.
Obtain the IDs of the devices that you want to send the commands to.
Execute the commands against a list of devices and obtain the task ID.
Use the task ID to query the Task API for a result, if successful obtain the file ID of the result.
Retrieve the file contents from the file API.

Get Device ID

First, we must authenticate and retrieve a token from the API.

Note: Do not use verify=False or urllib3.disable_warnings() if you are not sure of its purpose. Read Authentication and Authorization.

import requests
from requests.auth import HTTPBasicAuth
import urllib3
urllib3.disable_warnings()

BASE_URL = 'https://<IP Address>'
AUTH_URL = '/dna/system/api/v1/auth/token'
USERNAME = '<USERNAME>'
PASSWORD = '<PASSWORD>'

response = requests.post(BASE_URL + AUTH_URL, auth=HTTPBasicAuth(USERNAME, PASSWORD), verify=False)
token = response.json()['Token']

With the token, we can proceed to query the devices API to get the list of devices we want to send the command to. We filter based on platform ID:

headers = {'X-Auth-Token': token, 'Content-Type': 'application/json'}
DEVICES_URL = '/dna/intent/api/v1/network-device'
query_string_params = {'platformId': 'C9500-40X'}
response = requests.get(BASE_URL + DEVICES_URL, headers = headers, params=query_string_params, verify=False)
devices = []
for device in response.json()['response']:
  devices.append(device['id'])

Command runner

This API enables the user to send a list of commands and specify the devices where the commands should execute. Command runner is an asynchronous API, so the user doesn't obtain the execution result immediately. Check the async guide for more information.

The main parameters for the call are:

deviceUuidDs: List of devices IDs, captured from the Devices API call
commands: List of commands for execution.

We send the commands show version and show ip interface brief

payload = {
  "commands": [
    "show version",
    "show ip int brief"
  ],
  "deviceUuids": devices,
  "timeout": 0
}
COMMAND_RUNNER_SEND_URL = '/dna/intent/api/v1/network-device-poller/cli/read-request'
response = requests.post(BASE_URL + COMMAND_RUNNER_SEND_URL, data=json.dumps(payload), headers=headers, verify=False)
task_id = response.json()['response']['taskId']

Task API

The Task API consists of 5 endpoints. For this guide, we use the Get task by id endpoint to get the status of the task we executed with the command runner API.

TASK_BY_ID_URL = '/dna/intent/api/v1/task/{task_id}'
response = requests.get(BASE_URL + TASK_BY_ID_URL.format(task_id=task_id), headers=headers, verify=False)
progress_json = json.loads(response.json()['response']['progress'])
file_id = progress_json['fileId']

File API

Once the task is complete, the final step involves retrieving the generated file by using the Download file by fileId endpoint.

FILE_GET_BY_ID = '/dna/intent/api/v1/file/{file_id}'
response = requests.get(BASE_URL + FILE_GET_BY_ID.format(file_id=file_id), headers=headers, verify=False)
file_json = response.json()

The format of the response is a json with the following fields:

[
  {
    'deviceUuid': 'ABC',
    'commandResponses': {
      'SUCCESS': {
        'show version': 'show version\nCisco IOS XE Software...',
        'show ip int brief': 'show ip int brief\nInterface...'
      },
      'FAILURE': {},
      'BLACKLISTED': {}
    }
  },
   {
    'deviceUuid': 'XYZ',
    'commandResponses': {
      'SUCCESS': {
        'show version': 'show version\nCisco IOS XE Software...',
        'show ip int brief': 'show ip int brief\nInterface...'
      },
      'FAILURE': {},
      'BLACKLISTED': {}
    }
  }
]

With the preceding reference, to extract the response of the show ip int brief of the second device, it is possible to do:

print(file_json[1]['commandResponses']['SUCCESS']['show ip int brief'])

Code

The repository for this guide is here. The final code with functions appears as below.

# Modules import
import requests
from requests.auth import HTTPBasicAuth
import json
import time

# Disable SSL warnings. Not needed in production environments with valid certificates
import urllib3
urllib3.disable_warnings()

# Authentication
BASE_URL = 'https://<IP Address>'
AUTH_URL = '/dna/system/api/v1/auth/token'
USERNAME = '<USERNAME>'
PASSWORD = '<PASSWORD>'

# URLs
DEVICES_URL = '/dna/intent/api/v1/network-device'
COMMAND_RUNNER_SEND_URL = '/dna/intent/api/v1/network-device-poller/cli/read-request'
TASK_BY_ID_URL = '/dna/intent/api/v1/task/{task_id}'
FILE_GET_BY_ID = '/dna/intent/api/v1/file/{file_id}'

# Get Authentication token
def get_dnac_jwt_token():
    response = requests.post(BASE_URL + AUTH_URL,
                             auth=HTTPBasicAuth(USERNAME, PASSWORD),
                             verify=False)
    token = response.json()['Token']
    return token

# Get devices by platform ID
def get_devices_ids(headers, query_string_params):
    response = requests.get(BASE_URL + DEVICES_URL, headers=headers,
                            params=query_string_params, verify=False)
    devices = []
    for device in response.json()['response']:
        devices.append(device['id'])
    return devices

# Send commands to devices
def send_commands(headers, payload):
    response = requests.post(BASE_URL + COMMAND_RUNNER_SEND_URL, json=payload,
                            headers=headers, verify=False)
    return response.json()['response']

# Get Task result
def get_task(headers, task_id):
    response = requests.get(BASE_URL + TASK_BY_ID_URL.format(task_id=task_id), headers=headers, verify=False)
    return response

# Get file with command results
def get_file(headers, file_id):
    response = requests.get(BASE_URL + FILE_GET_BY_ID.format(file_id=file_id), headers=headers, verify=False)
    return response.json()

def main():
    # obtain the Catalyst Center Auth Token
    token = get_dnac_jwt_token()
    headers = {'X-Auth-Token': token, 'Content-Type': 'application/json'}
    query_string_params = {'platformId': 'C9500-40X'}
    devices = get_devices_ids(headers, query_string_params)
    payload = {
    "commands": [
        "show version",
        "show ip int brief"
    ],
    "deviceUuids": devices,
    "timeout": 0
    }

    response = send_commands(headers, payload)

    # Wait to have a response back from the devices
    time.sleep(10)

    response = get_task(headers, response['taskId'])
    progress_json = json.loads(response.json()['response']['progress'])
    file_id = progress_json['fileId']

    # Get file
    response = get_file(headers, file_id)
    print(response[0]['commandResponses']['SUCCESS']['show ip int brief'])

if __name__ == "__main__":
    main()