Asynchronous SNMP (asynio, v3arch)¶

The asyncio module first appeared in standard library since Python 3.3 (in provisional basis). Its main design feature is that it makes asynchronous code looking like synchronous one thus eliminating “callback hell”.

With asyncio built-in facilities, you could run many SNMP queries in parallel and/or sequentially, interleave SNMP queries with I/O operations with other systems. Other asyncio-compatible Python modules can be easily found as it is a popular programming paradigm.

In most examples approximate analogues of well known Net-SNMP snmp* tools command line options are shown. That may help those readers who, by chance are familiar with Net-SNMP tools, better understanding what example code does.

Here’s a quick example on a simple SNMP GET by high-level API:

with SNMPv1, community ‘public’

over IPv4/UDP

to an Agent at demo.snmplabs.com:161

for an instance of SNMPv2-MIB::sysDescr.0 MIB object

Based on asyncio I/O framework

import asyncio
from pysnmp.hlapi.v3arch.asyncio import *


async def run():
    snmpEngine = SnmpEngine()

    iterator = get_cmd(
        snmpEngine,
        CommunityData("public", mpModel=0),
        await UdpTransportTarget.create(("demo.pysnmp.com", 161)),
        ContextData(),
        ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)),
    )

    errorIndication, errorStatus, errorIndex, varBinds = await iterator

    if errorIndication:
        print(errorIndication)

    elif errorStatus:
        print(
            "{} at {}".format(
                errorStatus.prettyPrint(),
                errorIndex and varBinds[int(errorIndex) - 1][0] or "?",
            )
        )
    else:
        for varBind in varBinds:
            print(" = ".join([x.prettyPrint() for x in varBind]))

    snmpEngine.close_dispatcher()


asyncio.run(run())

To make use of SNMPv3 and USM, the following code performs a series of SNMP GETNEXT operations effectively fetching a table of SNMP variables from SNMP Agent:

with SNMPv3, user ‘usr-none-none’, no authentication, no privacy
over IPv4/UDP
to an Agent at demo.pysnmp.com:161
for all OIDs past SNMPv2-MIB::system
run till end-of-mib condition is reported by Agent
based on asyncio I/O framework

import asyncio
from pysnmp.hlapi.v3arch.asyncio import *


async def run(varBinds):
    snmpEngine = SnmpEngine()
    while True:
        errorIndication, errorStatus, errorIndex, varBindTable = await bulk_cmd(
            snmpEngine,
            UsmUserData("usr-none-none"),
            await UdpTransportTarget.create(("demo.pysnmp.com", 161)),
            ContextData(),
            0,
            50,
            *varBinds,
        )

        if errorIndication:
            print(errorIndication)
            break
        elif errorStatus:
            print(
                f"{errorStatus.prettyPrint()} at {varBinds[int(errorIndex) - 1][0] if errorIndex else '?'}"
            )
        else:
            for varBind in varBindTable:
                print(" = ".join([x.prettyPrint() for x in varBind]))

        varBinds = varBindTable
        if is_end_of_mib(varBinds):
            break
    return


asyncio.run(
    run([ObjectType(ObjectIdentity("TCP-MIB")), ObjectType(ObjectIdentity("IP-MIB"))])
)

More examples on Command Generator API usage follow.

Sending SNMP TRAP’s and INFORM’s is as easy with PySNMP library. The following code sends SNMP TRAP:

SNMPv1
with community name ‘public’
over IPv4/UDP
send TRAP notification
with Generic Trap #1 (warmStart) and Specific Trap 0
with default Uptime
with default Agent Address
with Enterprise OID 1.3.6.1.4.1.20408.4.1.1.2
include managed object information ‘1.3.6.1.2.1.1.1.0’ = ‘my system’

import asyncio
from pysnmp.hlapi.v3arch.asyncio import *


async def run():
    snmpEngine = SnmpEngine()
    errorIndication, errorStatus, errorIndex, varBinds = await send_notification(
        snmpEngine,
        CommunityData("public", mpModel=0),
        await UdpTransportTarget.create(("demo.pysnmp.com", 162)),
        ContextData(),
        "trap",
        NotificationType(ObjectIdentity("1.3.6.1.6.3.1.1.5.2"))
        .load_mibs("SNMPv2-MIB")
        .add_varbinds(
            ("1.3.6.1.6.3.1.1.4.3.0", "1.3.6.1.4.1.20408.4.1.1.2"),
            ("1.3.6.1.2.1.1.1.0", OctetString("my system")),
        ),
    )

    if errorIndication:
        print(errorIndication)

    snmpEngine.close_dispatcher()


asyncio.run(run())

More examples on Notification Originator API usage follow.

More sophisticated or less popular SNMP operations can still be performed with PySNMP through its Native API to Standard SNMP Applications.