Capstone: Automating a Network Through a Controller

import json
import time
import requests

BASE_URL = "http://127.0.0.1:8766"
AUTH_HEADERS = {"Authorization": "Bearer demo-token"}
REQUEST_TIMEOUT = 2
POLL_DELAY_SECONDS = 0.5
MAX_POLLS = 10

REQUIRED_VLANS = {
  "SW1": [{"id": 10, "name": "USERS"}, {"id": 20, "name": "VOICE"}],
  "SW2": [{"id": 10, "name": "USERS"}],
  "SW3": [{"id": 10, "name": "USERS"}, {"id": 30, "name": "MGMT"}],
  "SW4": [{"id": 10, "name": "USERS"}],
  "SW5": [{"id": 10, "name": "USERS"}],
}


def get_inventory():
  """Query the controller's own inventory -- no local devices.csv this time."""
  response = requests.get(f"{BASE_URL}/devices", timeout=REQUEST_TIMEOUT)
  response.raise_for_status()
  return response.json()["devices"]


def submit_task(hostname):
  """Submit desired state as intent -- the controller decides how to apply it."""
  response = requests.post(
      f"{BASE_URL}/tasks",
      json={"device_id": hostname, "type": "vlan_config", "desired_vlans": REQUIRED_VLANS[hostname]},
      headers=AUTH_HEADERS,
      timeout=REQUEST_TIMEOUT,
  )
  response.raise_for_status()  # confirms the task was ACCEPTED -- not that it's done
  return response.json()["task_id"]


def poll_task(task_id):
  """Poll until the task reaches a final state. Accepted is not completed."""
  for _ in range(MAX_POLLS):
      response = requests.get(f"{BASE_URL}/tasks/{task_id}", timeout=REQUEST_TIMEOUT)
      response.raise_for_status()
      body = response.json()
      if body["status"] in ("completed", "failed"):
          return body
      time.sleep(POLL_DELAY_SECONDS)
  return {"status": "timed-out"}


def check_observed_state(hostname):
  """A completed task claim is not proof -- check observed state separately."""
  response = requests.get(f"{BASE_URL}/devices/{hostname}/observed-state", timeout=REQUEST_TIMEOUT)
  response.raise_for_status()
  actual_vlans = response.json()["vlans"]
  required = REQUIRED_VLANS[hostname]
  missing = [v for v in required if v not in actual_vlans]
  return len(missing) == 0, missing


def process_device(device):
  hostname = device["hostname"]

  task_id = submit_task(hostname)
  result = poll_task(task_id)

  if result["status"] == "failed":
      return {"hostname": hostname, "status": "task-failed", "detail": result.get("reason", "unknown")}
  if result["status"] == "timed-out":
      return {"hostname": hostname, "status": "task-timed-out", "detail": "task never reached a final state"}

  # Task reports "completed" -- that alone is not proof of anything.
  if not result["changed"]:
      return {"hostname": hostname, "status": "already-compliant", "detail": "no change was needed"}

  verified, missing = check_observed_state(hostname)
  if verified:
      return {"hostname": hostname, "status": "changed-and-verified", "detail": "observed state confirms the change"}
  return {
      "hostname": hostname,
      "status": "change-not-verified",
      "detail": f"task reported completed, but observed state is still missing: {[v['id'] for v in missing]}",
  }


devices = get_inventory()
results = [process_device(device) for device in devices]

with open("controller_health_check_report.json", "w") as f:
  json.dump(results, f, indent=4)

by_status = {}
for r in results:
  by_status[r["status"]] = by_status.get(r["status"], 0) + 1

print("\n--- Summary ---")
print(f"Checked: {len(results)} devices (queried from the controller's own inventory)")
for status, count in by_status.items():
  print(f"{status}: {count}")
print("Report written to controller_health_check_report.json")

Genuinely complete and runnable against this module's mock controller (Option B). Notice: no devices.csv anywhere — get_inventory() queries the controller directly, the concrete proof of Lesson 3's central point.

--- Summary ---
Checked: 5 devices (queried from the controller's own inventory)
already-compliant: 1
changed-and-verified: 2
task-failed: 1
change-not-verified: 1
Report written to controller_health_check_report.json

Real, verified output. SW1 was already compliant. SW2 and SW3 each needed a change, and observed state confirms both actually took effect. SW4's task was accepted but genuinely failed. SW5's task reported success but observed state proved otherwise.

[
  {
      "hostname": "SW1",
      "status": "already-compliant",
      "detail": "no change was needed"
  },
  {
      "hostname": "SW2",
      "status": "changed-and-verified",
      "detail": "observed state confirms the change"
  },
  {
      "hostname": "SW3",
      "status": "changed-and-verified",
      "detail": "observed state confirms the change"
  },
  {
      "hostname": "SW4",
      "status": "task-failed",
      "detail": "device unreachable from controller"
  },
  {
      "hostname": "SW5",
      "status": "change-not-verified",
      "detail": "task reported completed, but observed state is still missing: [10]"
  }
]

Real, verified output — diffed exactly against a real run of the script above. Every device has one entry, the same reporting discipline as every JSON report since Phase 2.6, now carrying controller-specific detail (a task failure reason; an observed-state mismatch) neither Netmiko nor Ansible's own report shapes needed to express.

import requests

response = requests.post(
  "https://catalyst-center.example.com/dna/system/api/v1/auth/token",
  auth=("admin", "REPLACE_ME"),  # never hardcoded for real -- environment variable
  verify=True,  # never disable certificate validation to make a demo work
  timeout=5,
)
token = response.json()["Token"]

No real Catalyst Center instance was available in this authoring environment -- this was not executed. The token endpoint's shape (/dna/system/api/v1/auth/token, returning a Token field) is exactly as Cisco's own Catalyst Center API documentation specifies.

response = requests.get(
  "https://catalyst-center.example.com/dna/intent/api/v1/network-device",
  headers={"X-Auth-Token": token},
  timeout=5,
)
print(response.json())

Catalyst Center's real inventory endpoint -- the response.json() parsing skill from Phase 3.1 applies identically here, once real credentials and a real instance are available. Not required to complete this module.