Python SNMP Monitoring — Deep Dive

Implement production SNMP monitoring in Python with PySNMP — bulk polling, SNMPv3 authentication, trap handling, and scalable device fleet management.

System-level framing

SNMP monitoring at scale means polling hundreds or thousands of network devices every few minutes, processing counter deltas, detecting anomalies, and routing alerts. Python’s PySNMP library provides a complete SNMP implementation — not just a wrapper around C libraries — giving full control over protocol behavior, MIB resolution, and async operation.

This deep dive uses pysnmp-lextudio, the actively maintained fork of PySNMP.

Installation

pip install pysnmp-lextudio

For MIB compilation (converting vendor MIB files to Python):

pip install pysmi-lextudio

Basic GET operation

from pysnmp.hlapi import (
    getCmd, SnmpEngine, CommunityData, UdpTransportTarget,
    ContextData, ObjectType, ObjectIdentity,
)

iterator = getCmd(
    SnmpEngine(),
    CommunityData("public", mpModel=1),  # mpModel=1 for SNMPv2c
    UdpTransportTarget(("192.168.1.1", 161), timeout=5, retries=2),
    ContextData(),
    ObjectType(ObjectIdentity("1.3.6.1.2.1.1.1.0")),  # sysDescr
    ObjectType(ObjectIdentity("1.3.6.1.2.1.1.3.0")),  # sysUpTime
)

error_indication, error_status, error_index, var_binds = next(iterator)

if error_indication:
    print(f"Error: {error_indication}")
elif error_status:
    print(f"SNMP error: {error_status.prettyPrint()} at {error_index}")
else:
    for var_bind in var_binds:
        print(f"{var_bind[0].prettyPrint()} = {var_bind[1].prettyPrint()}")

SNMPv3 with authentication and encryption

from pysnmp.hlapi import (
    getCmd, SnmpEngine, UsmUserData, UdpTransportTarget,
    ContextData, ObjectType, ObjectIdentity,
    usmHMACMD5AuthProtocol, usmDESPrivProtocol,
    usmHMACSHAAuthProtocol, usmAesCfb128Protocol,
)

# SNMPv3 with SHA auth and AES encryption
iterator = getCmd(
    SnmpEngine(),
    UsmUserData(
        "monitoring-user",
        authKey="auth-passphrase-min-8-chars",
        privKey="priv-passphrase-min-8-chars",
        authProtocol=usmHMACSHAAuthProtocol,
        privProtocol=usmAesCfb128Protocol,
    ),
    UdpTransportTarget(("192.168.1.1", 161)),
    ContextData(),
    ObjectType(ObjectIdentity("1.3.6.1.2.1.1.1.0")),
)

error_indication, error_status, error_index, var_binds = next(iterator)

SNMPv3 security levels:

noAuthNoPriv — Username only (not recommended)
authNoPriv — Authentication but no encryption
authPriv — Both authentication and encryption (recommended)

Bulk operations for efficient polling

GET-BULK retrieves many rows at once, critical for interface tables on devices with hundreds of ports:

from pysnmp.hlapi import (
    bulkCmd, SnmpEngine, CommunityData, UdpTransportTarget,
    ContextData, ObjectType, ObjectIdentity,
)

# Walk the interface table in bulk
for error_indication, error_status, error_index, var_binds in bulkCmd(
    SnmpEngine(),
    CommunityData("public", mpModel=1),
    UdpTransportTarget(("192.168.1.1", 161)),
    ContextData(),
    0,   # nonRepeaters
    25,  # maxRepetitions — rows per request
    ObjectType(ObjectIdentity("1.3.6.1.2.1.2.2.1.2")),   # ifDescr
    ObjectType(ObjectIdentity("1.3.6.1.2.1.2.2.1.10")),  # ifInOctets
    ObjectType(ObjectIdentity("1.3.6.1.2.1.2.2.1.16")),  # ifOutOctets
    lexicographicMode=False,
):
    if error_indication:
        print(f"Error: {error_indication}")
        break
    elif error_status:
        print(f"SNMP error: {error_status.prettyPrint()}")
        break
    else:
        for var_bind in var_binds:
            print(f"{var_bind[0].prettyPrint()} = {var_bind[1].prettyPrint()}")

The maxRepetitions parameter controls how many rows are returned per packet. Tuning this value reduces round trips — 25 is a reasonable default; go higher for large tables on reliable networks.

Async polling with asyncio

For polling many devices concurrently:

import asyncio
from pysnmp.hlapi.asyncio import (
    getCmd, SnmpEngine, CommunityData, UdpTransportTarget,
    ContextData, ObjectType, ObjectIdentity,
)

async def poll_device(engine: SnmpEngine, host: str, community: str) -> dict:
    error_indication, error_status, error_index, var_binds = await getCmd(
        engine,
        CommunityData(community, mpModel=1),
        UdpTransportTarget((host, 161), timeout=5, retries=1),
        ContextData(),
        ObjectType(ObjectIdentity("1.3.6.1.2.1.1.3.0")),  # sysUpTime
        ObjectType(ObjectIdentity("1.3.6.1.2.1.1.5.0")),  # sysName
    )

    if error_indication:
        return {"host": host, "error": str(error_indication)}
    elif error_status:
        return {"host": host, "error": error_status.prettyPrint()}
    else:
        return {
            "host": host,
            "uptime": var_binds[0][1].prettyPrint(),
            "name": var_binds[1][1].prettyPrint(),
        }

async def poll_fleet(hosts: list[str], community: str = "public", concurrency: int = 50):
    engine = SnmpEngine()
    semaphore = asyncio.Semaphore(concurrency)

    async def limited(host):
        async with semaphore:
            return await poll_device(engine, host, community)

    results = await asyncio.gather(*[limited(h) for h in hosts])
    return results

# Usage
hosts = [f"192.168.1.{i}" for i in range(1, 255)]
results = asyncio.run(poll_fleet(hosts))
for r in results:
    if "error" not in r:
        print(f"{r['host']}: {r['name']} (up {r['uptime']})")

Counter delta calculation

SNMP counters are cumulative — they increase continuously and eventually wrap. To get rates (bytes per second, packets per second), calculate deltas between polls:

from dataclasses import dataclass
from typing import Optional
import time

@dataclass
class CounterState:
    value: int
    timestamp: float

class DeltaTracker:
    def __init__(self, counter_max: int = 2**32):
        self._state: dict[str, CounterState] = {}
        self._counter_max = counter_max

    def update(self, key: str, value: int) -> Optional[float]:
        now = time.monotonic()
        prev = self._state.get(key)
        self._state[key] = CounterState(value=value, timestamp=now)

        if prev is None:
            return None  # First reading — no delta yet

        elapsed = now - prev.timestamp
        if elapsed == 0:
            return 0.0

        if value >= prev.value:
            delta = value - prev.value
        else:
            # Counter wrapped
            delta = (self._counter_max - prev.value) + value

        return delta / elapsed  # Rate per second

Use 64-bit counters (HC counters, OID 1.3.6.1.2.1.31.1.1.1.6 for ifHCInOctets) on interfaces faster than 100 Mbps to avoid frequent wraps.

SNMP trap receiver

Devices can send unsolicited notifications (traps) when events occur:

from pysnmp.carrier.asyncio.dgram import udp
from pysnmp.entity import engine, config
from pysnmp.entity.rfc3413 import ntfrcv
import asyncio

snmp_engine = engine.SnmpEngine()

# Configure transport — listen on UDP port 162
config.addTransport(
    snmp_engine,
    udp.domainName,
    udp.UdpTransport().openServerMode(("0.0.0.0", 162)),
)

# Configure community for SNMPv2c traps
config.addV1System(snmp_engine, "my-area", "public")

def trap_handler(snmp_engine, state_reference, context_engine_id,
                  context_name, var_binds, cb_ctx):
    print("Trap received:")
    for oid, val in var_binds:
        print(f"  {oid.prettyPrint()} = {val.prettyPrint()}")

ntfrcv.NotificationReceiver(snmp_engine, trap_handler)

print("Listening for traps on port 162...")
snmp_engine.transportDispatcher.jobStarted(1)

try:
    snmp_engine.transportDispatcher.runDispatcher()
except KeyboardInterrupt:
    snmp_engine.transportDispatcher.closeDispatcher()

Common traps to handle: link up/down, authentication failure, cold/warm start, and vendor-specific events.

MIB resolution

Numeric OIDs are hard to read. MIB files map them to human-readable names:

from pysnmp.hlapi import ObjectIdentity
from pysnmp.smi import builder

# Use named OIDs instead of numeric
obj = ObjectIdentity("IF-MIB", "ifDescr", 1)
obj.resolveWithMib(builder.MibBuilder())

For vendor MIBs, compile them with mibdump.py (from pysmi-lextudio):

mibdump.py --mib-source=./vendor-mibs CISCO-PROCESS-MIB

Production architecture

A scalable SNMP monitoring system has these components:

Device inventory — Database of hosts, SNMP credentials, and poll intervals.
Poller — Async Python process that cycles through the inventory, queries each device, and publishes results.
State store — Redis or similar for counter state between poll cycles.
Alert evaluator — Compares metrics to thresholds and generates alerts.
Time-series store — InfluxDB, Prometheus, or similar for historical data and dashboards.

[Inventory DB] → [Async Poller] → [Redis State]
                       ↓
              [Alert Evaluator] → [PagerDuty / Slack]
                       ↓
              [InfluxDB / Prometheus] → [Grafana]

Tradeoffs

Approach	Pros	Cons
PySNMP (pure Python)	Full protocol control, async support, MIB compilation	Slower than C libraries, complex API
easysnmp (net-snmp wrapper)	Fast, simple API	Requires net-snmp C libraries, less flexible
subprocess + snmpwalk	Quick scripts, familiar CLI	Fragile parsing, slow for bulk
Prometheus SNMP Exporter	Integrates with existing Prometheus	Configuration-heavy, less custom logic

Security considerations

SNMPv2c community strings travel in plaintext — treat them like passwords, use SNMPv3 on untrusted networks.
Restrict SNMP access by IP on device ACLs — only allow your monitoring subnet.
Use read-only community strings for monitoring — never give monitoring tools write access.
Rotate SNMPv3 credentials on a schedule, just like SSH keys.
Monitor for unauthorized SNMP scanning — unexpected SNMP traffic may indicate reconnaissance.

One thing to remember: SNMP monitoring is a polling loop — ask devices questions on a schedule, compute deltas from counters, and alert on anomalies. Python’s async capabilities let you scale this pattern from a handful of devices to thousands, but the real expertise is in knowing which OIDs matter and what the numbers mean.

pythonnetworkingmonitoring