pygnmi: gNMI Client for Network Automation

0.8.15 · active · verified Thu Apr 16

pygnmi (current version 0.8.15) is a pure Python gNMI client designed for managing network functions and collecting telemetry data. It provides an interface to interact with gNMI-enabled network devices, supporting operations like Get, Set, and Subscribe. The library is actively maintained with frequent minor releases, typically addressing bug fixes, performance improvements, and adding support for new gNMI features.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to connect to a gNMI target, retrieve its capabilities, and perform a 'Get' operation for a specific path. It includes placeholders for host, port, and credentials, defaulting to common local development values or environment variables. An example for 'Subscribe' is provided but commented out.

import os
from pygnmi.client import gNMIclient

# Environment variables for connection details or default values
GNMI_HOST = os.environ.get('GNMI_HOST', 'localhost')
GNMI_PORT = int(os.environ.get('GNMI_PORT', '50051'))
GNMI_USERNAME = os.environ.get('GNMI_USERNAME', 'admin')
GNMI_PASSWORD = os.environ.get('GNMI_PASSWORD', 'admin')

# Configure the gNMI target
target = {
    'addr': f'{GNMI_HOST}:{GNMI_PORT}',
    'username': GNMI_USERNAME,
    'password': GNMI_PASSWORD,
    'insecure': True,  # Set to False for secure TLS connections and provide certificates
    'encoding': 'JSON_IETF', # Or None to auto-detect based on device capabilities
}

try:
    with gNMIclient(target=target) as c:
        # Get device capabilities
        capabilities = c.capabilities()
        print("\n--- Device Capabilities ---")
        print(capabilities)

        # Example: Get system hostname (path may vary by device/OS)
        # Using a common path, adjust as needed for your device
        hostname_path = ['/system/state/hostname']
        get_response = c.get(path=hostname_path, encoding='JSON_IETF')
        print("\n--- Get Response (Hostname) ---")
        print(get_response)

        # Example: Subscribe to an operational state (uncomment to run)
        # subscription_paths = [
        #    ('/interfaces/interface[name=eth0]/state/oper-status', True)
        # ]
        # print("\n--- Subscribe (Streaming) ---")
        # for update in c.subscribe2(path_tuples=subscription_paths, subscribe_mode='STREAM', encoding='JSON_IETF'):
        #    print(update)
        #    # Add logic to break from loop after a few updates or a timeout

except Exception as e:
    print(f"An error occurred: {e}")
    print("Please ensure the gNMI server is running and accessible with correct credentials.")

view raw JSON →