#compsci #python #software 

PySNMP is a cross-platform, pure-Python [[SNMP]] engine implementation.
Features fully-functional SNMP engine for Agent/Manager/Proxy roles, talking SNMP v1/v2c/v3 protocol versions over IPv4/IPv6 and other network transports.

Current stable version is 7.1.22

## History
![[Pasted image 20250801002403.png]]
![[Pasted image 20250801002611.png]]
![[Pasted image 20250801002620.png]]
![[Pasted image 20250801002630.png]]
## Quick start
### Fetching SNMP variable
```python
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())
```

(requires [[asyncio module]])

![[Pasted image 20250731203257.png]]

### Send SNMP TRAP
```python
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())
```


## PySNMP architecture
![[Pasted image 20250731204153.png]]

![[Pasted image 20250731204220.png]]
PySNMP inner components:

- SNMP Engine is a central, umbrella object that controls the other components of the SNMP system. Typical user application has a single instance of SNMP Engine class possibly shared by many SNMP Applications of all kinds. As the other linked-in components tend to buildup various configuration and housekeeping information in runtime, SNMP Engine object appears to be expensive to configure to a usable state.
    
- Transport subsystem is used for sending SNMP messages to and accepting them from network. The I/O subsystem consists of an abstract Dispatcher and one or more abstract Transport classes. Concrete Dispatcher implementation is I/O method-specific, consider BSD sockets for example. Concrete Transport classes are transport domain-specific. SNMP frequently uses UDP Transport but others are also possible. Transport Dispatcher interfaces are mostly used by Message And PDU Dispatcher. However, when using the SNMPv1/v2c-native API (the lowest-level one), these interfaces would be invoked directly.
    
- Message and PDU Dispatcher is the locus of SNMP message processing activities. Its main responsibilities include dispatching PDUs from SNMP Applications through various subsystems all the way down to Transport Dispatcher, and passing SNMP messages coming from network up to SNMP Applications. It maintains logical connection with Management Instrumentation Controller which carries out operations on Managed Objects, here for the purpose of LCD access.
    
- Message Processing Modules handle message-level protocol operations for present and possibly future versions of SNMP protocol. Most importantly, these include message parsing/building and possibly invoking security services whenever required.
    
- Message Security Modules perform message authentication and/or encryption. As of this writing, User-Based (for v3) and Community (for v1/2c) modules are implemented in PySNMP. All Security Modules share standard API used by Message Processing subsystem.
    
- Access Control subsystem uses LCD information to authorize remote access to Managed Objects. This is used when running in agent role.
    
- A collection of MIB modules and objects that are used by SNMP engine for keeping its configuration and operational statistics. They are collectively called Local Configuration Datastore (LCD).


## Common operations with high-level API (v3arch)
![[Pasted image 20250731204557.png]]
### Creating SNMP Engine
![[Pasted image 20250731204621.png]]

### Making SNMP Query
![[Pasted image 20250731204750.png]]

There's a choice of 3 SNMP protocol versions, to employ SNMPv1/v2c, we pass properly initialized instance of `CommunityData` class. For SNMPv3, we pass `UsmUserData` class instance.

![[Pasted image 20250731204913.png]]
![[Pasted image 20250731204927.png]]
PySNMP supports MD5 and SHA authentication altgorithms, as well as DES, AES128/192/256 and 3DES encryption (privacy) algorithms

![[Pasted image 20250731211956.png]]
![[Pasted image 20250731212146.png]]\

![[Pasted image 20250731212303.png]]
![[Pasted image 20250731212326.png]]
![[Pasted image 20250731212334.png]]
for tables, simply use OID.%index% where %index% is the index of the value you're looking for
![[Pasted image 20250731212417.png]]
### SNMP command operations
![[Pasted image 20250731212615.png]]
![[Pasted image 20250731212625.png]]![[Pasted image 20250731212639.png]]
![[Pasted image 20250731212644.png]]

### Sending SNMP notifications
Managed entity could send unsolicited messages to the managing entity. That is called notification in SNMP. Notifications help reduce polling, what may become a problem for large networks.

![[Pasted image 20250731213014.png]]
![[Pasted image 20250731213025.png]]
```PYTHON
>>> from pysnmp.hlapi.v3arch.asyncio import *
>>>
>>> g = await send_notification(SnmpEngine(),
...                      CommunityData('public'),
...                      await UdpTransportTarget.create(('demo.pysnmp.com', 162)),
...                      ContextData(),
...                      'trap',
...                      NotificationType(ObjectIdentity('IF-MIB', 'linkUp'), instanceIndex=(123,))
... )
>>> g
(None, 0, 0, [])
```

```python
>>> from pysnmp.hlapi.v3arch.asyncio import *
>>>
>>> g = await send_notification(SnmpEngine(),
...                      CommunityData('public'),
...                      await UdpTransportTarget.create(('demo.pysnmp.com', 162)),
...                      ContextData(),
...                      'inform',
...                      NotificationType(ObjectIdentity('IF-MIB', 'linkUp'), instanceIndex=(123,))
... )
>>> g
(None, 0, 0, [ObjectType(ObjectIdentity('1.3.6.1.2.1.1.3.0'), TimeTicks(0)), ObjectType(ObjectIdentity('1.3.6.1.6.3.1.1.4.1.0'), ObjectIdentity('1.3.6.1.6.3.1.1.5.4')), ObjectType(ObjectName('1.3.6.1.2.1.2.2.1.1.123'), Null('')), ObjectType(ObjectIdentity('1.3.6.1.2.1.2.2.1.7.123'), Null('')), ObjectType(ObjectIdentity('1.3.6.1.2.1.2.2.1.8.123'), Null(''))])
```

As you can see, the actual values for expanded MIB objects are NULLs. That’s because in these examples our simple scripts do not have access to those MIB objects. We can supply that missing information by passing _NotificationType_ a dictionary-like object that maps MIB object OIDs to current values:
```python
>>> from pysnmp.hlapi.v3arch.asyncio import *
>>>
>>> mib = {ObjectIdentifier('1.3.6.1.2.1.2.2.1.1.123'): 123,
...        ObjectIdentifier('1.3.6.1.2.1.2.2.1.7.123'): 'testing',
...        ObjectIdentifier('1.3.6.1.2.1.2.2.1.8.123'): 'up'}
>>>
>>> g = await send_notification(SnmpEngine(),
...                      CommunityData('public'),
...                      await UdpTransportTarget.create(('demo.pysnmp.com', 162)),
...                      ContextData(),
...                      'inform',
...                      NotificationType(ObjectIdentity('IF-MIB', 'linkUp'), instanceIndex=(123,), objects=mib)
... )
>>> g
(None, 0, 0, [ObjectType(ObjectIdentity('1.3.6.1.2.1.1.3.0'), TimeTicks(0)), ObjectType(ObjectIdentity('1.3.6.1.6.3.1.1.4.1.0'), ObjectIdentity('1.3.6.1.6.3.1.1.5.4')), ObjectType(ObjectName('1.3.6.1.2.1.2.2.1.1.123'), InterfaceIndex(123)), ObjectType(ObjectIdentity('1.3.6.1.2.1.2.2.1.7.123'), Integer(3)), ObjectType(ObjectIdentity('1.3.6.1.2.1.2.2.1.8.123'), Integer(1))])
```

![[Pasted image 20250731213320.png]]

## API Reference
### High-level v3arch asyncio
#### Engine
![[Pasted image 20250731224650.png]]

#### Transport Config
![[Pasted image 20250731224900.png]]
![[Pasted image 20250731224925.png]]

#### Auth
![[Pasted image 20250731225048.png]]
![[Pasted image 20250731225058.png]]
![[Pasted image 20250731225109.png]]


![[Pasted image 20250731225158.png]]
![[Pasted image 20250731225210.png]]
![[Pasted image 20250731225224.png]]
![[Pasted image 20250731225248.png]]![[Pasted image 20250731230523.png]]
![[Pasted image 20250731230541.png]]

#### Context
![[Pasted image 20250731230912.png]]
![[Pasted image 20250731231017.png]]

![[Pasted image 20250731231029.png]]
#### Operations
![[Pasted image 20250731231105.png]]
![[Pasted image 20250731231113.png]]


![[Pasted image 20250731231123.png]]
![[Pasted image 20250731231131.png]]


![[Pasted image 20250731231158.png]]
![[Pasted image 20250731231212.png]]


![[Pasted image 20250731231244.png]]
![[Pasted image 20250731231252.png]]

```python
>>> import asyncio
>>> from pysnmp.hlapi.v3arch.asyncio import *
>>>
>>> async def run():
...     errorIndication, errorStatus, errorIndex, varBinds = await bulk_cmd(
...         SnmpEngine(),
...         CommunityData('public'),
...         await UdpTransportTarget.create(('demo.pysnmp.com', 161)),
...         ContextData(),
...         0, 2,
...         ObjectType(ObjectIdentity('SNMPv2-MIB', 'system'))
...     )
...     print(errorIndication, errorStatus, errorIndex, varBinds)
>>>
>>> asyncio.run(run())
(None, 0, 0, [[ObjectType(ObjectIdentity(ObjectName('1.3.6.1.2.1.1.1.0')), DisplayString('SunOS zeus.pysnmp.com 4.1.3_U1 1 sun4m'))], [ObjectType(ObjectIdentity(ObjectName('1.3.6.1.2.1.1.2.0')), ObjectIdentifier('1.3.6.1.4.1.424242.1.1'))]])
>>>
```


![[Pasted image 20250731231728.png]]
![[Pasted image 20250731231741.png]]
EXAMPLES:
```python
>>> from pysnmp.hlapi.v3arch.asyncio import *
>>> objects = walk_cmd(SnmpEngine(),
...             CommunityData('public'),
...             await UdpTransportTarget.create(('demo.pysnmp.com', 161)),
...             ContextData(),
...             ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr')))
... g = [item async for item in objects]
>>> next(g)
(None, 0, 0, [ObjectType(ObjectIdentity(ObjectName('1.3.6.1.2.1.1.1.0')), DisplayString('SunOS zeus.pysnmp.com 4.1.3_U1 1 sun4m'))])
>>> g.send( [ ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets')) ] )
(None, 0, 0, [(ObjectName('1.3.6.1.2.1.2.2.1.10.1'), Counter32(284817787))])
```



![[Pasted image 20250731232058.png]]
![[Pasted image 20250731232109.png]]
EXAMPLES:
```python
>>> from pysnmp.hlapi.v3arch.asyncio import *
>>> objects = bulk_walk_cmd(SnmpEngine(),
...             CommunityData('public'),
...             await UdpTransportTarget.create(('demo.pysnmp.com', 161)),
...             ContextData(),
...             0, 25,
...             ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr')))
... g = [item async for item in objects]
>>> next(g)
(None, 0, 0, [ObjectType(ObjectIdentity(ObjectName('1.3.6.1.2.1.1.1.0')), DisplayString('SunOS zeus.pysnmp.com 4.1.3_U1 1 sun4m'))])
>>> g.send( [ ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets')) ] )
(None, 0, 0, [(ObjectName('1.3.6.1.2.1.2.2.1.10.1'), Counter32(284817787))])
```


![[Pasted image 20250731232202.png]]
![[Pasted image 20250731232240.png]]
![[Pasted image 20250731232303.png]]

### High-level v1arch asyncio
Less complex than v3arch asyncio API for a simple and more straightforward API for users who only need SNMPv1/v2c
#### Operations
The same as v3arch's one, except instead of SnmpEngine() it's SnmpDispatcher()

#### Transport config
![[Pasted image 20250731233326.png]]

#### SNMP Dispatcher
![[Pasted image 20250731233351.png]]
#### Auth
![[Pasted image 20250731233809.png]]

#### MIB stuff
![[Pasted image 20250731233832.png]]
![[Pasted image 20250731233839.png]]
![[Pasted image 20250731233846.png]]
![[Pasted image 20250731233855.png]]

![[Pasted image 20250731233911.png]]
![[Pasted image 20250731233927.png]]
![[Pasted image 20250731233947.png]]


![[Pasted image 20250731234448.png]]
![[Pasted image 20250731234501.png]]
Also supports add_asn_mib_source, add_mib_source and load_mibs



![[Pasted image 20250731234927.png]]
![[Pasted image 20250731234937.png]]
Also supports mib source chicanery

#### SNMP base types
Null, Integer32, Integer, OctetString, IpAddress, ObjectIdentifier, Counter32, Gauge32, TimeTicks, etc


## Samples
GET:
```python
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())
```

All OIDs bulk past SNMPv2-MIB::system
```python
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"))])
)
```

![[Pasted image 20250801000909.png]]
![[Pasted image 20250801000919.png]]

![[Pasted image 20250801001141.png]]
Concurrent queries:
```python
import asyncio
from pysnmp.hlapi.v3arch.asyncio import *


async def getone(snmpEngine, hostname):
    errorIndication, errorStatus, errorIndex, varBinds = await get_cmd(
        snmpEngine,
        CommunityData("public"),
        await UdpTransportTarget.create(hostname),
        ContextData(),
        ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)),
    )

    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]))


async def main():
    snmpEngine = SnmpEngine()
    await asyncio.gather(
        getone(snmpEngine, ("demo.pysnmp.com", 161)),
        getone(snmpEngine, ("demo.pysnmp.com", 161)),
        getone(snmpEngine, ("demo.pysnmp.com", 161)),
    )


asyncio.run(main())
```

Sequential queries:
```python
import asyncio
from pysnmp.hlapi.v3arch.asyncio import *


async def getone(snmpEngine, hostname):
    errorIndication, errorStatus, errorIndex, varBinds = await get_cmd(
        snmpEngine,
        CommunityData("public"),
        await UdpTransportTarget.create(hostname),
        ContextData(),
        ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)),
    )

    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]))


async def getall(snmpEngine, hostnames):
    for hostname in hostnames:
        await getone(snmpEngine, hostname)


snmpEngine = SnmpEngine()

asyncio.run(
    getall(
        snmpEngine,
        [
            ("demo.pysnmp.com", 161),
            ("demo.pysnmp.com", 161),
            ("demo.pysnmp.com", 161),
        ],
    )
)
```


![[Pasted image 20250801001233.png]]
```python
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())
```


Concurrent traps:
```python
import asyncio
from pysnmp.hlapi.v3arch.asyncio import *


async def run():
    # List of targets in the following format:
    # ( ( authData, transportTarget ), ... )
    TARGETS = (
        # 1-st target (SNMPv1 over IPv4/UDP)
        (
            CommunityData("public", mpModel=0),
            await UdpTransportTarget.create(("demo.pysnmp.com", 162)),
            ContextData(),
        ),
        # 2-nd target (SNMPv2c over IPv4/UDP)
        (
            CommunityData("public"),
            await UdpTransportTarget.create(("demo.pysnmp.com", 162)),
            ContextData(),
        ),
    )

    snmpEngine = SnmpEngine()

    for authData, transportTarget, contextData in TARGETS:
        await send_notification(
            snmpEngine,
            authData,
            transportTarget,
            contextData,
            "trap",  # NotifyType
            NotificationType(ObjectIdentity("SNMPv2-MIB", "coldStart")).add_varbinds(
                ("1.3.6.1.2.1.1.1.0", "my name")
            ),
        )

    snmpEngine.transport_dispatcher.run_dispatcher()


asyncio.run(run())
```

For further info, contact the git downloaded docs I have saved