Network Automation with Python | Automate Cisco Networks with Python Scripts

The Network Engineer's Dilemma: Manual Tasks vs. Strategic Work

It's 2 AM on a Saturday. You're manually configuring 50 switches for a new VLAN rollout when you realize:

• ⏰ 6 hours spent on repetitive CLI commands
• 🔧 Human errors creeping into configurations
• 📈 Business demands outpacing manual processes
• 🚀 Competitors automating while you're stuck in the CLI

What if you could deploy that same VLAN configuration to 100 devices in 2 minutes with zero errors? Welcome to Network Automation with Python - where manual tasks become automated workflows.

Network Automation Pyramid: From Scripting to AIOps

Automation Maturity Model:

[ AIOps ] ← Predictive automation, self-healing networks
   ↓
[ API Integration ] ← REST APIs, YANG models, telemetry
   ↓  
[ Configuration Management ] ← Ansible, Python scripts, templates
   ↓
[ Basic Scripting ] ← Netmiko, Paramiko, CLI automation

Why Python for Network Automation?

• ✅ Rich ecosystem (Netmiko, NAPALM, Nornir, PyATS)
• ✅ Vendor agnostic approach
• ✅ Extensive libraries for network protocols
• ✅ Community support and documentation
• ✅ Integration capabilities with other systems

Getting Started: Essential Python Libraries

Core Automation Libraries:

# requirements.txt - Essential network automation libraries
netmiko==4.1.0
napalm==4.1.0
nornir==3.3.0
paramiko==3.2.0
requests==2.28.0
jinja2==3.1.0
pyyaml==6.0
textfsm==1.1.0
ntc-templates==3.0.0

Installation and Setup:

# Create virtual environment
python -m venv netauto
source netauto/bin/activate  # Linux/Mac
# netauto\Scripts\activate  # Windows

# Install packages
pip install -r requirements.txt

Netmiko: Your Swiss Army Knife for CLI Automation

Basic Device Connection and Configuration:

#!/usr/bin/env python3
"""
Basic Netmiko Example - Connecting to Cisco Devices
Color-coded for different operations:
🟢 Green - Connection setup
🔵 Blue - Configuration commands  
🟡 Yellow - Data collection
🔴 Red - Error handling
"""

from netmiko import ConnectHandler
import json

# 🟢 Device connection parameters
cisco_device = {
    'device_type': 'cisco_ios',
    'host': '192.168.1.1',
    'username': 'admin',
    'password': 'SecurePass123!',
    'secret': 'EnablePass456!',  # Enable password
}

def basic_netmiko_operations():
    try:
        # 🟢 Establish connection
        print("🟢 Connecting to device...")
        connection = ConnectHandler(**cisco_device)
        connection.enable()  # Enter enable mode
        
        # 🟡 Collect device information
        print("🟡 Gathering device information...")
        show_version = connection.send_command('show version', use_textfsm=True)
        show_ip_int_brief = connection.send_command('show ip interface brief', use_textfsm=True)
        
        print(f"Device Model: {show_version[0]['hardware'][0] if show_version else 'N/A'}")
        print(f"Software Version: {show_version[0]['version'] if show_version else 'N/A'}")
        
        # 🔵 Apply configuration
        print("🔵 Applying configuration...")
        config_commands = [
            'interface GigabitEthernet0/1',
            'description Configured by Python Script',
            'no shutdown',
        ]
        
        output = connection.send_config_set(config_commands)
        print("Configuration output:")
        print(output)
        
        # 🟡 Verify configuration
        print("🟡 Verifying configuration...")
        show_run_interface = connection.send_command('show run interface GigabitEthernet0/1')
        print("Interface configuration:")
        print(show_run_interface)
        
        # 🟢 Close connection
        connection.disconnect()
        print("🟢 Disconnected from device")
        
    except Exception as e:
        # 🔴 Error handling
        print(f"🔴 Error: {str(e)}")

if __name__ == "__main__":
    basic_netmiko_operations()

Multi-Device Configuration Deployment:

#!/usr/bin/env python3
"""
Multi-Device Configuration Deployment
Color-coded for different device groups:
🟢 Green - Core devices
🔵 Blue - Distribution devices  
🟡 Yellow - Access devices
🟣 Purple - Results summary
"""

from netmiko import ConnectHandler
from concurrent.futures import ThreadPoolExecutor
import threading

# 🎨 Color-coded device inventory
devices = {
    '🟢 CORE': [
        {
            'device_type': 'cisco_ios',
            'host': 'core-switch-1.company.com',
            'username': 'admin',
            'password': 'SecurePass123!',
            'secret': 'EnablePass456!',
        }
    ],
    '🔵 DISTRIBUTION': [
        {
            'device_type': 'cisco_ios',
            'host': 'dist-switch-1.company.com',
            'username': 'admin', 
            'password': 'SecurePass123!',
            'secret': 'EnablePass456!',
        },
        {
            'device_type': 'cisco_ios',
            'host': 'dist-switch-2.company.com',
            'username': 'admin',
            'password': 'SecurePass123!', 
            'secret': 'EnablePass456!',
        }
    ],
    '🟡 ACCESS': [
        {
            'device_type': 'cisco_ios',
            'host': 'access-switch-1.company.com',
            'username': 'admin',
            'password': 'SecurePass123!',
            'secret': 'EnablePass456!',
        }
    ]
}

# 🔧 Configuration templates by device type
config_templates = {
    '🟢 CORE': [
        'spanning-tree mode rapid-pvst',
        'spanning-tree extend system-id',
        'vtp mode transparent',
    ],
    '🔵 DISTRIBUTION': [
        'spanning-tree mode rapid-pvst', 
        'spanning-tree backbonefast',
        'vtp mode client',
    ],
    '🟡 ACCESS': [
        'spanning-tree mode rapid-pvst',
        'spanning-tree portfast default',
        'spanning-tree bpduguard default',
    ]
}

def deploy_configuration(device_info, device_group):
    """Deploy configuration to a single device"""
    try:
        print(f"{device_group} Connecting to {device_info['host']}...")
        
        # Connect to device
        connection = ConnectHandler(**device_info)
        connection.enable()
        
        # Get appropriate configuration
        config_commands = config_templates.get(device_group, [])
        
        # Deploy configuration
        if config_commands:
            output = connection.send_config_set(config_commands)
            print(f"{device_group} ✅ Configuration deployed to {device_info['host']}")
        
        # Verify deployment
        show_spanning_tree = connection.send_command('show spanning-tree summary')
        connection.disconnect()
        
        return {
            'device': device_info['host'],
            'status': 'SUCCESS',
            'group': device_group,
            'output': show_spanning_tree[:100] + '...'  # Truncate for display
        }
        
    except Exception as e:
        print(f"{device_group} ❌ Failed to configure {device_info['host']}: {str(e)}")
        return {
            'device': device_info['host'], 
            'status': 'FAILED',
            'group': device_group,
            'error': str(e)
        }

def mass_configuration_deployment():
    """Deploy configuration to all devices concurrently"""
    print("🚀 Starting mass configuration deployment...")
    
    all_devices = []
    for group, device_list in devices.items():
        for device in device_list:
            all_devices.append((device, group))
    
    # 🟣 Use threading for parallel execution
    results = []
    with ThreadPoolExecutor(max_workers=5) as executor:
        future_to_device = {
            executor.submit(deploy_configuration, device, group): (device, group) 
            for device, group in all_devices
        }
        
        for future in future_to_device:
            result = future.result()
            results.append(result)
    
    # 🟣 Print summary
    print("\n" + "="*50)
    print("🟣 DEPLOYMENT SUMMARY")
    print("="*50)
    
    success_count = sum(1 for r in results if r['status'] == 'SUCCESS')
    failed_count = sum(1 for r in results if r['status'] == 'FAILED')
    
    print(f"✅ Successful: {success_count}")
    print(f"❌ Failed: {failed_count}")
    print(f"📊 Total: {len(results)}")
    
    # Print detailed results by group
    for group in devices.keys():
        group_results = [r for r in results if r['group'] == group]
        group_success = sum(1 for r in group_results if r['status'] == 'SUCCESS')
        print(f"\n{group}: {group_success}/{len(group_results)} successful")

if __name__ == "__main__":
    mass_configuration_deployment()

NAPALM: Multi-Vendor Configuration Management

Configuration Management with NAPALM:

#!/usr/bin/env python3
"""
NAPALM Configuration Management
Color-coded for configuration operations:
🟢 Green - Device connection
🔵 Blue - Configuration management  
🟡 Yellow - Compliance checking
🔴 Red - Rollback operations
"""

import napalm
import json
from datetime import datetime

def napalm_configuration_management():
    # 🟢 Device connection
    print("🟢 Connecting to device via NAPALM...")
    
    driver = napalm.get_network_driver('ios')
    device = driver(
        hostname='192.168.1.1',
        username='admin', 
        password='SecurePass123!',
        optional_args={'secret': 'EnablePass456!'}
    )
    
    try:
        device.open()
        
        # 🔵 Get current configuration
        print("🔵 Gathering current configuration...")
        running_config = device.get_config(retrieve='running')
        startup_config = device.get_config(retrieve='startup')
        
        # 🟡 Check compliance
        print("🟡 Checking configuration compliance...")
        
        # Desired configuration
        desired_config = """
        interface GigabitEthernet0/1
         description Managed by NAPALM
         no shutdown
        !
        logging host 10.1.100.100
        """
        
        # Compare configurations
        device.load_merge_candidate(config=desired_config)
        diff = device.compare_config()
        
        if diff:
            print("🔵 Configuration changes needed:")
            print(diff)
            
            # 🔵 Commit changes
            choice = input("Commit these changes? (yes/no): ")
            if choice.lower() == 'yes':
                device.commit_config()
                print("✅ Configuration committed")
            else:
                device.discard_config()
                print("❌ Changes discarded")
        else:
            print("✅ Device is compliant")
        
        # 🟡 Collect facts
        print("🟡 Collecting device facts...")
        facts = device.get_facts()
        print(json.dumps(facts, indent=2))
        
        # 🔴 Rollback example
        print("🔴 Rollback capability demonstration...")
        device.rollback()
        
    except Exception as e:
        print(f"🔴 Error: {str(e)}")
    finally:
        device.close()

if __name__ == "__main__":
    napalm_configuration_management()

Network Automation Use Cases

1. Automated Backup Configuration:

#!/usr/bin/env python3
"""
Automated Configuration Backup
Color-coded for backup operations:
🟢 Green - Successful backups
🔵 Blue - Backup in progress
🟡 Yellow - Archive operations  
🔴 Red - Backup failures
"""

from netmiko import ConnectHandler
import os
from datetime import datetime
import threading

def backup_device_config(device_info, backup_dir):
    """Backup configuration for a single device"""
    try:
        hostname = device_info['host']
        print(f"🔵 Backing up {hostname}...")
        
        # Connect to device
        connection = ConnectHandler(**device_info)
        connection.enable()
        
        # Get running configuration
        running_config = connection.send_command('show running-config')
        
        # Get startup configuration  
        startup_config = connection.send_command('show startup-config')
        
        # Create backup filename with timestamp
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_file = os.path.join(backup_dir, f"{hostname}_{timestamp}.cfg")
        
        # Save configuration to file
        with open(backup_file, 'w') as f:
            f.write(f"! Backup of {hostname} at {timestamp}\n")
            f.write("! Running Configuration:\n")
            f.write(running_config)
            f.write("\n\n! Startup Configuration:\n") 
            f.write(startup_config)
        
        connection.disconnect()
        print(f"🟢 Successfully backed up {hostname}")
        return True
        
    except Exception as e:
        print(f"🔴 Failed to backup {device_info['host']}: {str(e)}")
        return False

def automated_backup_system():
    """Automated backup system for all network devices"""
    backup_dir = "config_backups"
    os.makedirs(backup_dir, exist_ok=True)
    
    # Device list
    devices = [
        {
            'device_type': 'cisco_ios',
            'host': 'core-switch.company.com',
            'username': 'admin',
            'password': 'SecurePass123!',
            'secret': 'EnablePass456!',
        },
        # Add more devices...
    ]
    
    print("🚀 Starting automated configuration backup...")
    
    # Backup all devices concurrently
    threads = []
    results = []
    
    for device in devices:
        thread = threading.Thread(
            target=lambda d: results.append(backup_device_config(d, backup_dir)),
            args=(device,)
        )
        threads.append(thread)
        thread.start()
    
    # Wait for all backups to complete
    for thread in threads:
        thread.join()
    
    # 🟡 Archive old backups
    print("🟡 Archiving old backups...")
    success_count = sum(results)
    
    print(f"\n📊 Backup Summary:")
    print(f"✅ Successful: {success_count}")
    print(f"❌ Failed: {len(devices) - success_count}")
    print(f"💾 Location: {backup_dir}")

if __name__ == "__main__":
    automated_backup_system()

2. Network Health Monitoring:

#!/usr/bin/env python3
"""
Network Health Monitoring Script
Color-coded for health status:
🟢 Green - Healthy
🟡 Yellow - Warning
🔴 Red - Critical
🔵 Blue - Information
"""

from netmiko import ConnectHandler
import json
import smtplib
from email.mime.text import MIMEText

def check_device_health(device_info):
    """Check health status of a network device"""
    health_metrics = {}
    
    try:
        connection = ConnectHandler(**device_info)
        connection.enable()
        
        # 🔵 CPU utilization
        cpu_output = connection.send_command('show processes cpu', use_textfsm=True)
        if cpu_output and isinstance(cpu_output, list):
            health_metrics['cpu'] = float(cpu_output[0].get('cpu_5_sec', 0))
        
        # 🔵 Memory utilization  
        memory_output = connection.send_command('show memory statistics', use_textfsm=True)
        if memory_output and isinstance(memory_output, list):
            health_metrics['memory'] = float(memory_output[0].get('memory_util', 0))
        
        # 🔵 Interface errors
        interfaces_output = connection.send_command('show interfaces', use_textfsm=True)
        error_count = 0
        for interface in interfaces_output:
            error_count += int(interface.get('input_errors', 0))
            error_count += int(interface.get('output_errors', 0))
        health_metrics['interface_errors'] = error_count
        
        connection.disconnect()
        
        # Determine health status
        status = "🟢 HEALTHY"
        if health_metrics.get('cpu', 0) > 80:
            status = "🔴 CRITICAL - High CPU"
        elif health_metrics.get('memory', 0) > 90:
            status = "🔴 CRITICAL - High Memory" 
        elif health_metrics.get('interface_errors', 0) > 100:
            status = "🟡 WARNING - Interface Errors"
        
        return {
            'device': device_info['host'],
            'status': status,
            'metrics': health_metrics
        }
        
    except Exception as e:
        return {
            'device': device_info['host'],
            'status': "🔴 CRITICAL - Cannot Connect",
            'error': str(e)
        }

def send_health_alert(health_report):
    """Send email alert for critical devices"""
    critical_devices = [d for d in health_report if 'CRITICAL' in d['status']]
    
    if critical_devices:
        message = "Critical Network Health Alert!\n\n"
        for device in critical_devices:
            message += f"Device: {device['device']}\n"
            message += f"Status: {device['status']}\n"
            if 'metrics' in device:
                message += f"Metrics: {json.dumps(device['metrics'], indent=2)}\n"
            message += "\n"
        
        # Send email (configure with your SMTP settings)
        print("📧 Sending health alert...")
        # Implementation depends on your email setup

def network_health_dashboard():
    """Main health monitoring function"""
    devices = [
        {
            'device_type': 'cisco_ios',
            'host': 'core-switch.company.com',
            'username': 'admin',
            'password': 'SecurePass123!',
        },
        # Add more devices...
    ]
    
    print("🏥 Network Health Check Starting...")
    
    health_report = []
    for device in devices:
        health_status = check_device_health(device)
        health_report.append(health_status)
        print(f"{health_status['status']} - {health_status['device']}")
    
    # Generate report
    print("\n" + "="*50)
    print("📊 NETWORK HEALTH REPORT")
    print("="*50)
    
    for status in health_report:
        print(f"{status['status']} - {status['device']}")
    
    # Send alerts for critical issues
    send_health_alert(health_report)
    
    return health_report

if __name__ == "__main__":
    network_health_dashboard()

Advanced Automation: REST API Integration

Cisco DNA Center API Integration:

#!/usr/bin/env python3
"""
Cisco DNA Center API Integration
Color-coded for API operations:
🟢 Green - Authentication
🔵 Blue - GET requests
🟡 Yellow - POST/PUT requests  
🔴 Red - Error handling
"""

import requests
import json
from requests.auth import HTTPBasicAuth

class DNACenterManager:
    def __init__(self, base_url, username, password):
        self.base_url = base_url
        self.username = username
        self.password = password
        self.token = None
        
    def authenticate(self):
        """🟢 Authenticate with DNA Center"""
        print("🟢 Authenticating with DNA Center...")
        auth_url = f"{self.base_url}/dna/system/api/v1/auth/token"
        
        try:
            response = requests.post(
                auth_url,
                auth=HTTPBasicAuth(self.username, self.password),
                verify=False  # In production, use proper certificate verification
            )
            response.raise_for_status()
            
            self.token = response.json()['Token']
            print("✅ Authentication successful")
            return True
            
        except Exception as e:
            print(f"🔴 Authentication failed: {str(e)}")
            return False
    
    def get_devices(self):
        """🔵 Get list of network devices"""
        if not self.token:
            print("🔴 Not authenticated")
            return None
            
        print("🔵 Fetching device list...")
        devices_url = f"{self.base_url}/dna/intent/api/v1/network-device"
        
        headers = {
            'X-Auth-Token': self.token,
            'Content-Type': 'application/json'
        }
        
        try:
            response = requests.get(devices_url, headers=headers, verify=False)
            response.raise_for_status()
            
            devices = response.json()['response']
            print(f"✅ Found {len(devices)} devices")
            return devices
            
        except Exception as e:
            print(f"🔴 Failed to get devices: {str(e)}")
            return None
    
    def create_vlan(self, vlan_id, vlan_name):
        """🟡 Create new VLAN via API"""
        if not self.token:
            print("🔴 Not authenticated")
            return False
            
        print(f"🟡 Creating VLAN {vlan_id}: {vlan_name}...")
        vlan_url = f"{self.base_url}/dna/intent/api/v1/vlan"
        
        headers = {
            'X-Auth-Token': self.token,
            'Content-Type': 'application/json'
        }
        
        payload = {
            "vlanNumber": vlan_id,
            "vlanName": vlan_name,
            "status": "Active"
        }
        
        try:
            response = requests.post(vlan_url, headers=headers, json=payload, verify=False)
            response.raise_for_status()
            
            print("✅ VLAN created successfully")
            return True
            
        except Exception as e:
            print(f"🔴 Failed to create VLAN: {str(e)}")
            return False

# Usage example
def main():
    dnac = DNACenterManager(
        base_url="https://dnacenter.company.com",
        username="admin",
        password="SecurePass123!"
    )
    
    if dnac.authenticate():
        devices = dnac.get_devices()
        if devices:
            for device in devices[:5]:  # Show first 5 devices
                print(f"📱 {device['hostname']} - {device['platformId']}")
        
        # Create a new VLAN
        dnac.create_vlan(100, "Python-Automated-VLAN")

if __name__ == "__main__":
    main()

---

Ready to Transform Your Network Operations?

Network automation isn't about replacing network engineers - it's about empowering them. By automating repetitive tasks, you free up time for strategic work, reduce human error, and enable your network to scale with business demands.

Stop typing, start automating. Your future self will thank you.

📢 Follow for more network automation insights: LinkedIn Page WhatsApp Channel

Need help starting your automation journey? Contact us for network automation strategy and implementation services!

#NetworkAutomation #Python #CiscoAutomation #NetDevOps #NetworkProgramming #Automation #Cisco

---