0
# High-Level API Operations
1

2
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.
3

4
## Capabilities
5

6
### SNMP GET Operation
7

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

10
```python { .api }
11
async def get_cmd(
12
    snmpEngine: SnmpEngine,
13
    authData: CommunityData | UsmUserData,
14
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
15
    contextData: ContextData,
16
    *varBinds: ObjectType,
17
    **options
18
) -> tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]]:
19
    """
20
    Perform SNMP GET operation.
21
    
22
    Parameters:
23
    - snmpEngine: SNMP engine instance
24
    - authData: Authentication data (community or USM user)
25
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
26
    - contextData: SNMP context information
27
    - varBinds: Variable bindings to retrieve
28
    - options: Additional options (timeout, retries, etc.)
29
    
30
    Returns:
31
    - errorIndication: Transport/network error or None
32
    - errorStatus: SNMP error status or None  
33
    - errorIndex: Index of failed variable or None
34
    - varBinds: Tuple of retrieved variable bindings
35
    """
36
```
37

38
Usage Example:
39

40
```python
41
import asyncio
42
from pysnmp.hlapi.v3arch.asyncio import *
43

44
async def get_system_description():
45
    errorIndication, errorStatus, errorIndex, varBinds = await get_cmd(
46
        SnmpEngine(),
47
        CommunityData('public'),
48
        await UdpTransportTarget.create(('demo.pysnmp.com', 161)),
49
        ContextData(),
50
        ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0'))  # sysDescr
51
    )
52
    
53
    if errorIndication:
54
        print(f"Error: {errorIndication}")
55
    elif errorStatus:
56
        print(f"SNMP Error: {errorStatus.prettyPrint()} at index {errorIndex}")
57
    else:
58
        for varBind in varBinds:
59
            print(f"{varBind[0]} = {varBind[1]}")
60

61
asyncio.run(get_system_description())
62
```
63

64
### SNMP SET Operation
65

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

68
```python { .api }
69
async def set_cmd(
70
    snmpEngine: SnmpEngine,
71
    authData: CommunityData | UsmUserData,
72
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
73
    contextData: ContextData,
74
    *varBinds: ObjectType,
75
    **options
76
) -> tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]]:
77
    """
78
    Perform SNMP SET operation.
79
    
80
    Parameters:
81
    - snmpEngine: SNMP engine instance
82
    - authData: Authentication data (community or USM user)
83
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
84
    - contextData: SNMP context information
85
    - varBinds: Variable bindings with new values to set
86
    - options: Additional options (timeout, retries, etc.)
87
    
88
    Returns:
89
    - errorIndication: Transport/network error or None
90
    - errorStatus: SNMP error status or None
91
    - errorIndex: Index of failed variable or None
92
    - varBinds: Tuple of set variable bindings
93
    """
94
```
95

96
Usage Example:
97

98
```python
99
import asyncio
100
from pysnmp.hlapi.v3arch.asyncio import *
101

102
async def set_system_contact():
103
    errorIndication, errorStatus, errorIndex, varBinds = await set_cmd(
104
        SnmpEngine(),
105
        CommunityData('private'),
106
        await UdpTransportTarget.create(('demo.pysnmp.com', 161)),
107
        ContextData(),
108
        ObjectType(ObjectIdentity('1.3.6.1.2.1.1.4.0'), OctetString('admin@company.com'))
109
    )
110
    
111
    if errorIndication:
112
        print(f"Error: {errorIndication}")
113
    elif errorStatus:
114
        print(f"SNMP Error: {errorStatus.prettyPrint()} at index {errorIndex}")
115
    else:
116
        print("SET operation successful")
117
        for varBind in varBinds:
118
            print(f"{varBind[0]} = {varBind[1]}")
119

120
asyncio.run(set_system_contact())
121
```
122

123
### SNMP GETNEXT Operation
124

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

127
```python { .api }
128
async def next_cmd(
129
    snmpEngine: SnmpEngine,
130
    authData: CommunityData | UsmUserData,
131
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
132
    contextData: ContextData,
133
    *varBinds: ObjectType,
134
    **options
135
) -> tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]]:
136
    """
137
    Perform SNMP GETNEXT operation.
138
    
139
    Parameters:
140
    - snmpEngine: SNMP engine instance
141
    - authData: Authentication data (community or USM user)
142
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
143
    - contextData: SNMP context information
144
    - varBinds: Starting variable bindings for GETNEXT
145
    - options: Additional options (timeout, retries, etc.)
146
    
147
    Returns:
148
    - errorIndication: Transport/network error or None
149
    - errorStatus: SNMP error status or None
150
    - errorIndex: Index of failed variable or None
151
    - varBinds: Tuple of next variable bindings
152
    """
153
```
154

155
### SNMP GETBULK Operation
156

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

159
```python { .api }
160
async def bulk_cmd(
161
    snmpEngine: SnmpEngine,
162
    authData: CommunityData | UsmUserData,
163
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
164
    contextData: ContextData,
165
    nonRepeaters: int,
166
    maxRepetitions: int,
167
    *varBinds: ObjectType,
168
    **options
169
) -> tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]]:
170
    """
171
    Perform SNMP GETBULK operation.
172
    
173
    Parameters:
174
    - snmpEngine: SNMP engine instance
175
    - authData: Authentication data (community or USM user)
176
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
177
    - contextData: SNMP context information
178
    - nonRepeaters: Number of variables to retrieve only once
179
    - maxRepetitions: Maximum repetitions for remaining variables
180
    - varBinds: Variable bindings to retrieve
181
    - options: Additional options (timeout, retries, etc.)
182
    
183
    Returns:
184
    - errorIndication: Transport/network error or None
185
    - errorStatus: SNMP error status or None
186
    - errorIndex: Index of failed variable or None
187
    - varBinds: Tuple of retrieved variable bindings
188
    """
189
```
190

191
### SNMP WALK Operation
192

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

195
```python { .api }
196
async def walk_cmd(
197
    snmpEngine: SnmpEngine,
198
    authData: CommunityData | UsmUserData,
199
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
200
    contextData: ContextData,
201
    *varBinds: ObjectType,
202
    lexicographicMode: bool = True,
203
    maxRows: int = 0,
204
    ignoreNonIncreasingOid: bool = False,
205
    **options
206
) -> AsyncGenerator[tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]], None]:
207
    """
208
    Perform SNMP WALK operation (async generator).
209
    
210
    Parameters:
211
    - snmpEngine: SNMP engine instance
212
    - authData: Authentication data (community or USM user)
213
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
214
    - contextData: SNMP context information
215
    - varBinds: Starting variable bindings for walk
216
    - lexicographicMode: Continue beyond initial subtree if True
217
    - maxRows: Maximum number of rows to retrieve (0 = unlimited)
218
    - ignoreNonIncreasingOid: Ignore non-increasing OID errors
219
    - options: Additional options (timeout, retries, etc.)
220
    
221
    Yields:
222
    - errorIndication: Transport/network error or None
223
    - errorStatus: SNMP error status or None
224
    - errorIndex: Index of failed variable or None
225
    - varBinds: Tuple of walked variable bindings
226
    """
227
```
228

229
Usage Example:
230

231
```python
232
import asyncio
233
from pysnmp.hlapi.v3arch.asyncio import *
234

235
async def walk_interfaces():
236
    async for (errorIndication, errorStatus, errorIndex, varBinds) in walk_cmd(
237
        SnmpEngine(),
238
        CommunityData('public'),
239
        await UdpTransportTarget.create(('demo.pysnmp.com', 161)),
240
        ContextData(),
241
        ObjectType(ObjectIdentity('1.3.6.1.2.1.2.2.1.2')),  # ifDescr table
242
        lexicographicMode=False,
243
        maxRows=10
244
    ):
245
        if errorIndication:
246
            print(f"Error: {errorIndication}")
247
            break
248
        elif errorStatus:
249
            print(f"SNMP Error: {errorStatus.prettyPrint()}")
250
            break
251
        else:
252
            for varBind in varBinds:
253
                print(f"{varBind[0]} = {varBind[1]}")
254

255
asyncio.run(walk_interfaces())
256
```
257

258
### SNMP BULK WALK Operation
259

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

262
```python { .api }
263
async def bulk_walk_cmd(
264
    snmpEngine: SnmpEngine,
265
    authData: CommunityData | UsmUserData,
266
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
267
    contextData: ContextData,
268
    nonRepeaters: int,
269
    maxRepetitions: int,
270
    *varBinds: ObjectType,
271
    lexicographicMode: bool = True,
272
    maxRows: int = 0,
273
    ignoreNonIncreasingOid: bool = False,
274
    **options
275
) -> AsyncGenerator[tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]], None]:
276
    """
277
    Perform SNMP BULK WALK operation (async generator).
278
    
279
    Parameters:
280
    - snmpEngine: SNMP engine instance
281
    - authData: Authentication data (community or USM user)
282
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
283
    - contextData: SNMP context information
284
    - nonRepeaters: Number of variables to retrieve only once
285
    - maxRepetitions: Maximum repetitions for remaining variables
286
    - varBinds: Starting variable bindings for bulk walk
287
    - lexicographicMode: Continue beyond initial subtree if True
288
    - maxRows: Maximum number of rows to retrieve (0 = unlimited)
289
    - ignoreNonIncreasingOid: Ignore non-increasing OID errors
290
    - options: Additional options (timeout, retries, etc.)
291
    
292
    Yields:
293
    - errorIndication: Transport/network error or None
294
    - errorStatus: SNMP error status or None
295
    - errorIndex: Index of failed variable or None
296
    - varBinds: Tuple of walked variable bindings
297
    """
298
```
299

300
### SNMP Notification Sending
301

302
Sends SNMP TRAP or INFORM notifications to network management systems.
303

304
```python { .api }
305
async def send_notification(
306
    snmpEngine: SnmpEngine,
307
    authData: CommunityData | UsmUserData,
308
    transportTarget: UdpTransportTarget | Udp6TransportTarget,
309
    contextData: ContextData,
310
    notifyType: str,
311
    *varBinds: NotificationType | ObjectType,
312
    **options
313
) -> tuple[ErrorIndication | None, Integer32 | int | None, Integer32 | int | None, tuple[ObjectType, ...]]:
314
    """
315
    Send SNMP notification (TRAP or INFORM).
316
    
317
    Parameters:
318
    - snmpEngine: SNMP engine instance
319
    - authData: Authentication data (community or USM user)
320
    - transportTarget: Transport endpoint (UDP IPv4/IPv6)
321
    - contextData: SNMP context information
322
    - notifyType: 'trap' or 'inform'
323
    - varBinds: Notification variable bindings
324
    - options: Additional options (timeout, retries, etc.)
325
    
326
    Returns:
327
    - errorIndication: Transport/network error or None
328
    - errorStatus: SNMP error status or None
329
    - errorIndex: Index of failed variable or None
330
    - varBinds: Tuple of sent variable bindings
331
    """
332
```
333

334
Usage Example:
335

336
```python
337
import asyncio
338
from pysnmp.hlapi.v3arch.asyncio import *
339

340
async def send_trap():
341
    errorIndication, errorStatus, errorIndex, varBinds = await send_notification(
342
        SnmpEngine(),
343
        CommunityData('public'),
344
        await UdpTransportTarget.create(('demo.pysnmp.com', 162)),
345
        ContextData(),
346
        'trap',
347
        NotificationType(ObjectIdentity('1.3.6.1.4.1.20408.4.1.1.2')),
348
        ObjectType(ObjectIdentity('1.3.6.1.4.1.20408.4.1.1.2.1'), OctetString('Custom trap message'))
349
    )
350
    
351
    if errorIndication:
352
        print(f"Error: {errorIndication}")
353
    elif errorStatus:
354
        print(f"SNMP Error: {errorStatus.prettyPrint()}")
355
    else:
356
        print("Notification sent successfully")
357

358
asyncio.run(send_trap())
359
```
360

361
## Utility Functions
362

363
### End of MIB Detection
364

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

367
```python { .api }
368
def is_end_of_mib(varBinds: tuple[ObjectType, ...]) -> bool:
369
    """
370
    Check if variable bindings indicate end of MIB view.
371
    
372
    Parameters:
373
    - varBinds: Tuple of variable bindings from walk operation
374
    
375
    Returns:
376
    True if end of MIB reached, False otherwise
377
    """
378
```
379

380
Usage Example:
381

382
```python
383
import asyncio
384
from pysnmp.hlapi.v3arch.asyncio import *
385

386
async def walk_with_end_check():
387
    async for (errorIndication, errorStatus, errorIndex, varBinds) in walk_cmd(
388
        SnmpEngine(),
389
        CommunityData('public'),
390
        await UdpTransportTarget.create(('demo.pysnmp.com', 161)),
391
        ContextData(),
392
        ObjectType(ObjectIdentity('1.3.6.1.2.1.1')),
393
        lexicographicMode=False
394
    ):
395
        if errorIndication or errorStatus:
396
            break
397
        
398
        if is_end_of_mib(varBinds):
399
            print("End of MIB reached")
400
            break
401
            
402
        for varBind in varBinds:
403
            print(f"{varBind[0]} = {varBind[1]}")
404

405
asyncio.run(walk_with_end_check())
406
```
407

408
## Transport Targets
409

410
### UDP IPv4 Transport Target
411

412
```python { .api }
413
class UdpTransportTarget:
414
    @classmethod
415
    async def create(
416
        cls,
417
        transportAddr: tuple[str, int],
418
        timeout: float = 1.0,
419
        retries: int = 5,
420
        tagList: bytes = b""
421
    ) -> UdpTransportTarget:
422
        """
423
        Create UDP/IPv4 transport target.
424
        
425
        Parameters:
426
        - transportAddr: Tuple of (hostname/IP, port)
427
        - timeout: Response timeout in seconds
428
        - retries: Maximum number of request retries
429
        - tagList: Arbitrary bytes for endpoint selection
430
        
431
        Returns:
432
        UdpTransportTarget instance
433
        """
434
```
435

436
### UDP IPv6 Transport Target
437

438
```python { .api }
439
class Udp6TransportTarget:
440
    @classmethod
441
    async def create(
442
        cls,
443
        transportAddr: tuple[str, int],
444
        timeout: float = 1.0,
445
        retries: int = 5,
446
        tagList: bytes = b""
447
    ) -> Udp6TransportTarget:
448
        """
449
        Create UDP/IPv6 transport target.
450
        
451
        Parameters:
452
        - transportAddr: Tuple of (hostname/IP, port)
453
        - timeout: Response timeout in seconds
454
        - retries: Maximum number of request retries
455
        - tagList: Arbitrary bytes for endpoint selection
456
        
457
        Returns:
458
        Udp6TransportTarget instance
459
        """
460
```
461

462
## Core Components
463

464
### SNMP Engine
465

466
```python { .api }
467
class SnmpEngine:
468
    def __init__(
469
        self,
470
        snmpEngineID: bytes | None = None,
471
        maxMessageSize: int = 65507,
472
        msgAndPduDsp: object | None = None
473
    ):
474
        """
475
        Create SNMP engine instance.
476
        
477
        Parameters:
478
        - snmpEngineID: Unique engine identifier (auto-generated if None)
479
        - maxMessageSize: Maximum SNMP message size in bytes (default: 65507)
480
        - msgAndPduDsp: Message and PDU dispatcher (internal use, typically None)
481
        
482
        Notes:
483
        - snmpEngineID should be unique per SNMP engine instance
484
        - maxMessageSize affects maximum PDU size for operations
485
        - Auto-generated engine ID includes MAC address and timestamp
486
        """
487
```
488

489
### Context Data
490

491
```python { .api }
492
class ContextData:
493
    def __init__(
494
        self,
495
        contextEngineId: bytes | None = None,
496
        contextName: bytes = b""
497
    ):
498
        """
499
        Create SNMP context data.
500
        
501
        Parameters:
502
        - contextEngineId: SNMP context engine identifier
503
        - contextName: SNMP context name for MIB instance selection
504
        """
505
```
506

507
## Error Handling
508

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

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

515
Common error patterns:
516

517
```python
518
errorIndication, errorStatus, errorIndex, varBinds = await get_cmd(...)
519

520
if errorIndication:
521
    # Network/transport error
522
    print(f"Transport error: {errorIndication}")
523
elif errorStatus:
524
    # SNMP protocol error
525
    print(f"SNMP error: {errorStatus.prettyPrint()} at variable {errorIndex}")
526
else:
527
    # Success - process varBinds
528
    for varBind in varBinds:
529
        print(f"{varBind[0]} = {varBind[1]}")
530
```