Skip to Content
AgentTraps & Notifications

Traps & Notifications

Send SNMP traps (notifications) from your agent to alert managers about events.

Overview

SNMP traps are asynchronous notifications sent from an agent to a manager. Use them for:

  • Alerting on threshold violations
  • Reporting state changes
  • Notifying about errors or events

Sending Traps

Use send_trap() from within an Updater:

from snmpkit.agent import Updater
from snmpkit.core import Oid, Value, VarBind
 
 
class MonitorUpdater(Updater):
    def __init__(self, threshold: int = 90):
        super().__init__()
        self.threshold = threshold
        self._alerted = False
 
    async def update(self):
        cpu_usage = await get_cpu_usage()
        self.set_GAUGE32("1.0", cpu_usage)
 
        # Send trap when crossing threshold
        if cpu_usage > self.threshold and not self._alerted:
            await self.send_trap(
                "1.3.6.1.4.1.12345.0.1",  # Trap OID
                VarBind(
                    Oid("1.3.6.1.4.1.12345.1.1.0"),
                    Value.Gauge32(cpu_usage),
                ),
            )
            self._alerted = True
        elif cpu_usage <= self.threshold:
            self._alerted = False

send_trap() Reference

async def send_trap(self, oid: str, *varbinds: VarBind) -> None

Parameters:

  • oid - The trap/notification OID identifying the event type
  • varbinds - Variable bindings to include with the trap

Creating VarBinds

Import the core types to create VarBinds:

from snmpkit.core import Oid, Value, VarBind
 
# Integer value
vb1 = VarBind(Oid("1.3.6.1.4.1.12345.1.1.0"), Value.Integer(42))
 
# String value
vb2 = VarBind(Oid("1.3.6.1.4.1.12345.1.2.0"), Value.OctetString(b"error message"))
 
# Counter
vb3 = VarBind(Oid("1.3.6.1.4.1.12345.1.3.0"), Value.Counter64(1000000))

Value Types

MethodDescription
Value.Integer(n)32-bit signed integer
Value.OctetString(b)Bytes (use .encode() for strings)
Value.Counter32(n)32-bit counter
Value.Counter64(n)64-bit counter
Value.Gauge32(n)32-bit gauge
Value.TimeTicks(n)Time in centiseconds
Value.IpAddress(a,b,c,d)IPv4 address (4 octets)
Value.ObjectIdentifier(oid)OID value

Complete Example

import snmpkit
from snmpkit.agent import Agent, Updater
from snmpkit.core import Oid, Value, VarBind
 
 
class TemperatureMonitor(Updater):
    def __init__(self, warning_temp: int = 70, critical_temp: int = 85):
        super().__init__()
        self.warning_temp = warning_temp
        self.critical_temp = critical_temp
        self._last_state = "normal"
 
    async def update(self):
        temp = await read_temperature()
        self.set_GAUGE32("1.0", temp)
 
        # Determine state
        if temp >= self.critical_temp:
            state = "critical"
        elif temp >= self.warning_temp:
            state = "warning"
        else:
            state = "normal"
 
        # Send trap on state change
        if state != self._last_state:
            await self._send_state_trap(state, temp)
            self._last_state = state
 
    async def _send_state_trap(self, state: str, temp: int):
        # Trap OIDs by state
        trap_oids = {
            "warning": "1.3.6.1.4.1.12345.0.1",   # tempWarning
            "critical": "1.3.6.1.4.1.12345.0.2",  # tempCritical
            "normal": "1.3.6.1.4.1.12345.0.3",    # tempNormal
        }
 
        await self.send_trap(
            trap_oids[state],
            VarBind(
                Oid("1.3.6.1.4.1.12345.1.1.0"),  # currentTemp
                Value.Gauge32(temp),
            ),
            VarBind(
                Oid("1.3.6.1.4.1.12345.1.2.0"),  # tempState
                Value.OctetString(state.encode()),
            ),
        )
 
 
async def main():
    agent = Agent()
    agent.register("1.3.6.1.4.1.12345.1", TemperatureMonitor())
    await agent.start()
 
 
snmpkit.run(main())

Configuring snmpd for Traps

To receive traps, configure snmpd to forward them:

# /etc/snmp/snmpd.conf

# Forward traps to a trap receiver
trap2sink localhost:162 public

Or use snmptrapd to receive and log traps:

# Start trap receiver
sudo snmptrapd -f -Lo -c /etc/snmp/snmptrapd.conf

With /etc/snmp/snmptrapd.conf:

authCommunity log,execute,net public

Traps are sent via the AgentX Notify PDU. The master agent (snmpd) then forwards them to configured trap destinations.

Best Practices

  1. Use meaningful trap OIDs - Define a trap OID hierarchy in your MIB
  2. Include context - Add VarBinds that explain the event
  3. Avoid trap storms - Implement rate limiting or state-based sending
  4. Document your traps - Create a MIB file describing your notifications

Rate Limiting Example

import time
 
class RateLimitedUpdater(Updater):
    def __init__(self, min_interval: float = 60.0):
        super().__init__()
        self._last_trap_time = 0
        self._min_interval = min_interval
 
    async def _send_rate_limited_trap(self, oid: str, *varbinds):
        now = time.time()
        if now - self._last_trap_time >= self._min_interval:
            await self.send_trap(oid, *varbinds)
            self._last_trap_time = now

Next Steps

Last updated on
SET HandlerAdvanced