Files
FabledSteward/plugins/snmp/__init__.py
T
bvandeusenandClaude Opus 5.5 eee8469a17
CI / lint (push) Successful in 2s
CI / unit (push) Successful in 1m26s
CI / integration (push) Successful in 1m33s
CI / publish (push) Successful in 19s
feat(mcp): SNMP devices and Prometheus targets: their state, readings and series (#5840)
The snmp plugin registers list_snmp_devices and get_snmp_device; the
prometheus plugin registers list_prometheus_targets and
get_prometheus_target. Both are viewer-level, read what the poller and
scraper stored, and exist only while their plugin is enabled.
get_overview names a device or target whose last poll or scrape failed;
stale, pending and off are left out.

A device's community and v3 keys and a target's token are never
returned, and a password in a target URL is masked.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 13:31:46 -04:00

52 lines
1.3 KiB
Python

# plugins/snmp/__init__.py
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from quart import Quart
_app: "Quart | None" = None
def setup(app: "Quart") -> None:
global _app
_app = app
from .models import SnmpDevice # noqa: F401 — registers the table
def get_scheduled_tasks() -> list:
from .scheduler import make_poll_task
return [make_poll_task(_app)]
def get_blueprint():
from .routes import snmp_bp
return snmp_bp
def get_nav() -> list[dict]:
# Surfaces in the sidebar's Infrastructure group (SNMP device readings).
return [{"label": "SNMP", "href": "/plugins/snmp/"}]
def get_alert_metrics() -> dict[str, list[str]]:
"""SNMP metric names are the OID labels each device is configured with, so
the list is empty here and the rule form offers the labels recorded so far."""
return {"snmp": []}
def get_audit_entities() -> dict:
"""Audit rows about a device name it and link to its page."""
from .models import SnmpDevice
return {"snmp_device": (SnmpDevice, "/plugins/snmp/devices/{}")}
def register_mcp_tools(register) -> None:
from .mcp_tools import register_tools
register_tools(register)
async def mcp_overview(db) -> list[dict]:
from .mcp_tools import mcp_overview as findings
return await findings(db)