### Install Sphinx Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/README.txt Install the Sphinx documentation generator using pip. This is a prerequisite for building the documentation. ```bash # pip install sphinx ``` -------------------------------- ### Install and Set Up Pre-commit Hooks Source: https://github.com/lextudio/pysnmp.git/blob/main/Development.md Installs pre-commit as a global uv tool for consistent code formatting and checks. Use `--overwrite` to re-initialize hooks. ```bash uv tool install pre-commit pre-commit install --overwrite ``` ```bash pre-commit run --all-files ``` ```bash uv run pre-commit install --overwrite ``` -------------------------------- ### Asyncio GETNEXT Operation Example Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v3arch/asyncio/manager/cmdgen/nextcmd.md Demonstrates how to perform an asynchronous SNMP GETNEXT query using next_cmd. Ensure asyncio is installed and configured for running asynchronous code. This example queries for the system description. ```python import asyncio from pysnmp.hlapi.v3arch.asyncio import * async def run(): errorIndication, errorStatus, errorIndex, varBinds = await next_cmd( SnmpEngine(), CommunityData('public'), await UdpTransportTarget.create(('demo.pysnmp.com', 161)), ContextData(), ObjectType(ObjectIdentity('SNMPv2-MIB', 'system')) ) print(errorIndication, errorStatus, errorIndex, varBinds) asyncio.run(run()) ``` -------------------------------- ### Asyncio GET Operation Example Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v3arch/asyncio/manager/cmdgen/getcmd.md Demonstrates how to perform an asynchronous SNMP GET request for a specific MIB variable using `get_cmd`. Ensure you have an asyncio event loop running. The `UdpTransportTarget.create` method is asynchronous and must be awaited. ```python import asyncio from pysnmp.hlapi.v3arch.asyncio import * async def run(): errorIndication, errorStatus, errorIndex, varBinds = await get_cmd( SnmpEngine(), CommunityData('public'), await UdpTransportTarget.create(('demo.pysnmp.com', 161)), ContextData(), ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)) ) print(errorIndication, errorStatus, errorIndex, varBinds) asyncio.run(run()) ``` -------------------------------- ### Install PySNMP from PyPI Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/download.md Use this command to install the stable version of PySNMP from the Python Package Index. ```bash $ pip install pysnmp ``` -------------------------------- ### Build HTML Documentation Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/README.txt Once Sphinx is installed, run this command in the project directory to generate the HTML version of the documentation. ```bash $ make html ``` -------------------------------- ### Asyncio Agent Setup for Table Implementation Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/examples/v3arch/asyncio/agent/cmdrsp/agent-side-mib-implementations.md Configures an asyncio-based SNMP agent to listen for queries and respond to SNMPv2c requests. This snippet focuses on the initial setup for implementing MIB tables, including transport and security configurations. ```python from pysnmp.entity import engine, config from pysnmp.entity.rfc3413 import cmdrsp, context from pysnmp.carrier.asyncio.dgram import udp from pysnmp.proto.api import v2c # Create SNMP engine snmpEngine = engine.SnmpEngine() # Transport setup # UDP over IPv4 config.add_transport( snmpEngine, udp.DOMAIN_NAME, udp.UdpTransport().open_server_mode(("127.0.0.1", 161)) ) # SNMPv2c setup # SecurityName <-> CommunityName mapping. config.add_v1_system(snmpEngine, "my-area", "public") # Allow read MIB access for this user / securityModels at VACM config.add_vacm_user( snmpEngine, 2, "my-area", "noAuthNoPriv", (1, 3, 6, 6), (1, 3, 6, 6) ) # Create an SNMP context snmpContext = context.SnmpContext(snmpEngine) ``` -------------------------------- ### Install PySNMP using pip Source: https://github.com/lextudio/pysnmp.git/blob/main/README.md Use this command to download and install the PySNMP package and its dependencies from PyPI. ```bash pip install pysnmp ``` -------------------------------- ### Opaque Type Example Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/api-reference.md Shows how to create and manipulate Opaque types, which can hold arbitrary ASN.1 syntax. Examples include string initialization, concatenation, and hexadecimal initialization. ```python >>> from pysnmp.proto.rfc1902 import * >>> Opaque('some apples') Opaque('some apples') >>> Opaque('some apples') + ' and oranges' Opaque('some apples and oranges') >>> str(Opaque('some apples')) 'some apples' >>> str(Opaque(hexValue='deadbeef')) '\xde\xad\xbe\xef' >>> ``` -------------------------------- ### Asyncio SNMP v3arch Example Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/examples/hlapi/v3arch/asyncio/index.md This example demonstrates sending an SNMPv3 notification using asyncio. It requires setting up the SNMP engine, adding variable bindings, and running the asyncio event loop. ```python from pysnmp.hlapi import * import asyncio def run(): snmpEngine = SnmpEngine() errorIndication, errorStatus, errorIndex, varBinds = next( nextCmd( snmpEngine, UdpTransportTarget(('demo.snmplabs.org', 161)), CommunityData('public', mpModel=1), UdpTransportTarget(('localhost', 162)), ContextData(), NotifyCmd( '1.3.6.1.6.3.1.1.5.4', ( ('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()) ``` -------------------------------- ### Asyncio Bulk Command Example Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v3arch/asyncio/manager/cmdgen/bulkcmd.md Demonstrates how to perform an asynchronous bulk SNMP command to retrieve system information. Ensure asyncio and necessary pysnmp components are imported. ```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()) ``` -------------------------------- ### SNMPv3 GET with Master Auth/Priv Keys (Asyncio) Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/examples/hlapi/v3arch/asyncio/manager/cmdgen/advanced-topics.md Send an SNMP GET request using SNMPv3 with master authentication and privacy keys. This is useful when you have pre-generated keys and want to avoid passphrase-based key derivation. ```python import asyncio from pysnmp.hlapi.v3arch.asyncio import * async def run_snmp_get(): iterator = await get_cmd( SnmpEngine(), UsmUserData( "usr-md5-des", authKey=OctetString(hexValue="1dcf59e86553b3afa5d32fd5d61bf0cf"), privKey=OctetString(hexValue="ec5ab55e93e1d85cb6846d0f23e845e0"), authKeyType=USM_KEY_TYPE_MASTER, privKeyType=USM_KEY_TYPE_MASTER, ), await UdpTransportTarget.create(("demo.pysnmp.com", 161)), ContextData(), ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), ) errorIndication, errorStatus, errorIndex, varBinds = 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])) asyncio.run(run_snmp_get()) ``` -------------------------------- ### Counter64 Type Example Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/api-reference.md Demonstrates the Counter64 type, representing a non-negative integer that wraps around at its maximum value. Includes creation and arithmetic operations. ```python >>> from pysnmp.proto.rfc1902 import * >>> Counter64(1234) Counter64(1234) >>> Counter64(1) + 1 Counter64(2) >>> int(Counter64(321)) 321 >>> ``` -------------------------------- ### Build Documentation Source: https://github.com/lextudio/pysnmp.git/blob/main/Development.md Builds the project documentation. Use `make -C docs html` for the current version and `sphinx-polyversion` for multiple versions. ```bash uv run make -C docs html ``` ```bash uv run sphinx-polyversion docs/poly.py ``` -------------------------------- ### SNMPv1 GET Request with asyncio Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/examples/hlapi/v1arch/asyncio/manager/cmdgen/snmp-versions.md Use this snippet to send an SNMP GET request using SNMPv1 over IPv4/UDP. It requires specifying the community string, transport target, and the OID. This example is based on the asyncio I/O framework. ```python import asyncio from pysnmp.hlapi.v1arch.asyncio import * async def run(): with SnmpDispatcher() as snmpDispatcher: iterator = await get_cmd( snmpDispatcher, CommunityData("public", mpModel=0), await UdpTransportTarget.create(("demo.pysnmp.com", 161)), ("1.3.6.1.2.1.1.1.0", None), ) errorIndication, errorStatus, errorIndex, varBinds = 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])) asyncio.run(run()) ``` -------------------------------- ### Asyncio Bulk Walk Command Example Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v3arch/asyncio/manager/cmdgen/bulkwalkcmd.md Demonstrates initiating an asynchronous bulk walk and iterating through the results. It shows how to fetch initial OIDs and then send a request for additional OIDs within the same walk session. ```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))]) ``` -------------------------------- ### Set Up PySNMP Test Environment Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/quick-start.md Initializes a test folder, sets up a Python 3.12 virtual environment using pipenv, and installs the pysnmp library. Use this to prepare your system for PySNMP development. ```bash cd ~ mkdir test-field cd test-field pyenv local 3.12 pip install pipenv pipenv install pysnmp pipenv run pip list ``` -------------------------------- ### Sequential SNMP GET Requests with asyncio Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/examples/hlapi/v3arch/asyncio/manager/cmdgen/advanced-topics.md Execute multiple SNMP GET requests one after another in a sequential manner using asyncio. This approach is suitable when the order of operations matters or when dealing with dependencies between requests. The code iterates through a list of hostnames and performs a GET request for each. ```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), ], ) ) ``` -------------------------------- ### Sequential SNMP GET Queries with Asyncio Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/examples/hlapi/v1arch/asyncio/manager/cmdgen/advanced-topics.md Send multiple SNMP GET requests one after another using asyncio. This approach is useful when the order of operations matters or when dealing with agents that might not handle concurrent requests well. It iterates through a list of hostnames, performing a GET request for each. ```python import asyncio from pysnmp.hlapi.v1arch.asyncio import * async def getone(snmpDispatcher, hostname): iterator = await get_cmd( snmpDispatcher, CommunityData("public"), await UdpTransportTarget.create(hostname), ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), ) errorIndication, errorStatus, errorIndex, varBinds = 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])) async def getall(snmpDispatcher, hostnames): for hostname in hostnames: await getone(snmpDispatcher, hostname) with SnmpDispatcher() as snmpDispatcher: asyncio.run( getall( snmpDispatcher, [ ("demo.pysnmp.com", 161), ("demo.pysnmp.com", 161), ("demo.pysnmp.com", 161), ], ) ) ``` -------------------------------- ### Initialize MIB Builder and Load Modules Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/examples/smi/manager/browsing-mib-tree.md Sets up the MIB loader/builder and configures MIB sources. This is the initial step for any MIB introspection task. ```python from pysnmp.smi import builder, view, compiler, error # Create MIB loader/builder mibBuilder = builder.MibBuilder() # Optionally attach PySMI MIB compiler (if installed) # print('Attaching MIB compiler...'), # compiler.addMibCompiler(mibBuilder, sources=['/usr/share/snmp/mibs']) # print('done') # Optionally set an alternative path to compiled MIBs print("Setting MIB sources...") mibBuilder.add_mib_sources(builder.DirMibSource("/opt/pysnmp_mibs")) print(mibBuilder.get_mib_sources()) print("done") print("Loading MIB modules..."), mibBuilder.load_modules( "SNMPv2-MIB", "SNMP-FRAMEWORK-MIB", "SNMP-COMMUNITY-MIB", "IP-MIB" ) print("done") print("Indexing MIB objects..."), mibView = view.MibViewController(mibBuilder) print("done") ``` -------------------------------- ### Asyncio SNMP Walk Example Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v3arch/asyncio/manager/cmdgen/walkcmd.md Demonstrates an asynchronous SNMP walk using `walk_cmd`. Initializes SNMP engine, community data, UDP transport target, and context. Retrieves `sysDescr` and then `ifInOctets` using an async generator. ```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))]) ``` -------------------------------- ### SNMP GET Operation Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v3arch/asyncio/manager/cmdgen/getcmd.md Creates a generator to perform an SNMP GET query asynchronously. The generator yields results when the response arrives or an error occurs. ```APIDOC ## async pysnmp.hlapi.v3arch.asyncio.get_cmd ### Description Creates a generator to perform SNMP GET query. When iterator gets advanced by `asyncio` main loop, SNMP GET request is send ([**RFC 1905 Section 4.2.1**](https://datatracker.ietf.org/doc/html/rfc1905.html#section-4.2.1)). The iterator yields `asyncio.get_running_loop().create_future()` which gets done whenever response arrives or error occurs. ### Parameters * **snmpEngine** (`SnmpEngine`) – Class instance representing SNMP engine. * **authData** (`CommunityData` or `UsmUserData`) – Class instance representing SNMP credentials. * **transportTarget** (`UdpTransportTarget` or `Udp6TransportTarget`) – Class instance representing transport type along with SNMP peer address. * **contextData** (`ContextData`) – Class instance representing SNMP ContextEngineId and ContextName values. * **\*varBinds** (`ObjectType`) – One or more class instances representing MIB variables to place into SNMP request. ### Other Parameters * **\*\*options** – Request options: * lookupMib - load MIB and resolve response MIB variables at the cost of slightly reduced performance. Default is True. ### Yields * **errorIndication** (`ErrorIndication`) – True value indicates SNMP engine error. * **errorStatus** (*str*) – True value indicates SNMP PDU error. * **errorIndex** (*int*) – Non-zero value refers to varBinds[errorIndex-1] * **varBinds** (*tuple*) – A sequence of `ObjectType` class instances representing MIB variables returned in SNMP response. ### Raises **PySnmpError** – Or its derivative indicating that an error occurred while performing SNMP operation. ### Example ```pycon >>> import asyncio >>> from pysnmp.hlapi.v3arch.asyncio import * >>> >>> async def run(): ... errorIndication, errorStatus, errorIndex, varBinds = await get_cmd( ... SnmpEngine(), ... CommunityData('public'), ... await UdpTransportTarget.create(('demo.pysnmp.com', 161)), ... ContextData(), ... ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)) ... ) ... 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'))]) >>> ``` ``` -------------------------------- ### Create and use Unsigned32 Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/api-reference.md Shows how to create an Unsigned32 instance. Requires importing Unsigned32 from pysnmp.proto.rfc1902. ```pycon >>> from pysnmp.proto.rfc1902 import * >>> Unsigned32(1234) Unsigned32(1234) ``` -------------------------------- ### Asyncio WALK Operation Example Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v1arch/asyncio/manager/cmdgen/walkcmd.md Demonstrates how to use walk_cmd to perform an SNMP WALK operation asynchronously. It shows initial retrieval and subsequent calls to send new varBinds to the generator. ```python from pysnmp.hlapi.v1arch.asyncio import * objects = walk_cmd(SnmpEngine(), CommunityData('public'), await UdpTransportTarget.create(('demo.pysnmp.com', 161)), ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr'))) g = [item async for item in objects] next(g) g.send( [ ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets')) ] ) ``` -------------------------------- ### asyncio GET Operation Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v1arch/asyncio/manager/cmdgen/getcmd.md Creates a generator to perform SNMP GET query asynchronously. The iterator yields `asyncio.Future` which completes when the response arrives or an error occurs. ```APIDOC ## async pysnmp.hlapi.v1arch.asyncio.get_cmd ### Description Creates a generator to perform SNMP GET query. When the iterator is advanced by the `asyncio` main loop, an SNMP GET request is sent. The iterator yields `asyncio.Future` which gets done whenever a response arrives or an error occurs. ### Parameters * **snmpDispatcher** (`SnmpDispatcher`) – An instance representing the asyncio-based asynchronous event loop and associated state information. * **authData** (`CommunityData`) – An instance representing SNMPv1/v2c credentials. * **transportTarget** ([`UdpTransportTarget`](../../../../../api-reference.md#pysnmp.hlapi.v1arch.asyncio.UdpTransportTarget) or [`Udp6TransportTarget`](../../../../../api-reference.md#pysnmp.hlapi.v1arch.asyncio.Udp6TransportTarget)) – An instance representing the transport type along with the SNMP peer address. * **\*varBinds** (`tuple` of OID-value pairs or [`ObjectType`](../../../../../api-reference.md#pysnmp.smi.rfc1902.ObjectType)) – One or more instances representing MIB variables to include in the SNMP request. ### Other Parameters * **\*\*options** – Request options: * **lookupMib** (bool) - Load MIB and resolve response MIB variables at the cost of slightly reduced performance. Defaults to `False`, unless an `ObjectType` is present among `varBinds`, in which case `lookupMib` is automatically enabled. ### Yields * **errorIndication** (*str*) – If not `None`, indicates an SNMP engine error. * **errorStatus** (*str*) – If not `0`, indicates an SNMP PDU error. * **errorIndex** (*int*) – A non-zero value refers to `varBinds[errorIndex-1]`. * **varBinds** (*tuple*) – A sequence of OID-value pairs. If `lookupMib` is `False`, these are base SNMP types. If `lookupMib` is `True`, these are [`ObjectType`](../../../../../api-reference.md#pysnmp.smi.rfc1902.ObjectType) class instances. ### Raises * **PySnmpError** – Or its derivative, indicating an error occurred during the SNMP operation. ### Example ```pycon >>> import asyncio >>> from pysnmp.hlapi.v1arch.asyncio import * >>> >>> async def run(): ... errorIndication, errorStatus, errorIndex, varBinds = await get_cmd( ... SnmpDispatcher(), ... CommunityData('public'), ... await UdpTransportTarget.create(('demo.pysnmp.com', 161)), ... ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)) ... ) ... 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'))]) >>> ``` ``` -------------------------------- ### Verify GET Request Response Values Against MIB Constraints Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/faq/snmp-data-constraints-verification-failure.md Use this snippet to perform a GET request and verify that the Agent-supplied values satisfy MIB constraints. Set `lookupValues=True` to enable this verification. ```python errorIndication, errorStatus, errorIndex, varBinds = await cmdGen.get_cmd( cmdgen.CommunityData('public'), await cmdgen.UdpTransportTarget(('localhost', 161)), cmdgen.MibVariable('SNMPv2-MIB', 'sysName', 0), lookupValues=True ) ``` -------------------------------- ### Async SNMP GET Request Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/examples/hlapi/v3arch/asyncio/index.md Performs an SNMP GET request using the high-level API with asyncio. Requires specifying community data, transport target, context, and the MIB object to retrieve. ```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()) ``` -------------------------------- ### SNMPv3 Bulk Walk with Asyncio Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/examples/hlapi/v3arch/asyncio/manager/cmdgen/snmp-versions.md This example demonstrates sending a series of SNMP GETBULK requests using SNMPv3 with no authentication and no privacy, over IPv4/UDP. It runs until the end-of-mib condition is reported by the agent and is based on the asyncio I/O framework. ```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"))]) ) ``` -------------------------------- ### Asyncio Bulk Walk Command Usage Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v1arch/asyncio/manager/cmdgen/bulkwalkcmd.md Demonstrates how to use the `bulk_walk_cmd` function with asyncio to perform an SNMP bulk walk and retrieve objects. It shows how to initialize the command, iterate over the results, and send subsequent requests. ```APIDOC ## bulk_walk_cmd ### Description Performs an SNMP bulk walk operation asynchronously. ### Parameters - **snmpEngine** (SnmpEngine) - An instance of SnmpEngine. - **authData** (CommunityData or UsmUserData) - Authentication data for the SNMP session. - **transportTarget** (UdpTransportTarget or other TransportTarget) - The target device and port for the SNMP request. - **nonRepeaters** (int) - The number of non-repeating variables. - **maxRepetitions** (int) - The maximum number of repetitions for each repeating variable. - **objectTypes** (ObjectType or list of ObjectType) - The object(s) to walk. ### Request Example ```python from pysnmp.hlapi.v1arch.asyncio import * objects = bulk_walk_cmd(SnmpEngine(), CommunityData('public'), await UdpTransportTarget.create(('demo.pysnmp.com', 161)), 0, 25, ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr'))) g = [item async for item in objects] next(g) send( [ ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets')) ] ) ``` ### Response Returns an asynchronous generator yielding tuples of (errorIndication, errorStatus, errorIndex, varBinds). #### Success Response Example ```python (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'))]) ``` #### Subsequent Request Example ```python (None, 0, 0, [(ObjectName('1.3.6.1.2.1.2.2.1.10.1'), Counter32(284817787))]) ``` ``` -------------------------------- ### Asyncio Bulk Walk Example Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v1arch/asyncio/manager/cmdgen/bulkwalkcmd.md Demonstrates initiating an asynchronous bulk walk to retrieve SNMP objects. Requires an active asyncio event loop. The `bulk_walk_cmd` returns an async generator. ```python >>> from pysnmp.hlapi.v1arch.asyncio import * >>> objects = bulk_walk_cmd(SnmpEngine(), ... CommunityData('public'), ... await UdpTransportTarget.create(('demo.pysnmp.com', 161)), ... 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))]) ``` -------------------------------- ### Asyncio GETBULK Operation Example Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v1arch/asyncio/manager/cmdgen/bulkcmd.md Demonstrates how to perform an asynchronous SNMP GETBULK query using `bulk_cmd`. Ensure you have an asyncio event loop running and the necessary PySNMP components imported. ```python import asyncio from pysnmp.hlapi.v1arch.asyncio import * async def run(): errorIndication, errorStatus, errorIndex, varBinds = await bulk_cmd( SnmpDispatcher(), CommunityData('public'), await UdpTransportTarget.create(('demo.pysnmp.com', 161)), 0, 2, ObjectType(ObjectIdentity('SNMPv2-MIB', 'system')) ) print(errorIndication, errorStatus, errorIndex, varBinds) asyncio.run(run()) ``` -------------------------------- ### Concurrent SNMP GET Queries with Asyncio Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/examples/hlapi/v1arch/asyncio/manager/cmdgen/advanced-topics.md Send multiple SNMP GET requests simultaneously using asyncio. This snippet demonstrates how to leverage `asyncio.gather` for concurrent operations, suitable for querying multiple agents or multiple MIB objects at once. ```python import asyncio from pysnmp.hlapi.v1arch.asyncio import * async def getone(snmpDispatcher, hostname): iterator = await get_cmd( snmpDispatcher, CommunityData("public"), await UdpTransportTarget.create(hostname), ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), ) errorIndication, errorStatus, errorIndex, varBinds = 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])) async def main(): with SnmpDispatcher() as snmpDispatcher: await asyncio.gather( getone(snmpDispatcher, ("demo.pysnmp.com", 161)), getone(snmpDispatcher, ("demo.pysnmp.com", 161)), getone(snmpDispatcher, ("demo.pysnmp.com", 161)), ) asyncio.run(main()) ``` -------------------------------- ### Asynchronous Walk Command Usage Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v3arch/asyncio/manager/cmdgen/walkcmd.md Demonstrates how to perform an asynchronous SNMP walk to retrieve multiple MIB objects. It shows initializing the SNMP engine, transport targets, and context, then iterating over the results. It also illustrates how to send additional OIDs to the walk generator. ```APIDOC ## walk_cmd (asyncio) ### Description Performs an asynchronous SNMP walk operation to retrieve a range of MIB objects starting from a given OID. ### Method Signature ```python walk_cmd(snmpEngine, authData, transportTarget, contextData, *varBinds) ``` ### Parameters - **snmpEngine**: An instance of `SnmpEngine`. - **authData**: An instance of `CommunityData` or `UsmUserData` for authentication. - **transportTarget**: An instance of `TransportTarget` (e.g., `UdpTransportTarget`) specifying the target agent. - **contextData**: An instance of `ContextData`. - **varBinds**: One or more `ObjectType` instances specifying the OIDs to walk. ### Request Example ```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'))]) ``` ### Sending Additional OIDs After the initial walk, you can send additional OIDs to the generator to continue the walk. ### Request Example ```python >>> g.send( [ ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets')) ] ) (None, 0, 0, [(ObjectName('1.3.6.1.2.1.2.2.1.10.1'), Counter32(284817787))]) ``` ``` -------------------------------- ### Perform SNMP GET Request Asynchronously Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v1arch/asyncio/manager/cmdgen/getcmd.md Use `get_cmd` to send an SNMP GET request and await its response. Ensure an asyncio event loop is running and the necessary pysnmp components are imported. The `SnmpDispatcher` should be maintained for the application's lifecycle. ```python import asyncio from pysnmp.hlapi.v1arch.asyncio import * async def run(): errorIndication, errorStatus, errorIndex, varBinds = await get_cmd( SnmpDispatcher(), CommunityData('public'), await UdpTransportTarget.create(('demo.pysnmp.com', 161)), ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)) ) print(errorIndication, errorStatus, errorIndex, varBinds) asyncio.run(run()) ``` -------------------------------- ### Concurrent SNMP GET Requests with asyncio Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/examples/hlapi/v3arch/asyncio/manager/cmdgen/advanced-topics.md Execute multiple SNMP GET requests simultaneously using asyncio.gather. This is useful for querying several agents or multiple OIDs in parallel for faster data retrieval. Ensure the asyncio event loop is running. ```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()) ``` -------------------------------- ### Asyncio SET Operation Example Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/docs/hlapi/v1arch/asyncio/manager/cmdgen/setcmd.md Demonstrates how to perform an SNMP SET operation asynchronously using `set_cmd`. Ensure `asyncio` is imported and an event loop is running. The `SnmpDispatcher` should ideally be reused. ```python >>> import asyncio >>> from pysnmp.hlapi.v1arch.asyncio import * >>> >>> async def run(): ... errorIndication, errorStatus, errorIndex, varBinds = await set_cmd( ... SnmpDispatcher(), ... CommunityData('public'), ... await UdpTransportTarget.create(('demo.pysnmp.com', 161)), ... ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0), 'Linux i386') ... ) ... print(errorIndication, errorStatus, errorIndex, varBinds) >>> >>> asyncio.run(run()) (None, 0, 0, [ObjectType(ObjectIdentity(ObjectName('1.3.6.1.2.1.1.1.0')), DisplayString('Linux i386'))]) >>> ``` -------------------------------- ### SNMPv3 GET with Localized Auth/Priv Keys (Asyncio) Source: https://github.com/lextudio/pysnmp.git/blob/main/docs/source/examples/hlapi/v3arch/asyncio/manager/cmdgen/advanced-topics.md Send an SNMP GET request using SNMPv3 with localized authentication and privacy keys, specifying the authoritative SNMP engine ID. This method is used when keys are localized to a specific SNMP engine. ```python import asyncio from pysnmp.hlapi.v3arch.asyncio import * async def run_snmp_get(): iterator = await get_cmd( SnmpEngine(), UsmUserData( "usr-md5-des", authKey=OctetString(hexValue="6b99c475259ef7976cf8d028a3381eeb"), privKey=OctetString(hexValue="92b5ef98f0a216885e73944e58c07345"), authKeyType=USM_KEY_TYPE_LOCALIZED, privKeyType=USM_KEY_TYPE_LOCALIZED, securityEngineId=OctetString(hexValue="80004fb805636c6f75644dab22cc"), ), await UdpTransportTarget.create(("demo.pysnmp.com", 161)), ContextData(), ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)), ) errorIndication, errorStatus, errorIndex, varBinds = 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])) asyncio.run(run_snmp_get()) ``` -------------------------------- ### Perform Parallel SNMP GET Queries with asyncio Source: https://context7.com/lextudio/pysnmp.git/llms.txt This snippet demonstrates how to perform multiple SNMP GET queries concurrently using asyncio.gather. It requires the pysnmp library and targets specified hosts and ports. Ensure the SNMP community string and MIB objects are correctly defined. ```python import asyncio from pysnmp.hlapi.v3arch.asyncio import ( SnmpEngine, CommunityData, UdpTransportTarget, ContextData, ObjectType, ObjectIdentity, get_cmd, ) TARGETS = [ ('192.168.0.1', 161), ('192.168.0.2', 161), ('192.168.0.3', 161), ] async def query_agent(snmpEngine, host, port): errorIndication, errorStatus, errorIndex, varBinds = await get_cmd( snmpEngine, CommunityData('public'), await UdpTransportTarget.create((host, port), timeout=2, retries=1), ContextData(), ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysUpTime', 0)), ) if errorIndication: return host, None, str(errorIndication) if errorStatus: return host, None, errorStatus.prettyPrint() return host, {str(vb[0]): str(vb[1]) for vb in varBinds}, None async def run(): snmpEngine = SnmpEngine() tasks = [query_agent(snmpEngine, host, port) for host, port in TARGETS] results = await asyncio.gather(*tasks, return_exceptions=True) for host, data, err in results: if err: print(f'{host}: ERROR — {err}') else: for oid, val in data.items(): print(f'{host} {oid} = {val}') snmpEngine.close_dispatcher() asyncio.run(run()) # 192.168.0.1 SNMPv2-MIB::sysDescr.0 = Linux router 5.15.0 # 192.168.0.1 SNMPv2-MIB::sysUpTime.0 = 44430646 # 192.168.0.2 ... ``` -------------------------------- ### Iterate MIB Objects with SNMP WALK using `walk_cmd()` Source: https://context7.com/lextudio/pysnmp.git/llms.txt Employ `walk_cmd` as an async generator for iterative GETNEXT operations. It traverses MIB objects from a starting OID. Set `lexicographicMode=False` to restrict traversal to the starting OID's subtree. Requires `asyncio` and specific pysnmp imports. ```python import asyncio from pysnmp.hlapi.v3arch.asyncio import ( SnmpEngine, CommunityData, UdpTransportTarget, ContextData, ObjectType, ObjectIdentity, walk_cmd, ) async def run(): snmpEngine = SnmpEngine() objects = walk_cmd( snmpEngine, CommunityData('public'), await UdpTransportTarget.create(('demo.pysnmp.com', 161)), ContextData(), ObjectType(ObjectIdentity('SNMPv2-MIB', 'system')), lexicographicMode=False, # stay within SNMPv2-MIB::system subtree ) async for errorIndication, errorStatus, errorIndex, varBinds in objects: if errorIndication: print(f'Engine error: {errorIndication}') break elif errorStatus: print(f'PDU error: {errorStatus.prettyPrint()}') break else: for varBind in varBinds: print(' = '.join([x.prettyPrint() for x in varBind])) # SNMPv2-MIB::sysDescr.0 = SunOS zeus.pysnmp.com ... # SNMPv2-MIB::sysObjectID.0 = SNMPv2-SMI::enterprises.20408 # ... snmpEngine.close_dispatcher() asyncio.run(run()) ```