Operations

from snmpkit.manager import Manager
 
mgr = Manager(
    host="192.168.1.1",
    port=161,
    community="public",
    version=2,
    timeout=5.0,
    retries=3,
    # SNMPv3 USM (optional)
    user="admin",
    auth_protocol="SHA256",
    auth_password="auth_pass",
    priv_protocol="AES",
    priv_password="priv_pass",
    context_name="",
)

async with Manager("192.168.1.1") as mgr:
    # mgr.connect() called automatically
    value = await mgr.get("1.3.6.1.2.1.1.1.0")
# mgr.close() called automatically

mgr = Manager("192.168.1.1")
await mgr.connect()
try:
    value = await mgr.get("1.3.6.1.2.1.1.1.0")
finally:
    await mgr.close()

async def get(self, oid: str) -> Value

value = await mgr.get("1.3.6.1.2.1.1.1.0")

async def get_many(self, *oids: str) -> list[Value]

values = await mgr.get_many(
    "1.3.6.1.2.1.1.1.0",
    "1.3.6.1.2.1.1.3.0",
)
# values[0] = sysDescr
# values[1] = sysUpTime

async def get_next(self, oid: str) -> tuple[str, Value]

next_oid, value = await mgr.get_next("1.3.6.1.2.1.1.1")
# Returns ("1.3.6.1.2.1.1.1.0", <value>)

async def get_bulk(
    self,
    *oids: str,
    non_repeaters: int = 0,
    max_repetitions: int = 10,
) -> list[tuple[str, Value]]

results = await mgr.get_bulk(
    "1.3.6.1.2.1.2.2.1.1",
    max_repetitions=25,
)
for oid, value in results:
    print(f"{oid} = {value}")

async def walk(self, oid: str) -> AsyncIterator[tuple[str, Value]]

async for oid, value in mgr.walk("1.3.6.1.2.1.1"):
    print(f"{oid} = {value}")

async def bulk_walk(
    self,
    oid: str,
    bulk_size: int = 10,
) -> AsyncIterator[tuple[str, Value]]

async for oid, value in mgr.bulk_walk("1.3.6.1.2.1.2.2", bulk_size=25):
    print(f"{oid} = {value}")

async def set(self, oid: str, value: Value) -> None

from snmpkit.core import Value
 
await mgr.set("1.3.6.1.2.1.1.5.0", Value.OctetString(b"new-hostname"))
await mgr.set("1.3.6.1.2.1.1.6.0", Value.OctetString(b"Server Room"))

async def get_table(
    self,
    base_oid: str,
    columns: list[int] | None = None,
    bulk_size: int = 25,
) -> dict[tuple[int, ...], dict[int, Value]]

table = await mgr.get_table("1.3.6.1.2.1.2.2.1", columns=[2, 8])
 
for index, cols in table.items():
    print(f"Row {index}: {cols}")

async def send_trap(
    self,
    trap_oid: str,
    varbinds: list[tuple[str, Value]] | None = None,
    uptime: int | None = None,
) -> None

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))],
    uptime=123456,
)

async def send_inform(
    self,
    trap_oid: str,
    varbinds: list[tuple[str, Value]] | None = None,
    uptime: int | None = None,
) -> None

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))],
)

from snmpkit.manager import PollTarget, poll_many
 
targets = [
    PollTarget(host="10.0.0.1"),
    PollTarget(host="10.0.0.2", community="private"),
    PollTarget(host="10.0.0.3", version=3, user="admin",
               auth_protocol="SHA256", auth_password="pass"),
]
 
oids = ["1.3.6.1.2.1.1.1.0", "1.3.6.1.2.1.1.3.0"]
 
async for result in poll_many(targets, oids, concurrency=50):
    if result.error:
        print(f"{result.target}: ERROR {result.error}")
    else:
        print(f"{result.target}: {result.oid} = {result.value}")

from snmpkit.manager.exceptions import (
    SnmpError,           # Base exception
    TimeoutError,        # Request timed out
    NoSuchObjectError,   # OID does not exist
    NoSuchInstanceError, # Instance does not exist
    EndOfMibViewError,   # End of MIB reached
    GenericError,        # SNMP error status
)

try:
    await mgr.set("1.3.6.1.2.1.1.1.0", Value.Integer(1))
except GenericError as e:
    print(f"Error status: {e.status}")
    print(f"Error index: {e.index}")