Initial commit: Cisco Switch MCP with fixes C3, C4, C5, C6, S2

This commit is contained in:
root
2026-07-04 20:15:11 +00:00
commit 68e87cf6a5
30 changed files with 727 additions and 0 deletions

12
.env.example Normal file
View File

@@ -0,0 +1,12 @@
# Catalyst 3650 / IOS-XE SSH credentials
CISCO_SWITCH_HOST=10.77.30.40
CISCO_SWITCH_PORT=22
CISCO_SWITCH_USERNAME=REPLACE_WITH_YOUR_USERNAME
CISCO_SWITCH_PASSWORD=REPLACE_WITH_YOUR_PASSWORD
CISCO_SWITCH_ENABLE_PASSWORD=
CISCO_SWITCH_DEVICE_TYPE=cisco_xe
# Safety
CISCO_SWITCH_ALLOW_WRITES=false
CISCO_SWITCH_DRY_RUN=false
CISCO_SWITCH_AUDIT_LOG_PATH=/root/.hermes/logs/cisco-switch-mcp-audit.jsonl

58
README.md Normal file
View File

@@ -0,0 +1,58 @@
# Cisco Switch MCP
MCP server for **Cisco Catalyst 3650** and other **IOS-XE** switches. Connects via SSH using [Netmiko](https://github.com/ktbyers/netmiko).
## Prerequisites on the switch
1. **SSH enabled**: `ip domain-name local.lan` + `crypto key generate rsa` + `ip ssh version 2`
2. **User account** with privilege 15 (or enable password configured)
3. **Management IP** reachable from the Hermes LXC
## Install
```bash
/usr/local/lib/hermes-agent/venv/bin/pip install -e /root/cisco-switch-mcp
```
## Hermes config
Add to `~/.hermes/config.yaml`:
```yaml
mcp_servers:
cisco-switch:
command: /usr/local/lib/hermes-agent/venv/bin/cisco-switch-mcp
timeout: 120
enabled: true
env:
CISCO_SWITCH_HOST: "10.77.30.40"
CISCO_SWITCH_USERNAME: "admin"
CISCO_SWITCH_PASSWORD: "your_password"
CISCO_SWITCH_DEVICE_TYPE: "cisco_xe"
CISCO_SWITCH_ALLOW_WRITES: "false"
```
Or: `echo Y | hermes mcp add cisco-switch --command /usr/local/lib/hermes-agent/venv/bin/cisco-switch-mcp --env CISCO_SWITCH_HOST=...`
Restart dashboard/gateway and start a **new chat session**.
## Tools (40+)
| Category | Examples |
|----------|----------|
| System | version, inventory, CPU, memory, environment, stack status |
| Interfaces | status, descriptions, counters, port-channels |
| VLAN/STP | vlan brief, spanning-tree |
| L2 | MAC table, CDP/LLDP neighbors, ARP |
| Diagnostics | logging, ping, traceroute, running-config |
| Writes | shutdown port, set access VLAN, description, save config |
## Safety
- Writes disabled by default (`CISCO_SWITCH_ALLOW_WRITES=false`)
- Blocks `reload`, `write erase`, `format`
- Audit log at `CISCO_SWITCH_AUDIT_LOG_PATH`
## device_type
Catalyst 3650 runs IOS-XE. Use `cisco_xe` (default). If connection fails, try `cisco_ios`.

23
pyproject.toml Normal file
View File

@@ -0,0 +1,23 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "cisco-switch-mcp"
version = "1.0.0"
description = "Model Context Protocol server for Cisco Catalyst IOS-XE switch management"
readme = "README.md"
requires-python = ">=3.10"
license = "MIT"
dependencies = [
"mcp>=1.0.0",
"netmiko>=4.3.0",
"pydantic>=2.0.0",
"pydantic-settings>=2.0.0",
]
[project.scripts]
cisco-switch-mcp = "cisco_switch_mcp.server:main"
[tool.hatch.build.targets.wheel]
packages = ["src/cisco_switch_mcp"]

View File

@@ -0,0 +1,3 @@
"""Cisco switch MCP server for IOS-XE Catalyst switches."""
__version__ = "1.0.0"

View File

@@ -0,0 +1,4 @@
from cisco_switch_mcp.server import main
if __name__ == "__main__":
main()

Binary file not shown.

View File

@@ -0,0 +1,23 @@
"""Audit log for mutating operations."""
from __future__ import annotations
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from cisco_switch_mcp.config import settings
def audit_log(action: str, details: dict[str, Any], *, dry_run: bool = False) -> None:
entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"action": action,
"dry_run": dry_run,
**details,
}
path = Path(settings.audit_log_path)
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as fh:
fh.write(json.dumps(entry, default=str) + "\n")

View File

@@ -0,0 +1,131 @@
"""SSH client for Cisco IOS-XE switches via Netmiko."""
from __future__ import annotations
import json
from contextlib import contextmanager
from typing import Iterator
from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException
from cisco_switch_mcp.config import settings
from cisco_switch_mcp.guards import GuardError, require_credentials
class CiscoSwitchError(Exception):
def __init__(self, message: str, *, cause: Exception | None = None):
super().__init__(message)
self.cause = cause
def _device_params() -> dict:
require_credentials()
secret = settings.enable_password or settings.secret or settings.password
params = {
"device_type": settings.device_type,
"host": settings.host,
"port": settings.port,
"username": settings.username,
"password": settings.password,
"secret": secret,
"timeout": settings.timeout,
"conn_timeout": settings.timeout,
"fast_cli": False,
}
if settings.session_log_path:
params["session_log"] = settings.session_log_path
return params
@contextmanager
def connect() -> Iterator:
conn = None
try:
conn = ConnectHandler(**_device_params())
try:
conn.enable()
except Exception:
pass
yield conn
except NetmikoAuthenticationException as exc:
raise CiscoSwitchError("SSH authentication failed", cause=exc) from exc
except NetmikoTimeoutException as exc:
raise CiscoSwitchError(
f"SSH connection timed out to {settings.host}:{settings.port}", cause=exc
) from exc
except Exception as exc:
raise CiscoSwitchError(f"SSH connection failed: {exc}", cause=exc) from exc
finally:
if conn is not None:
try:
conn.disconnect()
except Exception:
pass
def run_command(command: str, *, use_textfsm: bool = False) -> str:
"""Execute a privileged exec-mode show command."""
cmd = command.strip()
if not cmd:
raise GuardError("command is empty")
with connect() as conn:
if use_textfsm:
try:
result = conn.send_command(cmd, use_textfsm=True)
if isinstance(result, list):
import json
return json.dumps(result, indent=2, default=str)
except Exception:
pass
output = conn.send_command(cmd, read_timeout=settings.timeout)
lines = output.splitlines()
if len(lines) > settings.max_output_lines:
trimmed = lines[: settings.max_output_lines]
trimmed.append(
f"... truncated ({len(lines) - settings.max_output_lines} more lines)"
)
return "\n".join(trimmed)
return output
def run_show_command(command: str) -> str:
"""Execute a show command with error wrapping.
Wraps run_command in a try/except and returns a JSON error
payload on failure instead of raising an exception, so that
MCP tools always return a string even when the switch is
unreachable or authentication fails.
"""
try:
return run_command(command)
except Exception as exc:
return json.dumps({"error": str(exc)}, indent=2)
def run_config_commands(commands: list[str]) -> str:
"""Apply configuration commands in config mode."""
from cisco_switch_mcp.guards import validate_config_commands
validate_config_commands(commands)
cleaned = [c.strip() for c in commands if c.strip()]
if not cleaned:
raise GuardError("no configuration commands provided")
with connect() as conn:
output = conn.send_config_set(cleaned)
return output
def test_connection() -> dict:
require_credentials()
with connect() as conn:
hostname = conn.find_prompt()
version = conn.send_command("show version | include Cisco IOS XE|Model Number|System serial|uptime")
return {
"status": "ok",
"host": settings.host,
"prompt": hostname,
"version_summary": version.strip(),
"device_type": settings.device_type,
}

View File

@@ -0,0 +1,27 @@
"""Configuration from environment variables."""
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="CISCO_SWITCH_", extra="ignore")
host: str = ""
port: int = 22
username: str = ""
password: str = ""
enable_password: str = ""
device_type: str = "cisco_xe"
secret: str = "" # alias: set CISCO_SWITCH_SECRET for enable if enable_password empty
timeout: float = 30.0
session_log_path: str = ""
allow_writes: bool = False
dry_run: bool = False
audit_log_path: str = "/root/.hermes/logs/cisco-switch-mcp-audit.jsonl"
block_config_replace: bool = True
max_output_lines: int = 500
settings = Settings()

View File

@@ -0,0 +1,55 @@
"""Safety guardrails."""
from __future__ import annotations
from cisco_switch_mcp.audit import audit_log
from cisco_switch_mcp.config import settings
class GuardError(Exception):
pass
def require_credentials() -> None:
if not settings.host or not settings.username or not settings.password:
raise GuardError(
"Set CISCO_SWITCH_HOST, CISCO_SWITCH_USERNAME, and CISCO_SWITCH_PASSWORD."
)
def require_writes(action: str, payload: dict | None = None) -> None:
require_credentials()
if settings.dry_run:
audit_log(action, {"payload": payload or {}}, dry_run=True)
return
if not settings.allow_writes:
raise GuardError(
f"Write operation '{action}' blocked. Set CISCO_SWITCH_ALLOW_WRITES=true."
)
def is_dry_run_response(action: str, payload: dict | None = None) -> dict:
audit_log(action, {"payload": payload or {}}, dry_run=True)
return {
"dry_run": True,
"action": action,
"payload": payload,
"message": "Dry run — no changes applied.",
}
def validate_config_commands(commands: list[str]) -> None:
if not settings.block_config_replace:
return
blocked = ("write erase", "reload", "format", "delete /force", "copy run start")
joined = " ".join(commands).lower()
for phrase in blocked:
if phrase in joined:
raise GuardError(f"Blocked dangerous command pattern: {phrase}")
def validate_interface(interface: str) -> str:
iface = interface.strip()
if not iface:
raise GuardError("interface name is required")
return iface

View File

@@ -0,0 +1,27 @@
"""Cisco Catalyst IOS-XE switch MCP server."""
from __future__ import annotations
from mcp.server.fastmcp import FastMCP
from cisco_switch_mcp import __version__
from cisco_switch_mcp.tools import register_all
mcp = FastMCP(
"cisco-switch",
instructions=(
"Cisco Catalyst IOS-XE switch management via SSH. "
"Use show_* tools for monitoring. Write tools require CISCO_SWITCH_ALLOW_WRITES=true. "
"Interface names use full form: GigabitEthernet1/0/1."
),
)
register_all(mcp)
def main() -> None:
mcp.run()
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,14 @@
"""Tool registration."""
from mcp.server.fastmcp import FastMCP
from cisco_switch_mcp.tools import diagnostics, interfaces, l2, system, vlan, write
def register_all(mcp: FastMCP) -> None:
system.register(mcp)
interfaces.register(mcp)
vlan.register(mcp)
l2.register(mcp)
diagnostics.register(mcp)
write.register(mcp)

View File

@@ -0,0 +1,75 @@
"""Diagnostics, logs, and config read tools."""
from __future__ import annotations
from mcp.server.fastmcp import FastMCP
from cisco_switch_mcp.client import run_show_command
def register(mcp: FastMCP) -> None:
@mcp.tool()
def cisco_switch_show_logging() -> str:
"""Get recent syslog messages."""
return run_show_command("show logging last 80")
@mcp.tool()
def cisco_switch_ping(
target: str,
count: int = 5,
source: str = "",
) -> str:
"""Ping from the switch (e.g. target=10.77.30.1)."""
target = target.strip()
if not target:
raise ValueError("target is required")
count = max(1, min(count, 10))
cmd = f"ping {target} repeat {count}"
if source.strip():
cmd = f"ping {target} source {source.strip()} repeat {count}"
return run_show_command(cmd)
@mcp.tool()
def cisco_switch_traceroute(target: str) -> str:
"""Traceroute from the switch."""
target = target.strip()
if not target:
raise ValueError("target is required")
return run_show_command(f"traceroute {target}")
@mcp.tool()
def cisco_switch_show_running_config(section: str = "") -> str:
"""Get running configuration. Optional section filter (e.g. 'interface GigabitEthernet1/0/1')."""
if section.strip():
return run_show_command(f"show running-config | section {section.strip()}")
return run_show_command("show running-config")
@mcp.tool()
def cisco_switch_show_startup_config() -> str:
"""Get startup configuration."""
return run_show_command("show startup-config")
@mcp.tool()
def cisco_switch_show_ip_route() -> str:
"""Get IP routing table on the switch."""
return run_show_command("show ip route")
@mcp.tool()
def cisco_switch_show_users() -> str:
"""Get logged-in users."""
return run_show_command("show users")
@mcp.tool()
def cisco_switch_show_power_inline() -> str:
"""Get PoE status (if PoE ports present)."""
return run_show_command("show power inline")
@mcp.tool()
def cisco_switch_show_interfaces_transceiver() -> str:
"""Get SFP DOM data: temperature, voltage, current, TX/RX power."""
return run_show_command("show interface transceiver")
@mcp.tool()
def cisco_switch_show_access_lists() -> str:
"""Get ACL configuration with hit counts."""
return run_show_command("show access-lists")

View File

@@ -0,0 +1,41 @@
"""Interface monitoring tools."""
from __future__ import annotations
from mcp.server.fastmcp import FastMCP
from cisco_switch_mcp.client import run_show_command
from cisco_switch_mcp.guards import validate_interface
def register(mcp: FastMCP) -> None:
@mcp.tool()
def cisco_switch_show_interfaces_status() -> str:
"""Get brief interface status (port, VLAN, speed, duplex, status)."""
return run_show_command("show interfaces status")
@mcp.tool()
def cisco_switch_show_interfaces_description() -> str:
"""Get interface descriptions."""
return run_show_command("show interfaces description")
@mcp.tool()
def cisco_switch_show_interface(interface: str) -> str:
"""Get detailed statistics for one interface (e.g. GigabitEthernet1/0/1)."""
iface = validate_interface(interface)
return run_show_command(f"show interfaces {iface}")
@mcp.tool()
def cisco_switch_show_interfaces_counters() -> str:
"""Get interface error and packet counters."""
return run_show_command("show interfaces counters errors")
@mcp.tool()
def cisco_switch_show_port_channel_summary() -> str:
"""Get EtherChannel / port-channel summary."""
return run_show_command("show etherchannel summary")
@mcp.tool()
def cisco_switch_show_ip_interface_brief() -> str:
"""Get L3 interface IP summary (SVIs and routed ports)."""
return run_show_command("show ip interface brief")

View File

@@ -0,0 +1,41 @@
"""Layer-2 discovery and MAC tools."""
from __future__ import annotations
from mcp.server.fastmcp import FastMCP
from cisco_switch_mcp.client import run_show_command
from cisco_switch_mcp.guards import validate_interface
def register(mcp: FastMCP) -> None:
@mcp.tool()
def cisco_switch_show_mac_address_table() -> str:
"""Get MAC address table (CAM table)."""
return run_show_command("show mac address-table")
@mcp.tool()
def cisco_switch_show_mac_address_table_vlan(vlan_id: int) -> str:
"""Get MAC address table filtered by VLAN."""
return run_show_command(f"show mac address-table vlan {vlan_id}")
@mcp.tool()
def cisco_switch_show_mac_address_table_interface(interface: str) -> str:
"""Get MAC addresses learned on one interface."""
iface = validate_interface(interface)
return run_show_command(f"show mac address-table interface {iface}")
@mcp.tool()
def cisco_switch_show_cdp_neighbors() -> str:
"""Get CDP neighbor discovery table."""
return run_show_command("show cdp neighbors detail")
@mcp.tool()
def cisco_switch_show_lldp_neighbors() -> str:
"""Get LLDP neighbor discovery table."""
return run_show_command("show lldp neighbors detail")
@mcp.tool()
def cisco_switch_show_arp() -> str:
"""Get ARP table from the switch."""
return run_show_command("show ip arp")

View File

@@ -0,0 +1,70 @@
"""System and hardware monitoring tools."""
from __future__ import annotations
import json
from mcp.server.fastmcp import FastMCP
from cisco_switch_mcp.client import CiscoSwitchError, run_show_command, test_connection
from cisco_switch_mcp.config import settings
def register(mcp: FastMCP) -> None:
@mcp.tool()
def cisco_switch_get_config() -> str:
"""Get MCP server configuration (no secrets)."""
return json.dumps(
{
"host": settings.host,
"port": settings.port,
"device_type": settings.device_type,
"username_configured": bool(settings.username),
"allow_writes": settings.allow_writes,
"dry_run": settings.dry_run,
},
indent=2,
)
@mcp.tool()
def cisco_switch_test_connection() -> str:
"""Test SSH connectivity and authentication to the switch."""
try:
return json.dumps(test_connection(), indent=2, default=str)
except CiscoSwitchError as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@mcp.tool()
def cisco_switch_show_version() -> str:
"""Get IOS-XE version, model, serial number, and uptime."""
return run_show_command("show version")
@mcp.tool()
def cisco_switch_show_inventory() -> str:
"""Get hardware inventory (chassis, power supplies, modules)."""
return run_show_command("show inventory")
@mcp.tool()
def cisco_switch_show_environment() -> str:
"""Get environmental status: temperature, fans, power supplies."""
return run_show_command("show environment all")
@mcp.tool()
def cisco_switch_show_processes_cpu() -> str:
"""Get CPU utilization summary."""
return run_show_command("show processes cpu sorted | exclude 0.00")
@mcp.tool()
def cisco_switch_show_memory() -> str:
"""Get memory utilization."""
return run_show_command("show memory statistics")
@mcp.tool()
def cisco_switch_show_clock() -> str:
"""Get switch system time."""
return run_show_command("show clock")
@mcp.tool()
def cisco_switch_show_redundancy() -> str:
"""Get stack/redundancy state (useful for Catalyst 3650 stacks)."""
return run_show_command("show redundancy")

View File

@@ -0,0 +1,34 @@
"""VLAN tools."""
from __future__ import annotations
from mcp.server.fastmcp import FastMCP
from cisco_switch_mcp.client import run_show_command
def register(mcp: FastMCP) -> None:
@mcp.tool()
def cisco_switch_show_vlan_brief() -> str:
"""List VLANs and port assignments (show vlan brief)."""
return run_show_command("show vlan brief")
@mcp.tool()
def cisco_switch_show_vlan(id: int = 0) -> str:
"""Get details for one VLAN ID, or all VLANs if id=0."""
if id > 0:
return run_show_command(f"show vlan id {id}")
return run_show_command("show vlan")
@mcp.tool()
def cisco_switch_show_spanning_tree() -> str:
"""Get spanning-tree overview."""
return run_show_command("show spanning-tree summary")
@mcp.tool()
def cisco_switch_show_spanning_tree_interface(interface: str) -> str:
"""Get spanning-tree details for one interface."""
from cisco_switch_mcp.guards import validate_interface
iface = validate_interface(interface)
return run_show_command(f"show spanning-tree interface {iface} detail")

View File

@@ -0,0 +1,89 @@
"""Controlled write/configuration tools."""
from __future__ import annotations
import json
from typing import Any
from mcp.server.fastmcp import FastMCP
from cisco_switch_mcp.audit import audit_log
from cisco_switch_mcp.client import run_config_commands
from cisco_switch_mcp.config import settings
from cisco_switch_mcp.guards import (
GuardError,
is_dry_run_response,
require_writes,
validate_interface,
)
def register(mcp: FastMCP) -> None:
@mcp.tool()
def cisco_switch_interface_shutdown(interface: str, shutdown: bool = True) -> str:
"""Administratively shut down or enable an interface."""
iface = validate_interface(interface)
action = "shutdown" if shutdown else "no shutdown"
commands = [f"interface {iface}", action]
payload = {"interface": iface, "shutdown": shutdown, "commands": commands}
if settings.dry_run:
return json.dumps(is_dry_run_response("interface_shutdown", payload), indent=2)
require_writes("interface_shutdown", payload)
result = run_config_commands(commands)
audit_log("interface_shutdown", payload | {"result": result})
return result
@mcp.tool()
def cisco_switch_set_access_vlan(interface: str, vlan_id: int) -> str:
"""Set switchport access VLAN on an access port."""
iface = validate_interface(interface)
if vlan_id < 1 or vlan_id > 4094:
raise GuardError("vlan_id must be 1-4094")
commands = [
f"interface {iface}",
"switchport mode access",
f"switchport access vlan {vlan_id}",
]
payload = {"interface": iface, "vlan_id": vlan_id, "commands": commands}
if settings.dry_run:
return json.dumps(is_dry_run_response("set_access_vlan", payload), indent=2)
require_writes("set_access_vlan", payload)
result = run_config_commands(commands)
audit_log("set_access_vlan", payload | {"result": result})
return result
@mcp.tool()
def cisco_switch_set_interface_description(interface: str, description: str) -> str:
"""Set interface description."""
iface = validate_interface(interface)
desc = description.strip()[:240]
commands = [f"interface {iface}", f"description {desc}"]
payload = {"interface": iface, "description": desc, "commands": commands}
if settings.dry_run:
return json.dumps(is_dry_run_response("set_description", payload), indent=2)
require_writes("set_description", payload)
result = run_config_commands(commands)
audit_log("set_description", payload | {"result": result})
return result
@mcp.tool()
def cisco_switch_run_config_commands(commands: list[str]) -> str:
"""Apply arbitrary configuration commands. Blocked patterns: reload, write erase, format."""
payload: dict[str, Any] = {"commands": commands}
if settings.dry_run:
return json.dumps(is_dry_run_response("run_config_commands", payload), indent=2)
require_writes("run_config_commands", payload)
result = run_config_commands(commands)
audit_log("run_config_commands", payload | {"result": result})
return result
@mcp.tool()
def cisco_switch_save_config() -> str:
"""Save running config to startup (copy running-config startup-config)."""
commands = ["end", "write memory"]
if settings.dry_run:
return json.dumps(is_dry_run_response("save_config", {}), indent=2)
require_writes("save_config", {})
result = run_config_commands(commands)
audit_log("save_config", {"result": result})
return result