Traps & Notifications

Send and receive SNMP traps and notifications.

Overview

SNMP devices send unsolicited messages to report events:

Type	Version	Confirmation
Trap (v1)	SNMPv1	No
Trap (v2c/v3)	SNMPv2c, SNMPv3	No
Inform	SNMPv2c, SNMPv3	Yes (acknowledged)

Sending Traps

Use the Manager to send traps and informs.

Send Trap (Fire-and-Forget)


import asyncio
from snmpkit.core import Value
from snmpkit.manager import Manager
 
async def main():
    async with Manager("192.168.1.100", port=162, community="public") as mgr:
        await mgr.send_trap(
            trap_oid="1.3.6.1.4.1.12345.0.1",
            varbinds=[
                ("1.3.6.1.4.1.12345.1.1.0", Value.Integer(95)),
                ("1.3.6.1.4.1.12345.1.2.0", Value.OctetString(b"CPU High")),
            ],
            uptime=123456,
        )
 
asyncio.run(main())

Traps are fire-and-forget — no response is expected. The sysUpTime.0 and snmpTrapOID.0 varbinds are added automatically.

Send Inform (Acknowledged)


async with Manager("192.168.1.100", port=162) as mgr:
    await mgr.send_inform(
        trap_oid="1.3.6.1.4.1.12345.0.1",
        varbinds=[
            ("1.3.6.1.4.1.12345.1.1.0", Value.Integer(95)),
        ],
    )
    print("Inform acknowledged by receiver")

Inform requests wait for an acknowledgement Response PDU. Use traps when acknowledgment isn’t needed.

SNMPv3 Traps


async with Manager(
    "192.168.1.100",
    port=162,
    version=3,
    user="trapuser",
    auth_protocol="SHA256",
    auth_password="auth_password_here",
    priv_protocol="AES",
    priv_password="priv_password_here",
) as mgr:
    await mgr.send_trap(
        trap_oid="1.3.6.1.4.1.12345.0.1",
        varbinds=[("1.3.6.1.4.1.12345.1.1.0", Value.Integer(42))],
    )

Receiving Traps

TrapReceiver

Listen for incoming traps with the async iterator pattern:


import asyncio
from snmpkit.manager import TrapReceiver
 
async def main():
    async with TrapReceiver(port=1162) as receiver:
        async for trap in receiver:
            print(f"Trap from {trap.source}")
            print(f"  OID: {trap.trap_oid}")
            print(f"  Uptime: {trap.uptime}")
            print(f"  Community: {trap.community}")
 
            for oid, value in trap.varbinds:
                print(f"  {oid} = {value}")
 
asyncio.run(main())

Port 162 requires root/administrator privileges. Use a higher port (e.g., 1162) for testing.

TrapMessage

Each received trap is a TrapMessage dataclass:


@dataclass
class TrapMessage:
    source: tuple[str, int]              # (ip, port)
    version: int                         # 1 or 3
    community: str                       # v2c community string
    pdu_type: int                        # 0xA7 (Trap) or 0xA6 (Inform)
    request_id: int                      # PDU request ID
    trap_oid: str                        # snmpTrapOID.0 value
    uptime: int                          # sysUpTime.0 value
    varbinds: list[tuple[str, Value]]    # User varbinds (without sysUpTime/snmpTrapOID)
    raw_varbinds: list[tuple[str, Value]]  # All varbinds including standard ones

Automatic Inform ACK

InformRequest PDUs are automatically acknowledged with a Response PDU. You don’t need to handle this — the receiver sends the ACK and delivers the message to your iterator like any other trap.

Manual Start/Stop


receiver = TrapReceiver(host="0.0.0.0", port=1162)
await receiver.start()
 
# Process traps...
trap = await receiver.__anext__()
 
await receiver.stop()

Common Trap OIDs

OID	Name	Description
`1.3.6.1.6.3.1.1.5.1`	coldStart	Agent reinitializing
`1.3.6.1.6.3.1.1.5.2`	warmStart	Agent reinitializing (config preserved)
`1.3.6.1.6.3.1.1.5.3`	linkDown	Interface went down
`1.3.6.1.6.3.1.1.5.4`	linkUp	Interface came up
`1.3.6.1.6.3.1.1.5.5`	authenticationFailure	Bad community/credentials

Example: Trap Logger

Console


import asyncio
from snmpkit.manager import TrapReceiver
 
async def main():
    async with TrapReceiver(port=1162) as receiver:
        print("Listening for traps on port 1162...")
        async for trap in receiver:
            pdu = "Inform" if trap.pdu_type == 0xA6 else "Trap"
            print(f"[{pdu}] {trap.source[0]} -> {trap.trap_oid}")
            for oid, value in trap.varbinds:
                print(f"  {oid} = {value}")
 
asyncio.run(main())

Example: Trap Forwarder


import asyncio
from snmpkit.manager import Manager, TrapReceiver
 
async def main():
    async with TrapReceiver(port=1162) as receiver:
        async with Manager("10.0.0.1", port=162) as mgr:
            async for trap in receiver:
                # Forward as inform to upstream collector
                await mgr.send_inform(
                    trap_oid=trap.trap_oid,
                    varbinds=trap.varbinds,
                    uptime=trap.uptime,
                )
 
asyncio.run(main())

Next Steps

SNMPv3 — Secure trap configuration
Operations — Full API reference