CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-pysnmp

A comprehensive Python SNMP library supporting v1/v2c/v3 with authentication and privacy protocols

84

0.94x
Overview
Eval results
Files

high-level-api.mddocs/

High-Level API Operations

The PySNMP High-Level API provides async/await functions for all common SNMP operations. These functions handle transport management, error processing, and provide a clean interface for SNMP communication.

Capabilities

SNMP GET Operation

Retrieves the value of specific SNMP variables from a remote agent.

async def get_cmd(
    snmpEngine: SnmpEngine,
    authData: CommunityData | UsmUserData,
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
    contextData: ContextData,
    *varBinds: ObjectType,
    **options
) -> tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]]:
    """
    Perform SNMP GET operation.
    
    Parameters:
    - snmpEngine: SNMP engine instance
    - authData: Authentication data (community or USM user)
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
    - contextData: SNMP context information
    - varBinds: Variable bindings to retrieve
    - options: Additional options (timeout, retries, etc.)
    
    Returns:
    - errorIndication: Transport/network error or None
    - errorStatus: SNMP error status or None  
    - errorIndex: Index of failed variable or None
    - varBinds: Tuple of retrieved variable bindings
    """

Usage Example:

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

async def get_system_description():
    errorIndication, errorStatus, errorIndex, varBinds = await get_cmd(
        SnmpEngine(),
        CommunityData('public'),
        await UdpTransportTarget.create(('demo.pysnmp.com', 161)),
        ContextData(),
        ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0'))  # sysDescr
    )
    
    if errorIndication:
        print(f"Error: {errorIndication}")
    elif errorStatus:
        print(f"SNMP Error: {errorStatus.prettyPrint()} at index {errorIndex}")
    else:
        for varBind in varBinds:
            print(f"{varBind[0]} = {varBind[1]}")

asyncio.run(get_system_description())

SNMP SET Operation

Modifies the value of writable SNMP variables on a remote agent.

async def set_cmd(
    snmpEngine: SnmpEngine,
    authData: CommunityData | UsmUserData,
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
    contextData: ContextData,
    *varBinds: ObjectType,
    **options
) -> tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]]:
    """
    Perform SNMP SET operation.
    
    Parameters:
    - snmpEngine: SNMP engine instance
    - authData: Authentication data (community or USM user)
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
    - contextData: SNMP context information
    - varBinds: Variable bindings with new values to set
    - options: Additional options (timeout, retries, etc.)
    
    Returns:
    - errorIndication: Transport/network error or None
    - errorStatus: SNMP error status or None
    - errorIndex: Index of failed variable or None
    - varBinds: Tuple of set variable bindings
    """

Usage Example:

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

async def set_system_contact():
    errorIndication, errorStatus, errorIndex, varBinds = await set_cmd(
        SnmpEngine(),
        CommunityData('private'),
        await UdpTransportTarget.create(('demo.pysnmp.com', 161)),
        ContextData(),
        ObjectType(ObjectIdentity('1.3.6.1.2.1.1.4.0'), OctetString('admin@company.com'))
    )
    
    if errorIndication:
        print(f"Error: {errorIndication}")
    elif errorStatus:
        print(f"SNMP Error: {errorStatus.prettyPrint()} at index {errorIndex}")
    else:
        print("SET operation successful")
        for varBind in varBinds:
            print(f"{varBind[0]} = {varBind[1]}")

asyncio.run(set_system_contact())

SNMP GETNEXT Operation

Retrieves the next available SNMP variable following the specified OID in lexicographic order.

async def next_cmd(
    snmpEngine: SnmpEngine,
    authData: CommunityData | UsmUserData,
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
    contextData: ContextData,
    *varBinds: ObjectType,
    **options
) -> tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]]:
    """
    Perform SNMP GETNEXT operation.
    
    Parameters:
    - snmpEngine: SNMP engine instance
    - authData: Authentication data (community or USM user)
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
    - contextData: SNMP context information
    - varBinds: Starting variable bindings for GETNEXT
    - options: Additional options (timeout, retries, etc.)
    
    Returns:
    - errorIndication: Transport/network error or None
    - errorStatus: SNMP error status or None
    - errorIndex: Index of failed variable or None
    - varBinds: Tuple of next variable bindings
    """

SNMP GETBULK Operation

Efficiently retrieves multiple SNMP variables using the GETBULK operation (SNMPv2c and v3 only).

async def bulk_cmd(
    snmpEngine: SnmpEngine,
    authData: CommunityData | UsmUserData,
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
    contextData: ContextData,
    nonRepeaters: int,
    maxRepetitions: int,
    *varBinds: ObjectType,
    **options
) -> tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]]:
    """
    Perform SNMP GETBULK operation.
    
    Parameters:
    - snmpEngine: SNMP engine instance
    - authData: Authentication data (community or USM user)
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
    - contextData: SNMP context information
    - nonRepeaters: Number of variables to retrieve only once
    - maxRepetitions: Maximum repetitions for remaining variables
    - varBinds: Variable bindings to retrieve
    - options: Additional options (timeout, retries, etc.)
    
    Returns:
    - errorIndication: Transport/network error or None
    - errorStatus: SNMP error status or None
    - errorIndex: Index of failed variable or None
    - varBinds: Tuple of retrieved variable bindings
    """

SNMP WALK Operation

Iterates through a MIB subtree, retrieving all variables under a specified OID.

async def walk_cmd(
    snmpEngine: SnmpEngine,
    authData: CommunityData | UsmUserData,
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
    contextData: ContextData,
    *varBinds: ObjectType,
    lexicographicMode: bool = True,
    maxRows: int = 0,
    ignoreNonIncreasingOid: bool = False,
    **options
) -> AsyncGenerator[tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]], None]:
    """
    Perform SNMP WALK operation (async generator).
    
    Parameters:
    - snmpEngine: SNMP engine instance
    - authData: Authentication data (community or USM user)
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
    - contextData: SNMP context information
    - varBinds: Starting variable bindings for walk
    - lexicographicMode: Continue beyond initial subtree if True
    - maxRows: Maximum number of rows to retrieve (0 = unlimited)
    - ignoreNonIncreasingOid: Ignore non-increasing OID errors
    - options: Additional options (timeout, retries, etc.)
    
    Yields:
    - errorIndication: Transport/network error or None
    - errorStatus: SNMP error status or None
    - errorIndex: Index of failed variable or None
    - varBinds: Tuple of walked variable bindings
    """

Usage Example:

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

async def walk_interfaces():
    async for (errorIndication, errorStatus, errorIndex, varBinds) in walk_cmd(
        SnmpEngine(),
        CommunityData('public'),
        await UdpTransportTarget.create(('demo.pysnmp.com', 161)),
        ContextData(),
        ObjectType(ObjectIdentity('1.3.6.1.2.1.2.2.1.2')),  # ifDescr table
        lexicographicMode=False,
        maxRows=10
    ):
        if errorIndication:
            print(f"Error: {errorIndication}")
            break
        elif errorStatus:
            print(f"SNMP Error: {errorStatus.prettyPrint()}")
            break
        else:
            for varBind in varBinds:
                print(f"{varBind[0]} = {varBind[1]}")

asyncio.run(walk_interfaces())

SNMP BULK WALK Operation

Efficiently walks a MIB subtree using GETBULK operations for improved performance.

async def bulk_walk_cmd(
    snmpEngine: SnmpEngine,
    authData: CommunityData | UsmUserData,
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
    contextData: ContextData,
    nonRepeaters: int,
    maxRepetitions: int,
    *varBinds: ObjectType,
    lexicographicMode: bool = True,
    maxRows: int = 0,
    ignoreNonIncreasingOid: bool = False,
    **options
) -> AsyncGenerator[tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]], None]:
    """
    Perform SNMP BULK WALK operation (async generator).
    
    Parameters:
    - snmpEngine: SNMP engine instance
    - authData: Authentication data (community or USM user)
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
    - contextData: SNMP context information
    - nonRepeaters: Number of variables to retrieve only once
    - maxRepetitions: Maximum repetitions for remaining variables
    - varBinds: Starting variable bindings for bulk walk
    - lexicographicMode: Continue beyond initial subtree if True
    - maxRows: Maximum number of rows to retrieve (0 = unlimited)
    - ignoreNonIncreasingOid: Ignore non-increasing OID errors
    - options: Additional options (timeout, retries, etc.)
    
    Yields:
    - errorIndication: Transport/network error or None
    - errorStatus: SNMP error status or None
    - errorIndex: Index of failed variable or None
    - varBinds: Tuple of walked variable bindings
    """

SNMP Notification Sending

Sends SNMP TRAP or INFORM notifications to network management systems.

async def send_notification(
    snmpEngine: SnmpEngine,
    authData: CommunityData | UsmUserData,
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
    contextData: ContextData,
    notifyType: str,
    *varBinds: NotificationType | ObjectType,
    **options
) -> tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]]:
    """
    Send SNMP notification (TRAP or INFORM).
    
    Parameters:
    - snmpEngine: SNMP engine instance
    - authData: Authentication data (community or USM user)
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
    - contextData: SNMP context information
    - notifyType: 'trap' or 'inform'
    - varBinds: Notification variable bindings
    - options: Additional options (timeout, retries, etc.)
    
    Returns:
    - errorIndication: Transport/network error or None
    - errorStatus: SNMP error status or None
    - errorIndex: Index of failed variable or None
    - varBinds: Tuple of sent variable bindings
    """

Usage Example:

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

async def send_trap():
    errorIndication, errorStatus, errorIndex, varBinds = await send_notification(
        SnmpEngine(),
        CommunityData('public'),
        await UdpTransportTarget.create(('demo.pysnmp.com', 162)),
        ContextData(),
        'trap',
        NotificationType(ObjectIdentity('1.3.6.1.4.1.20408.4.1.1.2')),
        ObjectType(ObjectIdentity('1.3.6.1.4.1.20408.4.1.1.2.1'), OctetString('Custom trap message'))
    )
    
    if errorIndication:
        print(f"Error: {errorIndication}")
    elif errorStatus:
        print(f"SNMP Error: {errorStatus.prettyPrint()}")
    else:
        print("Notification sent successfully")

asyncio.run(send_trap())

Utility Functions

End of MIB Detection

Utility function to check if walk operations have reached the end of the MIB tree.

def is_end_of_mib(varBinds: tuple[ObjectType, ...]) -> bool:
    """
    Check if variable bindings indicate end of MIB view.
    
    Parameters:
    - varBinds: Tuple of variable bindings from walk operation
    
    Returns:
    True if end of MIB reached, False otherwise
    """

Usage Example:

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

async def walk_with_end_check():
    async for (errorIndication, errorStatus, errorIndex, varBinds) in walk_cmd(
        SnmpEngine(),
        CommunityData('public'),
        await UdpTransportTarget.create(('demo.pysnmp.com', 161)),
        ContextData(),
        ObjectType(ObjectIdentity('1.3.6.1.2.1.1')),
        lexicographicMode=False
    ):
        if errorIndication or errorStatus:
            break
        
        if is_end_of_mib(varBinds):
            print("End of MIB reached")
            break
            
        for varBind in varBinds:
            print(f"{varBind[0]} = {varBind[1]}")

asyncio.run(walk_with_end_check())

Transport Targets

UDP IPv4 Transport Target

class UdpTransportTarget:
    @classmethod
    async def create(
        cls,
        transportAddr: tuple[str, int],
        timeout: float = 1.0,
        retries: int = 5,
        tagList: bytes = b""
    ) -> UdpTransportTarget:
        """
        Create UDP/IPv4 transport target.
        
        Parameters:
        - transportAddr: Tuple of (hostname/IP, port)
        - timeout: Response timeout in seconds
        - retries: Maximum number of request retries
        - tagList: Arbitrary bytes for endpoint selection
        
        Returns:
        UdpTransportTarget instance
        """

UDP IPv6 Transport Target

class Udp6TransportTarget:
    @classmethod
    async def create(
        cls,
        transportAddr: tuple[str, int],
        timeout: float = 1.0,
        retries: int = 5,
        tagList: bytes = b""
    ) -> Udp6TransportTarget:
        """
        Create UDP/IPv6 transport target.
        
        Parameters:
        - transportAddr: Tuple of (hostname/IP, port)
        - timeout: Response timeout in seconds
        - retries: Maximum number of request retries
        - tagList: Arbitrary bytes for endpoint selection
        
        Returns:
        Udp6TransportTarget instance
        """

Core Components

SNMP Engine

class SnmpEngine:
    def __init__(
        self,
        snmpEngineID: bytes | None = None,
        maxMessageSize: int = 65507,
        msgAndPduDsp: object | None = None
    ):
        """
        Create SNMP engine instance.
        
        Parameters:
        - snmpEngineID: Unique engine identifier (auto-generated if None)
        - maxMessageSize: Maximum SNMP message size in bytes (default: 65507)
        - msgAndPduDsp: Message and PDU dispatcher (internal use, typically None)
        
        Notes:
        - snmpEngineID should be unique per SNMP engine instance
        - maxMessageSize affects maximum PDU size for operations
        - Auto-generated engine ID includes MAC address and timestamp
        """

Context Data

class ContextData:
    def __init__(
        self,
        contextEngineId: bytes | None = None,
        contextName: bytes = b""
    ):
        """
        Create SNMP context data.
        
        Parameters:
        - contextEngineId: SNMP context engine identifier
        - contextName: SNMP context name for MIB instance selection
        """

Error Handling

All high-level API functions return error information as the first three elements of their return tuple:

  • errorIndication: Network/transport errors (timeouts, connection failures)
  • errorStatus: SNMP protocol errors (noSuchName, readOnly, etc.)
  • errorIndex: Index of the variable that caused the error (1-based)

Common error patterns:

errorIndication, errorStatus, errorIndex, varBinds = await get_cmd(...)

if errorIndication:
    # Network/transport error
    print(f"Transport error: {errorIndication}")
elif errorStatus:
    # SNMP protocol error
    print(f"SNMP error: {errorStatus.prettyPrint()} at variable {errorIndex}")
else:
    # Success - process varBinds
    for varBind in varBinds:
        print(f"{varBind[0]} = {varBind[1]}")

Install with Tessl CLI

npx tessl i tessl/pypi-pysnmp

docs

authentication.md

data-types.md

high-level-api.md

index.md

tile.json