Chapter 3: Python Network Automation with Netmiko

1.1 What Is Netmiko and Why Does It Exist?

SSH was designed for human operators, not scripts. The raw Paramiko library can open SSH connections but has no understanding of Cisco CLI state machines — it cannot detect when a prompt has appeared, distinguish between user EXEC and global config mode, or handle the timing quirks of different vendor devices.

Netmiko — created by Kirk Byers in 2014 — wraps Paramiko with a network-device-aware layer. It ships with built-in support for over 80 device types including every major Cisco platform. The analogy: if Paramiko is raw electrical current, Netmiko is a power outlet — same energy, but shaped for the devices you actually plug in.

1.2 The ConnectHandler: Your Entry Point

Every Netmiko session starts with ConnectHandler. Pass it a device dictionary with type, address, and credentials. Netmiko handles the SSH handshake, login, and prompt negotiation automatically.

from netmiko import ConnectHandler

device = {
    "device_type": "cisco_xe",      # IOS XE (Catalyst 9K, ASR, CSR)
    "host": "192.168.1.1",
    "username": "admin",
    "password": "cisco123",
    "secret": "enable_secret",      # For privilege escalation (optional)
}

connection = ConnectHandler(**device)
print(connection.find_prompt())     # Confirms successful login
connection.disconnect()

The device_type parameter is critical — it controls which prompt patterns Netmiko expects and how it handles mode transitions.

1.3 send_command vs. send_config_set

send_command() is for operational (read-only) commands: show, ping, traceroute. It sends one command, waits for the prompt to return, and returns the output as a string.

send_config_set() is for pushing configuration. It accepts a list of commands, automatically issues configure terminal, sends each command in sequence, then exits with end.

1.4 Session Management and the Context Manager Pattern

Cisco devices typically have only 5–16 VTY lines. Unclosed SSH connections hold VTY lines indefinitely. Use the context manager pattern to guarantee cleanup:

with ConnectHandler(**device) as connection:
    output = connection.send_command("show version")
    print(output)
# disconnect() is called automatically here

1.5 Privilege Mode and Enable

Include "secret" in the device dictionary and call enable() after connecting when privilege escalation is required. If the user account is already granted privilege 15 (common with RADIUS/TACACS+), this step is unnecessary.

2.1 Deploying Configuration at Scale

Encoding intended network state as Python lists and deploying it with send_config_set() ensures every device receives identical configuration — eliminating the typos and inconsistencies of manual CLI work.

standard_config = [
    "ntp server 10.0.1.100",
    "ntp server 10.0.1.101 prefer",
    "logging buffered 16384 informational",
    "logging host 10.0.2.50",
    "no logging console",
]

with ConnectHandler(**device) as conn:
    output = conn.send_config_set(standard_config)
    conn.save_config()   # Issues "write memory" — never skip this!

2.2 Configuration Backup Automation

A modular backup function that takes a device dictionary and backup directory is easy to reuse in loops and concurrent executors:

from datetime import datetime
import os

def backup_device_config(device: dict, backup_dir: str = "./backups") -> str:
    os.makedirs(backup_dir, exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    filename = f"{backup_dir}/backup_{device['host']}_{timestamp}.txt"

    with ConnectHandler(**device) as conn:
        running_config = conn.send_command("show running-config")

    with open(filename, "w") as f:
        f.write(running_config)

    return filename

2.3 Verifying Configuration After Push

The push-then-verify pattern is the foundation of reliable automation. After every configuration change, read back the relevant state to confirm it took effect:

with ConnectHandler(**device) as conn:
    conn.send_config_set(["ntp server 10.0.1.100"])
    conn.save_config()

    output = conn.send_command("show ntp associations")
    if "10.0.1.100" in output:
        print(f"[{device['host']}] NTP server confirmed.")
    else:
        print(f"[{device['host']}] WARNING: NTP server not yet visible.")

This pattern also becomes the basis for drift detection: run the verification step alone (without the push) to audit whether a device still matches intended state.

2.4 Commands That Require Confirmation

Use expect_string when a command prompts for confirmation, otherwise Netmiko will hang waiting for a prompt that never matches:

output = conn.send_command(
    "reload in 10",
    expect_string=r"Proceed with reload\?",
)
output += conn.send_command("yes", expect_string=r"#")

3.1 The Problem with Raw CLI Text

Raw show command output is human-readable but machine-hostile. Parsing column offsets, splitting lines, and handling platform variations across dozens of commands is unsustainable. Structured parsing converts CLI text into Python data structures — lists of dicts or nested dicts — so you access fields by name.

3.2 TextFSM with ntc-templates

Pass use_textfsm=True to send_command(). When a matching template from the community-maintained ntc-templates library exists, Netmiko returns a list of flat dictionaries:

parsed = conn.send_command("show ip interface brief", use_textfsm=True)

for intf in parsed:
    if intf["status"] != "up" or intf["proto"] != "up":
        print(f"ALERT: {intf['intf']} ({intf['ipaddr']}) is {intf['status']}/{intf['proto']}")

3.3 Genie Parser Integration

Cisco's official Genie library (part of pyATS) returns deeply nested dictionaries following a rich, vendor-documented schema. This is the right choice for complex protocol analysis:

bgp_data = conn.send_command("show ip bgp summary", use_genie=True)

neighbors = (
    bgp_data
    .get("vrf", {})
    .get("default", {})
    .get("neighbor", {})
)

for ip, data in neighbors.items():
    state = data.get("session_state", "unknown")
    print(f"BGP Neighbor: {ip} | State: {state}")

3.4 TextFSM vs. Genie: Choosing the Right Tool

3.5 Writing Reusable Parsing Libraries

Avoid scattering use_textfsm=True flags throughout ad-hoc scripts. Centralize parsing in a dedicated module to isolate parser changes from business logic and enable unit testing with recorded CLI output:

# netops/parsers.py

def get_interfaces(conn) -> list[dict]:
    return conn.send_command("show ip interface brief", use_textfsm=True)

def get_bgp_summary(conn) -> dict:
    return conn.send_command("show ip bgp summary", use_genie=True)

def get_routes(conn, prefix_filter: str = None) -> list[dict]:
    routes = conn.send_command("show ip route", use_textfsm=True)
    if prefix_filter:
        return [r for r in routes if r.get("network", "").startswith(prefix_filter)]
    return routes

4.1 Sequential vs. Concurrent Execution

At 3–5 seconds per device, a sequential loop over 100 devices takes 5–8 minutes. Netmiko SSH is I/O-bound — threads spend most of their time waiting for network responses, not computing. Threading overlaps those waits: ThreadPoolExecutor is the standard tool.

4.2 Loading Device Inventories from YAML

Store device lists in version-controlled YAML and inject credentials from environment variables — never commit credentials to source control:

import yaml, os

def load_inventory(path: str) -> list[dict]:
    password = os.environ.get("DEVICE_PASSWORD")
    if not password:
        raise EnvironmentError("DEVICE_PASSWORD not set.")
    with open(path) as f:
        data = yaml.safe_load(f)
    for device in data["devices"]:
        device["password"] = password
    return data["devices"]

4.3 Concurrent Execution with ThreadPoolExecutor

The production-standard pattern for parallel Netmiko operations — study the exception handling and finally disconnect carefully:

from concurrent.futures import ThreadPoolExecutor, as_completed
from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException

def collect_show_version(device: dict) -> dict:
    host = device["host"]
    conn = None
    try:
        conn = ConnectHandler(**device)
        output = conn.send_command("show version")
        log.info(f"[{host}] Success.")
        return {"host": host, "output": output, "status": "success"}

    except NetmikoTimeoutException:
        log.error(f"[{host}] Connection timed out.")
        return {"host": host, "output": None, "status": "timeout"}

    except NetmikoAuthenticationException:
        log.error(f"[{host}] Authentication failed.")
        return {"host": host, "output": None, "status": "auth_failed"}

    except Exception as e:
        log.exception(f"[{host}] Unexpected error: {e}")
        return {"host": host, "output": None, "status": f"error: {e}"}

    finally:
        if conn:
            conn.disconnect()   # Always runs, even after exceptions

results = []
with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {executor.submit(collect_show_version, dev): dev for dev in devices}
    for future in as_completed(futures):
        results.append(future.result())

4.4 Tuning max_workers

4.5 Exception Handling Reference

4.6 Tuning for Slow Devices

device = {
    "device_type": "cisco_xe",
    "host": "10.0.0.1",
    "username": "admin",
    "password": "cisco",
    "conn_timeout": 15,         # TCP connection timeout (default: 10s)
    "banner_timeout": 20,       # SSH banner wait (default: 15s)
    "global_delay_factor": 2,   # Multiplier for all internal wait timers
    "read_timeout": 30,         # Max wait for show command output
}

4.7 Production Logging Best Practices

Never use print() in production automation. Use Python's logging module with timestamps, levels, and dual handlers (file + console) to create a complete audit trail.

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    handlers=[
        logging.FileHandler("netops_automation.log"),
        logging.StreamHandler(),
    ]
)
log = logging.getLogger(__name__)

log.info("Config pushed successfully.")
log.error("Connection failed.")
log.exception("Unexpected exception.")  # Logs full traceback