pygnmi: gNMI Client for Network Automation
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
-
_InactiveRpcError: <_InactiveRpcError of RPC that terminated with: status = StatusCode.UNAUTHENTICATED details = "Authentication failed"
cause Incorrect username/password, or missing authentication details in the gNMI client configuration.fixVerify the `username` and `password` in the `target` dictionary. Ensure these credentials match what the gNMI server expects. -
_InactiveRpcError: <_InactiveRpcError of RPC that terminated with: status = StatusCode.UNAVAILABLE details = "Connect Failed"
cause The gNMI server is unreachable, incorrect host/port in the target address, or a firewall is blocking the connection.fixCheck the `addr` in the `target` dictionary for correct host and port. Ensure the gNMI server is running and network connectivity exists between the client and server. -
TypeError: 'NoneType' object is not subscriptable
cause Attempting to access elements of a gNMI response that is `None` or empty, often due to an invalid gNMI path or the requested data not existing on the device.fixVerify the gNMI `path` is correct and exists on the target device. Add checks for `None` or empty responses (e.g., `if response and response.notification:`) before processing data. -
TypeError: subscribe() got an unexpected keyword argument 'qos'
cause Using arguments (like `qos`) that are only supported by the `subscribe2()` method with the older `subscribe()` method, or attempting to use a feature not supported by the target device.fixMigrate from `subscribe()` to `subscribe2()` for all telemetry subscriptions, as `subscribe2()` is the recommended and actively developed method. Ensure all arguments are appropriate for `subscribe2()` and supported by your gNMI target.
Warnings
- deprecated The `subscribe()` method is being deprecated in favor of `subscribe2()`. It is strongly recommended to migrate telemetry collection to `subscribe2()` for future compatibility and improved functionality.
- gotcha Prior to v0.8.9, `encoding` had a default value. From v0.8.9, `encoding` defaults to `None` and `capabilities()` is called on connect to auto-detect the supported encoding (with `json` and `json_ietf` preferred). Explicitly setting an encoding might be necessary if auto-detection doesn't pick the desired one or if upgrading from older versions.
- breaking Version 0.8.11 fixed a telemetry break for Juniper devices caused by inconsistencies between `Capabilities()` communicated encodings and actual `Subscribe()` support. If experiencing issues with Juniper telemetry on versions prior to 0.8.11, upgrade immediately.
- gotcha By default, `gNMIclient` uses `insecure=True`, which is suitable for development but not for production. For secure deployments, TLS must be configured.
Install
-
pip install pygnmi
Imports
- gNMIclient
from pygnmi.client import gNMIclient
Quickstart
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.")