Skip to content

Integrating Network Automation with ITSM and Change Management

Integrating Network Automation with ITSM and Change Management


This post is part of our ongoing series on network automation best practices, grounded in the PRIME Framework and PRIME Philosophy.

Why This Blog Exists

Automation without change management is a compliance risk. This post explains why ITSM integration matters, how to connect your automation to ServiceNow, Jira, and other workflows, and how the PRIME Framework ensures auditability and ownership.


๐Ÿšฆ PRIME Philosophy: Measurability and Ownership

  • Measurability: Track every change, who made it, and why
  • Ownership: Your team controls the workflow, not a vendor
  • Transparency: Document approvals and outcomes
  • Safety: Prevent unauthorized or risky changes
  • Empowerment: Make compliance easy, not a burden

Why ITSM Integration Matters

  • Auditability: Prove who authorized what, when, and why
  • Compliance: Meet regulatory requirements (SOX, PCI, HIPAA)
  • Change control: Prevent unauthorized or risky changes
  • Traceability: Link configuration changes to business requirements and incidents
  • Rollback capability: If something breaks, know exactly what changed and revert it
  • Communication: Notify teams of planned and completed changes
  • Measurability: Track change success rates, duration, and business impact

Patterns for ITSM Integration

1. ServiceNow API Integration (Comprehensive Example)

1. ServiceNow API Integration (Comprehensive Example)

  • Create, update, and close change tickets from automation scripts
  • Query ticket status and approvals before making changes
  • Link automation runs to change records for full traceability
  • Example: Python requests to ServiceNow REST API with error handling and status checks
import requests
import os
import time
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

class ServiceNowChangeManager:
    """Manage change requests in ServiceNow."""

    def __init__(self):
        self.instance = os.environ['SNOW_INSTANCE']  # e.g., 'dev12345'
        self.user = os.environ['SNOW_USER']
        self.password = os.environ['SNOW_PASSWORD']
        self.base_url = f"https://{self.instance}.service-now.com"
        self.headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        }
        self.auth = (self.user, self.password)

    def create_change(self, short_description, description, risk_level='medium'):
        """Create a new change request in ServiceNow."""
        endpoint = f"{self.base_url}/api/now/table/change_request"

        data = {
            'short_description': short_description,
            'description': description,
            'category': 'Network Configuration',
            'subcategory': 'Automation',
            'assignment_group': 'Network Team',
            'priority': '3',  # 1=Critical, 3=Medium, 5=Low
            'risk': risk_level,  # low, medium, high
            'type': 'Standard',
            'cmdb_ci': 'Network Infrastructure',
            'chg_model': 'Standard Change',
        }

        try:
            response = requests.post(
                endpoint,
                headers=self.headers,
                auth=self.auth,
                json=data,
                timeout=30
            )
            response.raise_for_status()

            change = response.json()['result']
            logger.info(f"โœ“ Created change request: {change['number']}")
            return {
                'change_id': change['number'],
                'sys_id': change['sys_id'],
                'url': f"{self.base_url}/nav_to.do?uri=change_request.do?sys_id={change['sys_id']}"
            }
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to create change: {e}")
            raise

    def get_change_status(self, change_id):
        """Get current status of a change request."""
        endpoint = f"{self.base_url}/api/now/table/change_request?sysparm_query=number={change_id}"

        try:
            response = requests.get(
                endpoint,
                headers=self.headers,
                auth=self.auth,
                timeout=30
            )
            response.raise_for_status()

            changes = response.json()['result']
            if not changes:
                raise Exception(f"Change not found: {change_id}")

            change = changes[0]
            return {
                'number': change['number'],
                'state': change['state'],  # -5=Draft, -4=Pending Review, -3=Pending Approval, 0=Open, 1=Work in Progress, 3=Closed, etc.
                'approval': change.get('approval', 'not_requested'),  # not_requested, requested, approved, rejected
                'status': change.get('status', ''),
                'assignee': change.get('assigned_to', {}).get('display_value', 'Unassigned'),
            }
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to get change status: {e}")
            raise

    def wait_for_approval(self, change_id, max_wait_seconds=3600):
        """Wait for change to be approved. Blocks until approved or timeout."""
        logger.info(f"Waiting for approval of change {change_id}...")
        start_time = time.time()

        while time.time() - start_time < max_wait_seconds:
            status = self.get_change_status(change_id)

            if status['approval'] == 'approved':
                logger.info(f"โœ“ Change {change_id} approved!")
                return True
            elif status['approval'] == 'rejected':
                raise Exception(f"Change {change_id} rejected by {status['assignee']}")

            logger.debug(f"Change {change_id} still pending ({time.time() - start_time:.0f}s)")
            time.sleep(10)  # Check every 10 seconds

        raise Exception(f"Change approval timeout after {max_wait_seconds}s")

    def update_change_notes(self, change_id, notes):
        """Add work notes to a change (visible to support staff)."""
        endpoint = f"{self.base_url}/api/now/table/change_request?sysparm_query=number={change_id}"

        data = {
            'work_notes': f"[{datetime.now()}] {notes}"
        }

        try:
            response = requests.patch(
                endpoint,
                headers=self.headers,
                auth=self.auth,
                json=data,
                timeout=30
            )
            response.raise_for_status()
            logger.info(f"โœ“ Updated work notes for {change_id}")
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to update change notes: {e}")

    def close_change(self, change_id, success=True, close_notes=""):
        """Close a change request."""
        endpoint = f"{self.base_url}/api/now/table/change_request?sysparm_query=number={change_id}"

        data = {
            'state': 'closed',
            'close_code': 'successful' if success else 'unsuccessful',
            'close_notes': close_notes or f"Deployment completed: {'Success' if success else 'Failed'}",
            'actual_end_date': datetime.now().isoformat(),
        }

        try:
            response = requests.patch(
                endpoint,
                headers=self.headers,
                auth=self.auth,
                json=data,
                timeout=30
            )
            response.raise_for_status()
            logger.info(f"โœ“ Closed change {change_id}")
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to close change: {e}")
            raise

    def link_configuration_item(self, change_id, cmdb_ci_id):
        """Link a CI (device, service, app) to the change."""
        endpoint = f"{self.base_url}/api/now/table/change_request/{change_id}"

        data = {
            'cmdb_ci': cmdb_ci_id
        }

        try:
            response = requests.patch(
                endpoint,
                headers=self.headers,
                auth=self.auth,
                json=data,
                timeout=30
            )
            response.raise_for_status()
            logger.info(f"โœ“ Linked CI to change {change_id}")
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to link CI: {e}")

# Usage in automation workflow
def deploy_with_change_tracking(devices, config_changes):
    """Deploy config changes with full ServiceNow integration."""
    snow = ServiceNowChangeManager()

    # Step 1: Create change ticket
    change = snow.create_change(
        short_description="Deploy VLAN config to network devices",
        description=f"Updating VLANs on devices: {', '.join(devices)}",
        risk_level='medium'
    )
    change_id = change['change_id']

    try:
        # Step 2: Wait for approval (blocks until approved)
        snow.wait_for_approval(change_id, max_wait_seconds=1800)  # 30 min timeout

        # Step 3: Update with pre-deployment notes
        snow.update_change_notes(change_id, "Pre-deployment validation completed. Starting deployment...")

        # Step 4: Deploy config
        for device in devices:
            logger.info(f"Deploying to {device}...")
            deploy_config(device, config_changes)

        # Step 5: Post-deployment validation
        snow.update_change_notes(change_id, "Deployment completed. Running validation...")
        validate_deployment(devices)

        # Step 6: Close change successfully
        snow.close_change(change_id, success=True, close_notes="All devices validated successfully")
        logger.info("โœ“ Deployment completed with change tracking")

    except Exception as e:
        logger.error(f"Deployment failed: {e}")
        snow.close_change(change_id, success=False, close_notes=f"Error: {str(e)}")
        raise

Key Features: - Atomic operations โ€” Create change first, then deploy - Blocking approval โ€” Wait until approved before proceeding - Traceability โ€” Link devices and configs to change record - Error handling โ€” Close change with success/failure status - Audit trail โ€” All actions logged in work notes


2. Jira and Custom Workflows

  • Log changes as Jira issues
  • Use webhooks for automation triggers
  • Automate status transitions and comments from scripts
  • Integrate with CI/CD for pre- and post-change validation
from jira import JIRA
import os

class JiraChangeTracker:
    """Track automation changes in Jira."""

    def __init__(self):
        self.jira = JIRA(
            server=os.environ['JIRA_SERVER'],
            basic_auth=(os.environ['JIRA_USER'], os.environ['JIRA_KEY'])
        )

    def create_deployment_ticket(self, summary, description, devices):
        """Create a Jira ticket for the deployment."""
        issue_dict = {
            'project': {'key': 'OPS'},
            'issuetype': {'name': 'Task'},
            'summary': summary,
            'description': description,
            'customfield_devices': devices,  # Custom field for tracking affected devices
        }

        new_issue = self.jira.create_issue(fields=issue_dict)
        return new_issue.key

    def transition_to_in_progress(self, issue_key):
        """Move ticket to 'In Progress'."""
        transitions = self.jira.transitions(issue_key)
        in_progress = next(t for t in transitions['transitions'] if t['name'] == 'In Progress')
        self.jira.transition_issue(issue_key, in_progress['id'])

    def add_comment(self, issue_key, comment):
        """Add a comment to the ticket."""
        self.jira.add_comment(issue_key, comment)

    def transition_to_done(self, issue_key, success=True):
        """Move ticket to 'Done' or 'Failed'."""
        transitions = self.jira.transitions(issue_key)
        status = 'Done' if success else 'Blocked'
        target = next(t for t in transitions['transitions'] if t['name'] == status)
        self.jira.transition_issue(issue_key, target['id'])

# Usage
jira = JiraChangeTracker()
issue_key = jira.create_deployment_ticket(
    "Deploy VLAN configuration",
    "Update VLANs on core and access switches",
    "switch-01, switch-02, switch-03"
)
jira.transition_to_in_progress(issue_key)

try:
    deploy_vlan_config()
    jira.add_comment(issue_key, "Deployment succeeded!")
    jira.transition_to_done(issue_key, success=True)
except Exception as e:
    jira.add_comment(issue_key, f"Deployment failed: {e}")
    jira.transition_to_done(issue_key, success=False)

3. Change Logging & Compliance Reporting

  • Log every change to a central database, SIEM, or data lake
  • Generate compliance and audit reports automatically
  • Use structured logs and unique run IDs for traceability
  • Integrate with dashboards for real-time compliance monitoring
import json
import logging
from datetime import datetime
from elasticsearch import Elasticsearch

class ChangeAuditLog:
    """Log all changes to Elasticsearch for compliance."""

    def __init__(self):
        self.es = Elasticsearch(['http://elasticsearch:9200'])
        self.index = 'network-automation-changes'

    def log_change(self, run_id, device, change_type, change_date, status, details):
        """Log a change event."""
        doc = {
            'timestamp': datetime.now().isoformat(),
            'run_id': run_id,
            'device': device,
            'change_type': change_type,  # e.g., 'vlan_added', 'route_removed'
            'change_date': change_date,
            'status': status,  # 'success', 'failed', 'rolled_back'
            'details': details,
            'user': os.environ.get('USER', 'automation'),
            'source_ip': get_source_ip(),
        }

        self.es.index(index=self.index, body=doc)

    def generate_compliance_report(self, start_date, end_date):
        """Generate a compliance report."""
        query = {
            'range': {
                'timestamp': {
                    'gte': start_date,
                    'lte': end_date
                }
            }
        }

        results = self.es.search(index=self.index, body={'query': query, 'size': 10000})

        # Aggregate by device, change type, etc.
        report = {
            'total_changes': results['hits']['total']['value'],
            'successful': sum(1 for hit in results['hits']['hits'] if hit['_source']['status'] == 'success'),
            'failed': sum(1 for hit in results['hits']['hits'] if hit['_source']['status'] == 'failed'),
        }

        return report

PRIME in Action: Automated Compliance

  • Automate ticket creation, status checks, and closure
  • Link automation runs to change records for full traceability
  • Alert on unauthorized or out-of-process changes
  • Generate audit reports and compliance dashboards automatically
  • Integrate ITSM checks into CI/CD and pre-change validation


Summary: Blog Takeaways

  • ITSM integration is essential for auditability and compliance
  • Use APIs and logging to connect automation to change management
  • PRIME principles make compliance sustainable and empowering
  • Automate change tracking, approvals, and reporting for production-grade safety
  • Integrate ITSM with CI/CD, observability, and incident response

๐Ÿ“ฃ Want More?