Python NETCONF & YANG — Deep Dive

System-level framing

NETCONF and YANG bring software engineering discipline to network management. Instead of screen-scraping CLI output and hoping regex patterns hold across firmware versions, you work with structured XML data validated against a formal schema. Python’s ncclient library makes NETCONF operations feel like database transactions — connect, lock, edit, validate, commit, unlock.

This deep dive covers real implementation patterns for single-device and fleet-wide configuration management.

Installation

pip install ncclient
pip install lxml  # For XML manipulation

Connecting and discovering capabilities

from ncclient import manager

with manager.connect(
    host="192.168.1.1",
    port=830,
    username="admin",
    password="admin",
    hostkey_verify=False,  # Set True in production with known_hosts
    device_params={"name": "default"},
    timeout=30,
) as m:
    # List supported YANG models
    for capability in m.server_capabilities:
        if "module=" in capability:
            print(capability)

The capabilities exchange reveals which YANG models the device supports, including version information. This is critical for multi-vendor environments where the same logical operation requires different models.

Retrieving configuration

from ncclient import manager
from lxml import etree

with manager.connect(
    host="192.168.1.1", port=830,
    username="admin", password="admin",
    hostkey_verify=False,
) as m:
    # Get entire running config
    config = m.get_config(source="running")
    print(etree.tostring(config.data_ele, pretty_print=True).decode())

    # Filtered get — only interfaces
    filter_xml = """
    <filter>
      <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>
    </filter>
    """
    interfaces = m.get_config(source="running", filter=filter_xml)
    print(etree.tostring(interfaces.data_ele, pretty_print=True).decode())

Subtree filters reduce response size dramatically. On a router with thousands of configuration lines, fetching only the interface section saves bandwidth and parsing time.

Editing configuration — transactional workflow

from ncclient import manager

config_payload = """
<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">
  <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
    <interface>
      <name>GigabitEthernet0/0/0</name>
      <description>Uplink to core switch</description>
      <enabled>true</enabled>
    </interface>
  </interfaces>
</config>
"""

with manager.connect(
    host="192.168.1.1", port=830,
    username="admin", password="admin",
    hostkey_verify=False,
) as m:
    # Full transactional workflow
    with m.locked(target="candidate"):
        m.edit_config(
            target="candidate",
            config=config_payload,
            default_operation="merge",
        )

        # Validate before committing
        m.validate(source="candidate")

        # Commit — atomically applies all changes
        m.commit()

The locked context manager handles lock/unlock automatically, even on exceptions. The default_operation parameter controls merge behavior:

  • merge — Combine with existing config (default).
  • replace — Replace the matching section.
  • none — Only apply explicitly tagged operations.

Delete and replace operations

# Delete an interface
delete_payload = """
<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">
  <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
    <interface xc:operation="delete">
      <name>Loopback99</name>
    </interface>
  </interfaces>
</config>
"""

# Replace entire interface config
replace_payload = """
<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">
  <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"
              xc:operation="replace">
    <interface>
      <name>GigabitEthernet0/0/0</name>
      <description>New description</description>
      <enabled>true</enabled>
    </interface>
  </interfaces>
</config>
"""

Confirmed commit with rollback timer

Some devices support confirmed commits — the change auto-reverts if not confirmed within a timeout:

with manager.connect(...) as m:
    with m.locked(target="candidate"):
        m.edit_config(target="candidate", config=config_payload)

        # Commit with 120-second confirmation window
        m.commit(confirmed=True, timeout="120")

        # Verify the change works (e.g., ping test)
        # ...

        # Confirm — makes the change permanent
        m.commit()

        # If you do NOT confirm within 120 seconds,
        # the device automatically rolls back

This is invaluable for remote configuration changes. If a network change cuts your SSH connection, the device recovers automatically.

Retrieving operational state

NETCONF separates configuration from operational state. Use get() for operational data:

with manager.connect(...) as m:
    # Get interface statistics (operational data)
    filter_xml = """
    <filter>
      <interfaces-state xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>
    </filter>
    """
    state = m.get(filter=filter_xml)
    root = etree.fromstring(state.data_xml.encode())

    ns = {"if": "urn:ietf:params:xml:ns:yang:ietf-interfaces"}
    for iface in root.findall(".//if:interface", ns):
        name = iface.find("if:name", ns).text
        stats = iface.find("if:statistics", ns)
        if stats is not None:
            in_octets = stats.find("if:in-octets", ns)
            out_octets = stats.find("if:out-octets", ns)
            print(f"{name}: in={in_octets.text}, out={out_octets.text}")

Multi-vendor abstraction

Different vendors use different YANG models for the same concept. Abstract this:

from dataclasses import dataclass
from abc import ABC, abstractmethod

@dataclass
class InterfaceConfig:
    name: str
    description: str
    enabled: bool
    ip_address: str | None = None
    prefix_length: int | None = None

class NetconfDriver(ABC):
    @abstractmethod
    def build_interface_config(self, iface: InterfaceConfig) -> str:
        """Return XML config payload for the vendor."""

class IETFDriver(NetconfDriver):
    def build_interface_config(self, iface: InterfaceConfig) -> str:
        return f"""
        <config>
          <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
            <interface>
              <name>{iface.name}</name>
              <description>{iface.description}</description>
              <enabled>{'true' if iface.enabled else 'false'}</enabled>
            </interface>
          </interfaces>
        </config>
        """

class JunosDriver(NetconfDriver):
    def build_interface_config(self, iface: InterfaceConfig) -> str:
        return f"""
        <config>
          <configuration xmlns="http://xml.juniper.net/xnm/1.1/xnm">
            <interfaces>
              <interface>
                <name>{iface.name}</name>
                <description>{iface.description}</description>
              </interface>
            </interfaces>
          </configuration>
        </config>
        """

def get_driver(vendor: str) -> NetconfDriver:
    drivers = {"ietf": IETFDriver, "junos": JunosDriver}
    return drivers[vendor]()

Fleet configuration management

Apply config changes across multiple devices with error handling:

import concurrent.futures
from ncclient import manager

def configure_device(host: str, config_xml: str, credentials: dict) -> dict:
    try:
        with manager.connect(
            host=host, port=830,
            username=credentials["username"],
            password=credentials["password"],
            hostkey_verify=False,
            timeout=30,
        ) as m:
            with m.locked(target="candidate"):
                m.edit_config(target="candidate", config=config_xml)
                m.validate(source="candidate")
                m.commit()
            return {"host": host, "status": "success"}
    except Exception as e:
        return {"host": host, "status": "failed", "error": str(e)}

def configure_fleet(
    hosts: list[str], config_xml: str, credentials: dict, max_workers: int = 10
) -> list[dict]:
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(configure_device, host, config_xml, credentials): host
            for host in hosts
        }
        for future in concurrent.futures.as_completed(futures):
            results.append(future.result())
    return results

YANG model exploration with pyang

Before writing NETCONF payloads, understand the model:

# Install pyang
pip install pyang

# Display a YANG model as a tree
pyang -f tree ietf-interfaces.yang

# Output:
# module: ietf-interfaces
#   +--rw interfaces
#      +--rw interface* [name]
#         +--rw name          string
#         +--rw description?  string
#         +--rw type          identityref
#         +--rw enabled?      boolean

The tree format shows which fields are read-write (rw), read-only (ro), mandatory, optional (?), and list keys ([name]).

Error handling patterns

from ncclient.operations import RPCError
from ncclient.transport import TransportError, SSHError

try:
    with manager.connect(...) as m:
        m.edit_config(target="candidate", config=bad_xml)
        m.commit()
except RPCError as e:
    # NETCONF-level error (invalid config, constraint violation)
    print(f"RPC error: {e.message}")
    print(f"Error type: {e.type}")
    print(f"Error tag: {e.tag}")
    print(f"Error severity: {e.severity}")
    # The candidate datastore may need to be discarded
    m.discard_changes()
except SSHError as e:
    # SSH connection failure
    print(f"SSH error: {e}")
except TransportError as e:
    # Low-level transport issue
    print(f"Transport error: {e}")

Tradeoffs

ApproachProsCons
NETCONF + ncclientTransactional, validated, structuredXML verbosity, vendor model differences
CLI over SSH (Netmiko)Works with any device that has a CLIScreen-scraping, no validation, fragile
RESTCONFJSON-friendly REST API over YANG modelsLess widely supported than NETCONF
gNMIHigh-performance streaming telemetryNewer, smaller vendor support
Ansible network modulesDeclarative, version-controlledSlower, less flexible for custom logic

Security considerations

  1. Use SSH host key verification in production — hostkey_verify=True with a managed known_hosts file.
  2. Store credentials in a vault (HashiCorp Vault, AWS Secrets Manager), not in scripts.
  3. Use role-based access control on devices — monitoring accounts should not have config write access.
  4. Audit all changes — Log every edit-config and commit with timestamps, user, and payload hash.
  5. Test on lab devices first — A bad NETCONF commit on a core router can cause an outage.

One thing to remember: NETCONF brings database-level transactional guarantees to network configuration. Python’s ncclient makes it practical — but the real power is in the workflow: lock, edit, validate, commit, verify. That discipline is what separates reliable network automation from scripted hope.

pythonnetworkingautomation

See Also

  • Python Dns Resolver Understand how Python translates website names into addresses, like a phone book for the entire internet.
  • Python Dpkt Packet Parsing Understand how Python reads and decodes captured network traffic, like opening envelopes to see what is inside each message.
  • Python Ftp Sftp Transfers Understand how Python moves files between computers over a network, like a digital delivery truck with a locked or unlocked cargo door.
  • Python Impacket Security Tools Understand how Python speaks the secret languages of Windows networks, helping security teams find weaknesses before attackers do.
  • Python Pcap Analysis Understand how Python reads recordings of network traffic, like playing back security camera footage to see what happened on your network.