Contexts

from snmpkit.manager import Manager
 
async with Manager("192.168.1.1", version=2) as mgr:
    # Default context (empty string)
    value = await mgr.get("1.3.6.1.2.1.1.1.0")
 
    # Specific context
    value = await mgr.get("1.3.6.1.2.1.1.1.0", context="vlan100")

# Traditional community string
async with Manager("192.168.1.1", community="public") as mgr:
    value = await mgr.get("1.3.6.1.2.1.1.1.0")
 
# Community with context (Cisco format)
async with Manager("192.168.1.1", community="public@vlan100") as mgr:
    value = await mgr.get("1.3.6.1.2.1.17.4.3.1.1")  # dot1dTpFdbAddress

from snmpkit.manager import Manager, USMUser
 
user = USMUser(username="admin", auth_key="password", auth_protocol="SHA256")
 
async with Manager("192.168.1.1", version=3, user=user) as mgr:
    # Query default context
    value = await mgr.get("1.3.6.1.2.1.1.1.0")
 
    # Query specific context
    value = await mgr.get("1.3.6.1.2.1.1.1.0", context="vrf-mgmt")
 
    # Walk with context
    async for oid, value in mgr.bulk_walk("1.3.6.1.2.1.17.4.3", context="vlan100"):
        print(f"{oid} = {value}")

async def get_mac_table(host: str, vlan: int) -> dict:
    community = f"public@{vlan}"
    macs = {}
 
    async with Manager(host, community=community) as mgr:
        async for oid, value in mgr.bulk_walk("1.3.6.1.2.1.17.4.3.1.1"):
            idx = oid.split(".")[-6:]
            mac = ":".join(f"{int(x):02x}" for x in idx)
            macs[mac] = value
 
    return macs
 
# Get MAC addresses for VLAN 100
macs = await get_mac_table("192.168.1.1", 100)

async def get_all_vlans(host: str) -> list[int]:
    vlans = []
    async with Manager(host) as mgr:
        async for oid, value in mgr.bulk_walk("1.3.6.1.4.1.9.9.46.1.3.1.1.2"):
            vlan_id = int(oid.split(".")[-1])
            if value == 1:  # operational
                vlans.append(vlan_id)
    return vlans
 
async def get_full_mac_table(host: str) -> dict:
    vlans = await get_all_vlans(host)
    all_macs = {}
 
    for vlan in vlans:
        macs = await get_mac_table(host, vlan)
        for mac, port in macs.items():
            all_macs[mac] = {"vlan": vlan, "port": port}
 
    return all_macs

async def get_vrf_routes(host: str, vrf: str) -> list[dict]:
    routes = []
 
    async with Manager(host, version=3, user=user) as mgr:
        async for oid, value in mgr.bulk_walk(
            "1.3.6.1.2.1.4.24.4.1",  # ipCidrRouteTable
            context=vrf,
        ):
            # Parse route entry
            routes.append(parse_route(oid, value))
 
    return routes

async def compare_vrfs(host: str, vrf1: str, vrf2: str):
    async with Manager(host, version=3, user=user) as mgr:
        routes1 = set()
        async for oid, _ in mgr.bulk_walk("1.3.6.1.2.1.4.24.4.1.1", context=vrf1):
            routes1.add(oid)
 
        routes2 = set()
        async for oid, _ in mgr.bulk_walk("1.3.6.1.2.1.4.24.4.1.1", context=vrf2):
            routes2.add(oid)
 
        only_in_vrf1 = routes1 - routes2
        only_in_vrf2 = routes2 - routes1
 
        print(f"Only in {vrf1}: {len(only_in_vrf1)}")
        print(f"Only in {vrf2}: {len(only_in_vrf2)}")

async with Manager("192.168.1.1", version=3, user=user) as mgr:
    # Context engine defaults to the remote agent's engine
    value = await mgr.get("1.3.6.1.2.1.1.1.0", context="mycontext")
 
    # Explicit context engine ID (for proxy scenarios)
    value = await mgr.get(
        "1.3.6.1.2.1.1.1.0",
        context="mycontext",
        context_engine_id=bytes.fromhex("80001234..."),
    )

async def poll_all_contexts(host: str, oid: str, contexts: list[str]) -> dict:
    results = {}
 
    async with Manager(host, version=3, user=user) as mgr:
        for ctx in contexts:
            try:
                value = await mgr.get(oid, context=ctx)
                results[ctx] = value
            except Exception as e:
                results[ctx] = f"error: {e}"
 
    return results
 
# Poll interface status across VRFs
results = await poll_all_contexts(
    "192.168.1.1",
    "1.3.6.1.2.1.2.2.1.8.1",  # ifOperStatus.1
    ["", "vrf-a", "vrf-b"],
)

import asyncio
 
async def poll_contexts_parallel(host: str, oid: str, contexts: list[str]) -> dict:
    async def query_context(ctx: str):
        async with Manager(host, version=3, user=user) as mgr:
            try:
                return (ctx, await mgr.get(oid, context=ctx))
            except Exception as e:
                return (ctx, f"error: {e}")
 
    results = await asyncio.gather(*[query_context(ctx) for ctx in contexts])
    return dict(results)

# VLAN-based (community@vlan)
async with Manager("switch", community="public@100") as mgr:
    pass
 
# VRF-based (SNMPv3 context)
async with Manager("router", version=3, user=user) as mgr:
    await mgr.get("1.3.6.1.2.1.4.1.0", context="vrf:customer-a")

# Routing instance context
async with Manager("router", version=3, user=user) as mgr:
    await mgr.get("1.3.6.1.2.1.4.1.0", context="customer-vrf")

# VRF context
async with Manager("switch", version=3, user=user) as mgr:
    await mgr.get("1.3.6.1.2.1.4.1.0", context="vrf-mgmt")

from snmpkit.manager.exceptions import ContextError, NoSuchContextError
 
try:
    async with Manager("192.168.1.1", version=3, user=user) as mgr:
        value = await mgr.get("1.3.6.1.2.1.1.1.0", context="nonexistent")
except NoSuchContextError:
    print("Context does not exist on this device")
except ContextError as e:
    print(f"Context error: {e}")

async def list_contexts(host: str) -> list[str]:
    contexts = []
 
    async with Manager(host, version=3, user=user) as mgr:
        # vacmContextTable
        async for oid, value in mgr.bulk_walk("1.3.6.1.6.3.16.1.1"):
            context_name = str(value)
            contexts.append(context_name)
 
    return contexts