taodeng/network-mcp-server
1
0
Fork 0
mirror of https://github.com/dtzp555-max/network-mcp-server.git synced 2026-07-19 09:43:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Network MCP Server v0.1.0 — SNMP backend with 5 tools

Browse Source 
MVP with zero-dependency SNMP backend for AI-powered network management:
- get_snmp_value: query single OID
- walk_snmp_tree: walk OID subtree
- get_device_info: vendor auto-detection, uptime, location
- get_interfaces: interface status with human-readable output
- discover_devices: subnet SNMP discovery

Tech: FastMCP + pysnmp-lextudio, Python 3.10+
Works with: Cisco, Juniper, Arista, UniFi, Fortinet, and any SNMP device
23 tests passing

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
taodeng
2026-04-05 07:48:18 +10:00
co-authored by Claude Opus 4.6
commit 761bb8c16c
14 changed files with 1155 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+14
View File
@@ -0,0 +1,14 @@
__pycache__/
*.py[cod]
*$py.class
*.egg-info/
dist/
build/
.eggs/
*.egg
.venv/
venv/
.env
.pytest_cache/
.ruff_cache/
.DS_Store
LICENSE
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2026 Tao Deng
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
README.md
+160
View File
@@ -0,0 +1,160 @@
# Network MCP Server
AI-powered network device management via SNMP. Query any vendor's devices — Cisco, Juniper, Arista, UniFi, Fortinet, and more — through the Model Context Protocol.
## Overview
Network MCP Server bridges the gap between AI assistants (Claude, Cursor, etc.) and network infrastructure. It provides 5 SNMP-based tools that let AI query device status, discover networks, and retrieve interface statistics — all through a standardized MCP interface.
**Zero vendor lock-in**: Works with any SNMP-enabled device from any manufacturer.
## Architecture
```
AI Assistant (Claude Code / Cursor / etc.)
↕ MCP Protocol (stdio or HTTP)
Network MCP Server (FastMCP)
↕ SNMP v1/v2c (pysnmp-lextudio)
Network Devices (Cisco / Juniper / Arista / UniFi / ...)
```
The server uses FastMCP for the MCP transport layer and pysnmp-lextudio for SNMP operations. All queries are read-only by design — no configuration changes are made to devices.
## Dependencies
| Package | Version | Purpose |
|---------|---------|---------|
| fastmcp | >=2.0.0 | MCP server framework |
| pysnmp-lextudio | >=6.1.0 | SNMP protocol implementation (actively maintained pysnmp fork) |
| pydantic | >=2.0 | Input validation |
| pyyaml | >=6.0 | Configuration file parsing |
**Runtime**: Python 3.10+
## Usage
### Installation
```bash
pip install network-mcp-server
# or from source:
git clone https://github.com/dtzp555-max/network-mcp-server.git
cd network-mcp-server
pip install -e .
```
### Quick Start
```bash
# Run the server (stdio mode for Claude Code)
network-mcp
# Or run directly
python -m network_mcp.server
```
### Claude Code Integration
Add to `~/.claude/settings.json`:
```json
{
"mcpServers": {
"network": {
"command": "network-mcp",
"env": {
"SNMP_COMMUNITY": "your_community_string"
}
}
}
}
```
Then in Claude Code:
- "Show me the device info for 192.168.1.1"
- "What's the interface status on the core switch?"
- "Discover all SNMP devices on 10.0.0.0/24"
### Available Tools
| Tool | Description |
|------|-------------|
| `get_snmp_value` | Query a single SNMP OID |
| `walk_snmp_tree` | Walk an OID subtree |
| `get_device_info` | Get device name, vendor, uptime, location |
| `get_interfaces` | Get interface status table with traffic stats |
| `discover_devices` | Scan a subnet for SNMP-enabled devices |
## Configuration
### Environment Variables
| Variable | Default | Description |
|----------|---------|-------------|
| `SNMP_COMMUNITY` | `public` | Default SNMP community string |
| `SNMP_TIMEOUT` | `5.0` | Query timeout in seconds |
| `SNMP_RETRIES` | `1` | Number of retries on timeout |
| `NETWORK_MCP_CONFIG_DIR` | `~/.network-mcp` | Config directory path |
### Device Inventory (Optional)
Create `~/.network-mcp/devices.yaml`:
```yaml
devices:
core-switch:
host: 192.168.1.1
community: private_string
firewall:
host: 10.0.0.1
community: fw_community
```
## Error Handling
All tools return structured error responses instead of raising exceptions:
```json
{
"error": "SNMP error: No SNMP response received before timeout",
"host": "192.168.1.1"
}
```
Common error scenarios:
- **Timeout**: Device is unreachable or SNMP is not enabled
- **Authentication**: Wrong community string (no SNMP response)
- **Large subnet**: Subnet scan limited to /22 (1024 hosts) max
## Known Limitations
- **SNMP v1/v2c only** — SNMPv3 support planned for v0.2.0
- **Read-only** — No SET operations by design (safety first)
- **64-bit counters** — ifHCInOctets/ifHCOutOctets used when available, falls back to 32-bit
- **Subnet discovery** — Max /22 (1024 hosts) to prevent accidental network floods
- **No persistent state** — Each query is independent, no caching
## Development
```bash
# Setup
git clone https://github.com/dtzp555-max/network-mcp-server.git
cd network-mcp-server
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
# Test
pytest tests/ -v
# Lint
ruff check src/ tests/
```
## Roadmap
- **v0.2.0**: SNMPv3 support, SSH/Netmiko backend (Cisco IOS commands)
- **v0.3.0**: UniFi REST API backend
- **v0.4.0**: Configuration change tools (with safety guardrails)
- **v1.0.0**: Multi-backend, production-ready
pyproject.toml
+57
View File
@@ -0,0 +1,57 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "network-mcp-server"
version = "0.1.0"
description = "MCP server for network device management via SNMP — query any vendor's devices with AI"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.10"
authors = [{name = "Tao Deng", email = "dtzp555@gmail.com"}]
keywords = ["mcp", "snmp", "network", "cisco", "juniper", "monitoring"]
classifiers = [
"Development Status :: 3 - Alpha",
"Intended Audience :: System Administrators",
"Topic :: System :: Networking :: Monitoring",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
]
dependencies = [
"fastmcp>=2.0.0",
"pysnmp-lextudio>=6.1.0",
"pydantic>=2.0",
"pyyaml>=6.0",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.23",
"ruff>=0.4",
]
[project.urls]
Homepage = "https://github.com/dtzp555-max/network-mcp-server"
Issues = "https://github.com/dtzp555-max/network-mcp-server/issues"
[project.scripts]
network-mcp = "network_mcp.server:main"
[tool.hatch.build.targets.wheel]
packages = ["src/network_mcp"]
[tool.ruff]
line-length = 100
target-version = "py310"
[tool.ruff.lint]
select = ["E", "F", "I", "N", "W"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
src/network_mcp/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Network MCP Server — AI-powered network device management via SNMP."""
__version__ = "0.1.0"
src/network_mcp/config.py
+58
View File
@@ -0,0 +1,58 @@
"""Configuration management for Network MCP Server."""
from __future__ import annotations
import os
from pathlib import Path
import yaml
def get_config_dir() -> Path:
"""Get or create config directory."""
config_dir = Path(os.environ.get("NETWORK_MCP_CONFIG_DIR", "~/.network-mcp")).expanduser()
config_dir.mkdir(parents=True, exist_ok=True)
return config_dir
def get_default_community() -> str:
"""Get default SNMP community string from env or config."""
if env_val := os.environ.get("SNMP_COMMUNITY"):
return env_val
config_file = get_config_dir() / "config.yaml"
if config_file.exists():
with open(config_file) as f:
config = yaml.safe_load(f) or {}
return config.get("snmp", {}).get("community", "public")
return "public"
def get_snmp_timeout() -> float:
"""Get SNMP timeout in seconds."""
if env_val := os.environ.get("SNMP_TIMEOUT"):
return float(env_val)
return 5.0
def get_snmp_retries() -> int:
"""Get SNMP retry count."""
if env_val := os.environ.get("SNMP_RETRIES"):
return int(env_val)
return 1
def load_devices() -> dict[str, dict]:
"""Load device inventory from config file.
Returns dict of {device_name: {host, community, ...}}
"""
config_file = get_config_dir() / "devices.yaml"
if not config_file.exists():
return {}
with open(config_file) as f:
data = yaml.safe_load(f) or {}
return data.get("devices", {})
src/network_mcp/server.py
+117
View File
@@ -0,0 +1,117 @@
"""Network MCP Server — AI-powered network device management via SNMP.
Run: python -m network_mcp.server
Or: network-mcp (after pip install)
"""
from __future__ import annotations
from fastmcp import FastMCP
from network_mcp.snmp.tools import (
snmp_discover,
snmp_get,
snmp_get_device_info,
snmp_get_interfaces,
snmp_walk,
)
mcp = FastMCP(
"Network MCP Server",
description=(
"Query and manage network devices via SNMP. "
"Works with any vendor: Cisco, Juniper, Arista, UniFi, Fortinet, and more."
),
)
@mcp.tool()
async def get_snmp_value(host: str, oid: str, community: str | None = None) -> dict:
"""Query a single SNMP OID from a network device.
Use this for raw SNMP queries when you know the exact OID.
Common OIDs: 1.3.6.1.2.1.1.1.0 (sysDescr), 1.3.6.1.2.1.1.5.0 (sysName)
Args:
host: Device IP address or hostname
oid: SNMP OID to query (dotted notation)
community: SNMP community string (default: from config)
"""
return await snmp_get(host, oid, community)
@mcp.tool()
async def walk_snmp_tree(
host: str,
oid_prefix: str,
community: str | None = None,
max_results: int = 500,
) -> dict:
"""Walk an SNMP OID subtree and return all values underneath it.
Use this to enumerate tables (like interface table, routing table) or
explore what OIDs a device supports.
Args:
host: Device IP address or hostname
oid_prefix: Starting OID for the walk
community: SNMP community string (default: from config)
max_results: Maximum results to return (default: 500, max: 5000)
"""
return await snmp_walk(host, oid_prefix, community, max_results)
@mcp.tool()
async def get_device_info(host: str, community: str | None = None) -> dict:
"""Get comprehensive information about a network device.
Returns device name, description, vendor, OS version, uptime,
contact info, and location. Auto-detects vendor from sysDescr.
Args:
host: Device IP address or hostname
community: SNMP community string (default: from config)
"""
return await snmp_get_device_info(host, community)
@mcp.tool()
async def get_interfaces(host: str, community: str | None = None) -> dict:
"""Get interface status table with human-readable output.
Returns all interfaces with their admin/operational status, speed,
traffic counters (formatted as KB/MB/GB), and error counts.
Args:
host: Device IP address or hostname
community: SNMP community string (default: from config)
"""
return await snmp_get_interfaces(host, community)
@mcp.tool()
async def discover_devices(
subnet: str,
community: str | None = None,
max_concurrent: int = 50,
) -> dict:
"""Discover SNMP-enabled devices in a subnet.
Scans all IPs in the given CIDR range for SNMP-responsive devices.
Returns device name, description, and auto-detected vendor for each.
Args:
subnet: CIDR notation (e.g., "192.168.1.0/24"). Max /22 subnet.
community: SNMP community string (default: from config)
max_concurrent: Max concurrent probes (default: 50, max: 100)
"""
return await snmp_discover(subnet, community, max_concurrent)
def main():
"""Entry point for the Network MCP Server."""
mcp.run()
if __name__ == "__main__":
main()
src/network_mcp/snmp/__init__.py
+1
View File
@@ -0,0 +1 @@
"""SNMP backend for Network MCP Server."""
src/network_mcp/snmp/client.py
+234
View File
@@ -0,0 +1,234 @@
"""SNMP client wrapper using pysnmp-lextudio for async SNMP operations."""
from __future__ import annotations
import asyncio
import ipaddress
from dataclasses import dataclass
from pysnmp.hlapi.v3arch.asyncio import (
CommunityData,
ObjectIdentity,
ObjectType,
SnmpEngine,
UdpTransportTarget,
bulkWalkCmd,
getCmd,
walkCmd,
)
@dataclass
class SnmpResult:
oid: str
value: str
type: str
class SnmpClient:
"""Async SNMP client for querying network devices."""
def __init__(
self,
community: str = "public",
timeout: float = 5.0,
retries: int = 1,
port: int = 161,
):
self.community = community
self.timeout = timeout
self.retries = retries
self.port = port
self._engine = SnmpEngine()
def _community_data(self, community: str | None = None) -> CommunityData:
return CommunityData(community or self.community)
def _transport(self, host: str, port: int | None = None) -> UdpTransportTarget:
return UdpTransportTarget(
(host, port or self.port),
timeout=self.timeout,
retries=self.retries,
)
async def get(
self, host: str, oid: str, community: str | None = None
) -> SnmpResult:
"""SNMP GET for a single OID."""
error_indication, error_status, error_index, var_binds = await getCmd(
self._engine,
self._community_data(community),
self._transport(host),
ObjectType(ObjectIdentity(oid)),
)
if error_indication:
raise SnmpError(f"SNMP error: {error_indication}")
if error_status:
raise SnmpError(
f"SNMP error at {error_index}: {error_status.prettyPrint()}"
)
name, val = var_binds[0]
return SnmpResult(
oid=str(name),
value=str(val),
type=val.__class__.__name__,
)
async def get_bulk(
self, host: str, oids: list[str], community: str | None = None
) -> dict[str, SnmpResult]:
"""SNMP GET for multiple OIDs in one request."""
object_types = [ObjectType(ObjectIdentity(oid)) for oid in oids]
error_indication, error_status, error_index, var_binds = await getCmd(
self._engine,
self._community_data(community),
self._transport(host),
*object_types,
)
if error_indication:
raise SnmpError(f"SNMP error: {error_indication}")
if error_status:
raise SnmpError(
f"SNMP error at {error_index}: {error_status.prettyPrint()}"
)
results = {}
for name, val in var_binds:
results[str(name)] = SnmpResult(
oid=str(name),
value=str(val),
type=val.__class__.__name__,
)
return results
async def walk(
self,
host: str,
oid_prefix: str,
community: str | None = None,
max_results: int = 1000,
) -> list[SnmpResult]:
"""SNMP WALK (GETNEXT) over an OID subtree."""
results = []
count = 0
walk_iter = walkCmd(
self._engine,
self._community_data(community),
self._transport(host),
ObjectType(ObjectIdentity(oid_prefix)),
)
async for error_indication, error_status, error_index, var_binds in walk_iter:
if error_indication:
raise SnmpError(f"SNMP walk error: {error_indication}")
if error_status:
break
for name, val in var_binds:
results.append(
SnmpResult(
oid=str(name),
value=str(val),
type=val.__class__.__name__,
)
)
count += 1
if count >= max_results:
return results
return results
async def bulk_walk(
self,
host: str,
oid_prefix: str,
community: str | None = None,
max_repetitions: int = 25,
max_results: int = 1000,
) -> list[SnmpResult]:
"""SNMP GETBULK walk for faster table retrieval."""
results = []
count = 0
walk_iter = bulkWalkCmd(
self._engine,
self._community_data(community),
self._transport(host),
0,
max_repetitions,
ObjectType(ObjectIdentity(oid_prefix)),
)
async for error_indication, error_status, error_index, var_binds in walk_iter:
if error_indication:
raise SnmpError(f"SNMP bulk walk error: {error_indication}")
if error_status:
break
for name, val in var_binds:
results.append(
SnmpResult(
oid=str(name),
value=str(val),
type=val.__class__.__name__,
)
)
count += 1
if count >= max_results:
return results
return results
async def is_reachable(
self, host: str, community: str | None = None
) -> bool:
"""Check if a host responds to SNMP."""
try:
await self.get(host, "1.3.6.1.2.1.1.1.0", community)
return True
except (SnmpError, asyncio.TimeoutError, OSError):
return False
async def discover_subnet(
self,
subnet: str,
community: str | None = None,
max_concurrent: int = 50,
) -> list[dict]:
"""Discover SNMP-enabled devices in a subnet (CIDR notation)."""
network = ipaddress.ip_network(subnet, strict=False)
hosts = [str(ip) for ip in network.hosts()]
if len(hosts) > 1024:
raise SnmpError(f"Subnet too large: {len(hosts)} hosts. Max /22 (1024).")
semaphore = asyncio.Semaphore(max_concurrent)
async def probe(ip: str) -> dict | None:
async with semaphore:
try:
result = await asyncio.wait_for(
self.get(ip, "1.3.6.1.2.1.1.1.0", community),
timeout=self.timeout + 1,
)
name_result = await self.get(ip, "1.3.6.1.2.1.1.5.0", community)
return {
"host": ip,
"sysDescr": result.value,
"sysName": name_result.value,
}
except (SnmpError, asyncio.TimeoutError, OSError):
return None
tasks = [probe(ip) for ip in hosts]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [r for r in results if isinstance(r, dict)]
class SnmpError(Exception):
"""SNMP operation error."""
src/network_mcp/snmp/oids.py
+138
View File
@@ -0,0 +1,138 @@
"""Standard SNMP OID definitions for network device management.
Reference: RFC 1213 (MIB-II), RFC 2863 (IF-MIB), RFC 2790 (HOST-RESOURCES-MIB)
"""
# --- System MIB (RFC 1213) ---
SYS_DESCR = "1.3.6.1.2.1.1.1.0"
SYS_OBJECT_ID = "1.3.6.1.2.1.1.2.0"
SYS_UPTIME = "1.3.6.1.2.1.1.3.0"
SYS_CONTACT = "1.3.6.1.2.1.1.4.0"
SYS_NAME = "1.3.6.1.2.1.1.5.0"
SYS_LOCATION = "1.3.6.1.2.1.1.6.0"
SYS_SERVICES = "1.3.6.1.2.1.1.7.0"
# System info OIDs for get_device_info
SYSTEM_OIDS = [SYS_DESCR, SYS_OBJECT_ID, SYS_UPTIME, SYS_CONTACT, SYS_NAME, SYS_LOCATION]
# --- Interfaces MIB (RFC 2863) ---
IF_NUMBER = "1.3.6.1.2.1.2.1.0"
IF_TABLE = "1.3.6.1.2.1.2.2"
IF_INDEX = "1.3.6.1.2.1.2.2.1.1"
IF_DESCR = "1.3.6.1.2.1.2.2.1.2"
IF_TYPE = "1.3.6.1.2.1.2.2.1.3"
IF_MTU = "1.3.6.1.2.1.2.2.1.4"
IF_SPEED = "1.3.6.1.2.1.2.2.1.5"
IF_PHYS_ADDRESS = "1.3.6.1.2.1.2.2.1.6"
IF_ADMIN_STATUS = "1.3.6.1.2.1.2.2.1.7"
IF_OPER_STATUS = "1.3.6.1.2.1.2.2.1.8"
IF_IN_OCTETS = "1.3.6.1.2.1.2.2.1.10"
IF_IN_ERRORS = "1.3.6.1.2.1.2.2.1.14"
IF_OUT_OCTETS = "1.3.6.1.2.1.2.2.1.16"
IF_OUT_ERRORS = "1.3.6.1.2.1.2.2.1.20"
# IF-MIB extensions (64-bit counters)
IF_X_TABLE = "1.3.6.1.2.1.31.1.1"
IF_NAME = "1.3.6.1.2.1.31.1.1.1.1"
IF_HC_IN_OCTETS = "1.3.6.1.2.1.31.1.1.1.6"
IF_HC_OUT_OCTETS = "1.3.6.1.2.1.31.1.1.1.10"
IF_HIGH_SPEED = "1.3.6.1.2.1.31.1.1.1.15"
IF_ALIAS = "1.3.6.1.2.1.31.1.1.1.18"
# Interface columns for walk
INTERFACE_COLUMNS = {
IF_DESCR: "ifDescr",
IF_TYPE: "ifType",
IF_SPEED: "ifSpeed",
IF_ADMIN_STATUS: "ifAdminStatus",
IF_OPER_STATUS: "ifOperStatus",
IF_IN_OCTETS: "ifInOctets",
IF_OUT_OCTETS: "ifOutOctets",
IF_IN_ERRORS: "ifInErrors",
IF_OUT_ERRORS: "ifOutErrors",
}
IF_X_COLUMNS = {
IF_NAME: "ifName",
IF_HC_IN_OCTETS: "ifHCInOctets",
IF_HC_OUT_OCTETS: "ifHCOutOctets",
IF_HIGH_SPEED: "ifHighSpeed",
IF_ALIAS: "ifAlias",
}
# --- Status value mappings ---
IF_OPER_STATUS_MAP = {1: "up", 2: "down", 3: "testing", 4: "unknown", 5: "dormant", 6: "notPresent", 7: "lowerLayerDown"}
IF_ADMIN_STATUS_MAP = {1: "up", 2: "down", 3: "testing"}
# --- Vendor detection patterns in sysDescr ---
VENDOR_PATTERNS = [
("Cisco", "cisco"),
("Juniper", "juniper"),
("Arista", "arista"),
("Ubiquiti", "ubnt"),
("Ubiquiti", "ubiquiti"),
("Ubiquiti", "unifi"),
("Huawei", "huawei"),
("MikroTik", "routeros"),
("MikroTik", "mikrotik"),
("Fortinet", "fortinet"),
("Fortinet", "fortigate"),
("Palo Alto", "paloalto"),
("Palo Alto", "pan-os"),
("HP/Aruba", "procurve"),
("HP/Aruba", "aruba"),
("Dell", "dell"),
("Extreme", "extreme"),
("Brocade", "brocade"),
("Nokia", "nokia"),
("Linux", "linux"),
("FreeBSD", "freebsd"),
("Windows", "windows"),
]
def detect_vendor(sys_descr: str) -> str:
"""Detect vendor from sysDescr string."""
lower = sys_descr.lower()
for vendor_name, pattern in VENDOR_PATTERNS:
if pattern in lower:
return vendor_name
return "Unknown"
def format_uptime(timeticks: int) -> str:
"""Convert SNMP timeticks (1/100th second) to human-readable format."""
total_seconds = timeticks // 100
days = total_seconds // 86400
hours = (total_seconds % 86400) // 3600
minutes = (total_seconds % 3600) // 60
seconds = total_seconds % 60
if days > 0:
return f"{days}d {hours}h {minutes}m {seconds}s"
if hours > 0:
return f"{hours}h {minutes}m {seconds}s"
return f"{minutes}m {seconds}s"
def format_speed(speed_bps: int) -> str:
"""Convert speed in bps to human-readable format."""
if speed_bps >= 1_000_000_000:
return f"{speed_bps / 1_000_000_000:.0f} Gbps"
if speed_bps >= 1_000_000:
return f"{speed_bps / 1_000_000:.0f} Mbps"
if speed_bps >= 1_000:
return f"{speed_bps / 1_000:.0f} Kbps"
return f"{speed_bps} bps"
def format_bytes(octets: int) -> str:
"""Convert octets to human-readable format."""
if octets >= 1_099_511_627_776:
return f"{octets / 1_099_511_627_776:.2f} TB"
if octets >= 1_073_741_824:
return f"{octets / 1_073_741_824:.2f} GB"
if octets >= 1_048_576:
return f"{octets / 1_048_576:.2f} MB"
if octets >= 1_024:
return f"{octets / 1_024:.2f} KB"
return f"{octets} B"
src/network_mcp/snmp/tools.py
+243
View File
@@ -0,0 +1,243 @@
"""SNMP MCP tools — registered with FastMCP server."""
from __future__ import annotations
from network_mcp.config import get_default_community, get_snmp_retries, get_snmp_timeout
from network_mcp.snmp.client import SnmpClient, SnmpError
from network_mcp.snmp.oids import (
IF_ADMIN_STATUS,
IF_ADMIN_STATUS_MAP,
IF_DESCR,
IF_IN_ERRORS,
IF_IN_OCTETS,
IF_NAME,
IF_OPER_STATUS,
IF_OPER_STATUS_MAP,
IF_OUT_ERRORS,
IF_OUT_OCTETS,
IF_SPEED,
SYSTEM_OIDS,
detect_vendor,
format_bytes,
format_speed,
format_uptime,
)
def _client(community: str | None = None) -> SnmpClient:
return SnmpClient(
community=community or get_default_community(),
timeout=get_snmp_timeout(),
retries=get_snmp_retries(),
)
async def snmp_get(host: str, oid: str, community: str | None = None) -> dict:
"""Query a single SNMP OID from a device.
Args:
host: IP address or hostname of the device
oid: SNMP OID to query (e.g., "1.3.6.1.2.1.1.1.0" for sysDescr)
community: SNMP community string (default: from config or "public")
Returns:
dict with oid, value, and type fields
"""
try:
result = await _client(community).get(host, oid, community)
return {"oid": result.oid, "value": result.value, "type": result.type}
except SnmpError as e:
return {"error": str(e), "host": host, "oid": oid}
async def snmp_walk(
host: str,
oid_prefix: str,
community: str | None = None,
max_results: int = 500,
) -> dict:
"""Walk an SNMP OID subtree and return all values.
Args:
host: IP address or hostname of the device
oid_prefix: Starting OID for the walk (e.g., "1.3.6.1.2.1.2.2" for ifTable)
community: SNMP community string (default: from config or "public")
max_results: Maximum number of results (default: 500, max: 5000)
Returns:
dict with results list and count
"""
max_results = min(max_results, 5000)
try:
results = await _client(community).bulk_walk(
host, oid_prefix, community, max_results=max_results
)
return {
"host": host,
"oid_prefix": oid_prefix,
"count": len(results),
"results": [{"oid": r.oid, "value": r.value, "type": r.type} for r in results],
}
except SnmpError as e:
return {"error": str(e), "host": host, "oid_prefix": oid_prefix}
async def snmp_get_device_info(host: str, community: str | None = None) -> dict:
"""Get comprehensive device information (vendor, model, uptime, location).
Args:
host: IP address or hostname of the device
community: SNMP community string (default: from config or "public")
Returns:
dict with sysName, sysDescr, vendor, uptime, contact, location
"""
try:
client = _client(community)
results = await client.get_bulk(host, SYSTEM_OIDS, community)
sys_descr = results.get("1.3.6.1.2.1.1.1.0")
sys_name = results.get("1.3.6.1.2.1.1.5.0")
sys_uptime = results.get("1.3.6.1.2.1.1.3.0")
sys_contact = results.get("1.3.6.1.2.1.1.4.0")
sys_location = results.get("1.3.6.1.2.1.1.6.0")
sys_oid = results.get("1.3.6.1.2.1.1.2.0")
descr_str = sys_descr.value if sys_descr else ""
vendor = detect_vendor(descr_str)
uptime_raw = int(sys_uptime.value) if sys_uptime else 0
uptime_str = format_uptime(uptime_raw) if uptime_raw else "unknown"
return {
"host": host,
"sysName": sys_name.value if sys_name else "",
"sysDescr": descr_str,
"vendor": vendor,
"sysObjectID": sys_oid.value if sys_oid else "",
"uptime": uptime_str,
"uptimeTicks": uptime_raw,
"sysContact": sys_contact.value if sys_contact else "",
"sysLocation": sys_location.value if sys_location else "",
}
except SnmpError as e:
return {"error": str(e), "host": host}
async def snmp_get_interfaces(host: str, community: str | None = None) -> dict:
"""Get interface status table with human-readable output.
Args:
host: IP address or hostname of the device
community: SNMP community string (default: from config or "public")
Returns:
dict with interface list including name, status, speed, traffic counters
"""
try:
client = _client(community)
# Walk ifDescr to get interface count and indices
descr_results = await client.walk(host, IF_DESCR, community)
if not descr_results:
return {"host": host, "count": 0, "interfaces": []}
# Extract interface indices
indices = []
descr_map = {}
for r in descr_results:
idx = r.oid.split(".")[-1]
indices.append(idx)
descr_map[idx] = r.value
# Walk additional columns
name_results = await client.walk(host, IF_NAME, community)
speed_results = await client.walk(host, IF_SPEED, community)
oper_results = await client.walk(host, IF_OPER_STATUS, community)
admin_results = await client.walk(host, IF_ADMIN_STATUS, community)
in_octets_results = await client.walk(host, IF_IN_OCTETS, community)
out_octets_results = await client.walk(host, IF_OUT_OCTETS, community)
in_errors_results = await client.walk(host, IF_IN_ERRORS, community)
out_errors_results = await client.walk(host, IF_OUT_ERRORS, community)
def _map_by_index(results: list) -> dict[str, str]:
return {r.oid.split(".")[-1]: r.value for r in results}
name_map = _map_by_index(name_results)
speed_map = _map_by_index(speed_results)
oper_map = _map_by_index(oper_results)
admin_map = _map_by_index(admin_results)
in_oct_map = _map_by_index(in_octets_results)
out_oct_map = _map_by_index(out_octets_results)
in_err_map = _map_by_index(in_errors_results)
out_err_map = _map_by_index(out_errors_results)
interfaces = []
for idx in indices:
speed_raw = int(speed_map.get(idx, 0))
in_octets = int(in_oct_map.get(idx, 0))
out_octets = int(out_oct_map.get(idx, 0))
oper_raw = int(oper_map.get(idx, 0))
admin_raw = int(admin_map.get(idx, 0))
interfaces.append({
"index": int(idx),
"ifName": name_map.get(idx, ""),
"ifDescr": descr_map.get(idx, ""),
"adminStatus": IF_ADMIN_STATUS_MAP.get(admin_raw, f"unknown({admin_raw})"),
"operStatus": IF_OPER_STATUS_MAP.get(oper_raw, f"unknown({oper_raw})"),
"speed": format_speed(speed_raw),
"speedBps": speed_raw,
"inTraffic": format_bytes(in_octets),
"outTraffic": format_bytes(out_octets),
"inErrors": int(in_err_map.get(idx, 0)),
"outErrors": int(out_err_map.get(idx, 0)),
})
return {
"host": host,
"count": len(interfaces),
"interfaces": interfaces,
}
except SnmpError as e:
return {"error": str(e), "host": host}
async def snmp_discover(
subnet: str,
community: str | None = None,
max_concurrent: int = 50,
) -> dict:
"""Discover SNMP-enabled devices in a subnet.
Args:
subnet: CIDR notation (e.g., "192.168.1.0/24"). Max /22 (1024 hosts).
community: SNMP community string (default: from config or "public")
max_concurrent: Maximum concurrent probes (default: 50)
Returns:
dict with discovered devices list (host, sysName, sysDescr, vendor)
"""
try:
client = _client(community)
raw_results = await client.discover_subnet(
subnet, community, max_concurrent=min(max_concurrent, 100)
)
devices = []
for r in raw_results:
vendor = detect_vendor(r.get("sysDescr", ""))
devices.append({
"host": r["host"],
"sysName": r.get("sysName", ""),
"sysDescr": r.get("sysDescr", ""),
"vendor": vendor,
})
return {
"subnet": subnet,
"devicesFound": len(devices),
"devices": devices,
}
except SnmpError as e:
return {"error": str(e), "subnet": subnet}
tests/__init__.py
View File
tests/test_config.py
+27
View File
@@ -0,0 +1,27 @@
"""Tests for configuration module."""
import os
from network_mcp.config import get_default_community, get_snmp_timeout
def test_default_community_from_env(monkeypatch):
monkeypatch.setenv("SNMP_COMMUNITY", "my_secret")
assert get_default_community() == "my_secret"
def test_default_community_fallback(monkeypatch):
monkeypatch.delenv("SNMP_COMMUNITY", raising=False)
# Without config file, should return "public"
monkeypatch.setenv("NETWORK_MCP_CONFIG_DIR", "/tmp/nonexistent-mcp-test")
assert get_default_community() == "public"
def test_snmp_timeout_from_env(monkeypatch):
monkeypatch.setenv("SNMP_TIMEOUT", "10.0")
assert get_snmp_timeout() == 10.0
def test_snmp_timeout_default(monkeypatch):
monkeypatch.delenv("SNMP_TIMEOUT", raising=False)
assert get_snmp_timeout() == 5.0
tests/test_oids.py
+82
View File
@@ -0,0 +1,82 @@
"""Tests for SNMP OID utilities."""
from network_mcp.snmp.oids import detect_vendor, format_bytes, format_speed, format_uptime
def test_detect_vendor_cisco():
assert detect_vendor("Cisco IOS Software, C2960 Software") == "Cisco"
def test_detect_vendor_juniper():
assert detect_vendor("Juniper Networks, Inc. ex4300") == "Juniper"
def test_detect_vendor_ubiquiti():
assert detect_vendor("Linux ubnt 4.14.0 UniFi Dream Machine") == "Ubiquiti"
def test_detect_vendor_arista():
assert detect_vendor("Arista Networks EOS version 4.28") == "Arista"
def test_detect_vendor_fortinet():
assert detect_vendor("FortiGate-60F v7.2.0") == "Fortinet"
def test_detect_vendor_mikrotik():
assert detect_vendor("RouterOS 7.12") == "MikroTik"
def test_detect_vendor_unknown():
assert detect_vendor("Some Unknown Device") == "Unknown"
def test_detect_vendor_case_insensitive():
assert detect_vendor("CISCO IOS XE") == "Cisco"
def test_format_uptime_days():
# 2 days, 3 hours, 4 minutes, 5 seconds = 183845 seconds * 100
assert format_uptime(18384500) == "2d 3h 4m 5s"
def test_format_uptime_hours():
assert format_uptime(370500) == "1h 1m 45s"
def test_format_uptime_minutes():
assert format_uptime(12000) == "2m 0s"
def test_format_speed_gbps():
assert format_speed(1_000_000_000) == "1 Gbps"
assert format_speed(10_000_000_000) == "10 Gbps"
def test_format_speed_mbps():
assert format_speed(100_000_000) == "100 Mbps"
assert format_speed(1_000_000) == "1 Mbps"
def test_format_speed_kbps():
assert format_speed(64_000) == "64 Kbps"
def test_format_bytes_tb():
assert "TB" in format_bytes(2_000_000_000_000)
def test_format_bytes_gb():
assert "GB" in format_bytes(5_000_000_000)
def test_format_bytes_mb():
assert "MB" in format_bytes(50_000_000)
def test_format_bytes_kb():
assert "KB" in format_bytes(50_000)
def test_format_bytes_bytes():
assert format_bytes(500) == "500 B"