01 API endpoints for RADKit
This guide outlines the process for creating API endpoints that leverage your RADKit Server to query and process network information.
The core of this process is the fastapi-middleware container, defined in this repository's Docker Compose file. This Python-based container utilizes the official RADKit SDK to interact with your RADKit server's device inventory, send commands, and retrieve/parse results.
Upon creation, the container mounts a FastAPI web server and authenticates the user specified in your config.yaml file with the RADKit cloud.
Important: This user must have been previously onboarded in the RADKit cloud using the
make onboardcommand. Please follow the instructions in the main README for this crucial step if you haven't already.
Once authenticated, the FastAPI server is ready to accept API requests. Your task is to create new endpoints that query, parse, and process network configurations from your RADKit Server.
To begin, open the file radkit-to-grafana-fastapi-middleware/main.py. The main() function in this file handles the initial user login to the RADKit cloud and RADKit Server using the details provided in config.yaml.
Creating an API endpoint in a FastAPI server follows this general structure:
@app.operation_type("/my_url/{parameter}")
def my_function(parameter):
# logic of my endpoint
return result_of_this_operation| Component | Description |
|---|---|
operation_type |
Can be any standard REST API method (e.g., GET, PUT, POST, PATCH, DELETE). |
/my_url/{parameter} |
The URL path for your endpoint, which can include optional parameters. |
def my_function(parameter) |
The Python function executed when the endpoint is invoked. Any URL parameters must be defined as function arguments. |
A special service variable is made available for your functions. This object provides access to various RADKit features, including the device inventory and device-specific interactions.
For instance, to retrieve all devices in your inventory, you can use the service.inventory property:
@app.get("/devices")
def get_devices():
return { device.name for device in service.inventory.values() }This endpoint will return a JSON list of device names. You can test it using cURL in your terminal:
curl "http://localhost:8000/devices"["p0-2e","p0-1e"]For a comprehensive list of properties and parameters available through the service object, consult the python SDK official documentation.
You can send specific CLI commands to any target device using the exec method of the service object. The RADKit server executes the command, and a raw text result is returned to your endpoint.
Here's an example demonstrating raw command execution:
@app.get("/device/{device_name}/exec/{cmd}")
def exec_show_cmd(device_name: str, cmd:str) -> Any:
single_result = service.inventory[device_name].exec(cmd).wait()
return single_result.result.dataTesting with cURL to run show ip interface brief on device p0-1e:
curl "http://localhost:8000/device/p0-1e/exec/show%20ip%20interface%20brief"p0-1E#show ip interface brief\nInterface IP-Address OK? Method Status Protocol\nVlan1 unassigned YES NVRAM up up \nGigabitEthernet0/0 10.48.172.58 YES NVRAM up up \nGigabitEthernet1/0/1 unassigned YES unset down down \nGigabitEthernet1/0/2 192.168.112.35 YES NVRAM up up \nGigabitEthernet1/0/3 unassigned YES unset down down \nGigabitEthernet1/0/4 unassigned YES unset up up . . ."
The challenge with raw command querying is that processing the results (e.g., extracting specific values) requires manual parsing using tools like regex or TextFSM.
Fortunately, the Cisco RADKit framework integrates Genie: a powerful utility with predefined parsers. Genie automatically transforms raw CLI outputs into structured Python dictionaries, making data handling significantly easier.
Cisco maintains a list of commands supported by Genie parsers, filtered by command and platform. If a required parser isn't available, you can request it from the Cisco RADKit business unit or create your own. Instructions for creating and mounting custom parsers are available here and here.
Consider an example where you want to retrieve only the status and protocol of device interfaces, supporting both iosxr and iosxe platforms (which have different parsers).
The following function demonstrates using Genie to parse show ip interface brief output:
import radkit_genie
@app.get("/device/{device_name}/interfaces/brief")
def parse_ip_int(device_name: str) -> Any:
interfaces_list = []
# Determine device type for correct Genie parsing
device_type = 'iosxr' if service.inventory[device_name].device_type == 'IOS_XR' else 'iosxe'
# Execute command and parse with Genie
single_result = service.inventory[device_name].exec("show ip interface brief").wait()
parsed = radkit_genie.parse(single_result, os=device_type)
values = parsed[device_name]["show ip interface brief"].data
# The Genie parser attempts to guess the OS type, but providing it (e.g., `os=device_type`) is good practice.
# The `device_type` property of your `service` instance can provide this.
# Example of the structured output from Genie (partial):
# {
# "interface": {
# "Vlan1": {
# "ip_address": "unassigned",
# "interface_is_ok": "YES",
# "method": "NVRAM",
# "status": "up",
# "protocol": "up",
# "name": "Vlan1",
# "interface_status": "up",
# "protocol_status": "up"
# },
# "GigabitEthernet0/0": {
# "ip_address": "10.48.172.58",
# "interface_is_ok": "YES",
# "method": "NVRAM",
# "status": "up",
# "protocol": "up",
# "name": "GigabitEthernet0/0",
# "interface_status": "up",
# "protocol_status": "up"
# },
# . . .
# }
# Data manipulation to homogenize format for iosxe platform (e.g., 'status' vs 'interface_status')
for interface_name in values['interface'].keys():
if device_type == 'iosxe':
values['interface'][interface_name]['interface_status'] = values['interface'][interface_name]['status']
values['interface'][interface_name]['protocol_status'] = values['interface'][interface_name]['protocol']
# Data must be put on a list format so it is easily rendered on a Grafana visualization
interface_dict = values['interface'][interface_name]
interface_dict['name'] = interface_name
interfaces_list.append(interface_dict)
# Return as JSON list for easier Grafana rendering
return json.loads(json.dumps(interfaces_list))When invoked, this endpoint returns a list of interface dictionaries, ideal for Grafana dashboards:
curl "http://localhost:8000/device/p0-1e/interfaces/brief" [
{
"ip_address": "unassigned",
"interface_is_ok": "YES",
"method": "NVRAM",
"status": "up",
"protocol": "up",
"name": "Vlan1"
},
{
"ip_address": "192.168.113.1",
"interface_is_ok": "YES",
"method": "NVRAM",
"status": "up",
"protocol": "up",
"name": "Vlan1021"
},
{
"ip_address": "172.16.111.1",
"interface_is_ok": "YES",
"method": "NVRAM",
"status": "up",
"protocol": "up",
"name": "Vlan1022"
},
. . .
]While the previous endpoints provide current device configurations, you often need to track changes over time. For this, a time-series database is essential. This project integrates InfluxDB, a popular choice for this purpose, with a pre-configured container ready for use.
Let's look at an example to track the evolution of traffic rates on all interfaces of a specific device.
The following function will:
- Use the RADKit service to execute the
show interfacescommand on the specified device. - Parse the results into a Python dictionary using
Genie. - Extract
in_rateandout_ratevalues for each interface. - Create a record in the InfluxDB container (within the
my-radkit-bucketdefined in Docker Compose) with these measurements.
from radkit_client.sync import Client
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
INFLUXDB_URL = "http://influxdb:8086"
INFLUXDB_TOKEN = "radkit-to-grafana"
INFLUXDB_ORG = "radkit-to-grafana"
INFLUXDB_BUCKET = "my-radkit-bucket"
@app.get("/device/{device_name}/interfaces/traffic")
def get_interface_traffic(device_name: str) -> Any:
influxdb_points = []
# Execute command and parse with Genie
raw_result = service.inventory[device_name].exec("show interfaces").wait()
parsed_result = radkit_genie.parse(raw_result).to_dict()
# Process each interface and prepare data for InfluxDB
for interface in parsed_result[device_name]["show interfaces"]:
if "rate" in parsed_result[device_name]["show interfaces"][interface]["counters"].keys():
in_rate = parsed_result[device_name]["show interfaces"][interface]["counters"]["rate"]["in_rate"]
out_rate = parsed_result[device_name]["show interfaces"][interface]["counters"]["rate"]["out_rate"]
total_rate = in_rate + out_rate
else:
total_rate = 0
# Create an InfluxDB Point for each interface
influxdb_points.append(
Point("interface_traffic") \
.tag("device", device_name) \
.tag("interface", str(interface)) \
.field("total_rate", total_rate)
)
# Write data to InfluxDB
try:
with InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG) as client:
write_api = client.write_api(write_options=SYNCHRONOUS)
write_api.write(bucket=INFLUXDB_BUCKET, org=INFLUXDB_ORG, record=influxdb_points)
return {"status": True, "msg": f"{len(influxdb_points)} traffic points successfully written to InfluxDB."}
except Exception as e:
print(f"Error writing to InfluxDB: {e}")
return {"status": True, "msg": f"Error: {e}"}When invoked via cURL, this endpoint will return a confirmation message, and a timestamped record will be created in the InfluxDB container:
curl "http://localhost:8000/device/p0-1e/interfaces/traffic"{"status":true,"msg":"84 traffic points successfully written to InfluxDB."}To verify the data creation in InfluxDB, you can navigate to the container's web UI and visually explore the data:
http://localhost:8086/
Log in using the credentials specified in the Docker Compose service (admin/cisco123). On the left sidebar, click the up-arrow icon, then navigate to Buckets to view your data:
