commit 68e87cf6a5b2239b7803c49b32f6d5e28c80fabe Author: root Date: Sat Jul 4 20:15:11 2026 +0000 Initial commit: Cisco Switch MCP with fixes C3, C4, C5, C6, S2 diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..ddf1f23 --- /dev/null +++ b/.env.example @@ -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 diff --git a/README.md b/README.md new file mode 100644 index 0000000..5d96167 --- /dev/null +++ b/README.md @@ -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`. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..60027cc --- /dev/null +++ b/pyproject.toml @@ -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"] diff --git a/src/cisco_switch_mcp/__init__.py b/src/cisco_switch_mcp/__init__.py new file mode 100644 index 0000000..af46bbb --- /dev/null +++ b/src/cisco_switch_mcp/__init__.py @@ -0,0 +1,3 @@ +"""Cisco switch MCP server for IOS-XE Catalyst switches.""" + +__version__ = "1.0.0" diff --git a/src/cisco_switch_mcp/__main__.py b/src/cisco_switch_mcp/__main__.py new file mode 100644 index 0000000..b3a5a95 --- /dev/null +++ b/src/cisco_switch_mcp/__main__.py @@ -0,0 +1,4 @@ +from cisco_switch_mcp.server import main + +if __name__ == "__main__": + main() diff --git a/src/cisco_switch_mcp/__pycache__/__init__.cpython-311.pyc b/src/cisco_switch_mcp/__pycache__/__init__.cpython-311.pyc new file mode 100644 index 0000000..2925b8f Binary files /dev/null and b/src/cisco_switch_mcp/__pycache__/__init__.cpython-311.pyc differ diff --git a/src/cisco_switch_mcp/__pycache__/audit.cpython-311.pyc b/src/cisco_switch_mcp/__pycache__/audit.cpython-311.pyc new file mode 100644 index 0000000..120f40c Binary files /dev/null and b/src/cisco_switch_mcp/__pycache__/audit.cpython-311.pyc differ diff --git a/src/cisco_switch_mcp/__pycache__/client.cpython-311.pyc b/src/cisco_switch_mcp/__pycache__/client.cpython-311.pyc new file mode 100644 index 0000000..91a7074 Binary files /dev/null and b/src/cisco_switch_mcp/__pycache__/client.cpython-311.pyc differ diff --git a/src/cisco_switch_mcp/__pycache__/config.cpython-311.pyc b/src/cisco_switch_mcp/__pycache__/config.cpython-311.pyc new file mode 100644 index 0000000..9c75528 Binary files /dev/null and b/src/cisco_switch_mcp/__pycache__/config.cpython-311.pyc differ diff --git a/src/cisco_switch_mcp/__pycache__/guards.cpython-311.pyc b/src/cisco_switch_mcp/__pycache__/guards.cpython-311.pyc new file mode 100644 index 0000000..f1197a5 Binary files /dev/null and b/src/cisco_switch_mcp/__pycache__/guards.cpython-311.pyc differ diff --git a/src/cisco_switch_mcp/__pycache__/server.cpython-311.pyc b/src/cisco_switch_mcp/__pycache__/server.cpython-311.pyc new file mode 100644 index 0000000..4231bee Binary files /dev/null and b/src/cisco_switch_mcp/__pycache__/server.cpython-311.pyc differ diff --git a/src/cisco_switch_mcp/audit.py b/src/cisco_switch_mcp/audit.py new file mode 100644 index 0000000..f57daa6 --- /dev/null +++ b/src/cisco_switch_mcp/audit.py @@ -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") diff --git a/src/cisco_switch_mcp/client.py b/src/cisco_switch_mcp/client.py new file mode 100644 index 0000000..f9ad425 --- /dev/null +++ b/src/cisco_switch_mcp/client.py @@ -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, + } diff --git a/src/cisco_switch_mcp/config.py b/src/cisco_switch_mcp/config.py new file mode 100644 index 0000000..e579e88 --- /dev/null +++ b/src/cisco_switch_mcp/config.py @@ -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() diff --git a/src/cisco_switch_mcp/guards.py b/src/cisco_switch_mcp/guards.py new file mode 100644 index 0000000..fd88520 --- /dev/null +++ b/src/cisco_switch_mcp/guards.py @@ -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 diff --git a/src/cisco_switch_mcp/server.py b/src/cisco_switch_mcp/server.py new file mode 100644 index 0000000..38d0bac --- /dev/null +++ b/src/cisco_switch_mcp/server.py @@ -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() diff --git a/src/cisco_switch_mcp/tools/__init__.py b/src/cisco_switch_mcp/tools/__init__.py new file mode 100644 index 0000000..70decf7 --- /dev/null +++ b/src/cisco_switch_mcp/tools/__init__.py @@ -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) diff --git a/src/cisco_switch_mcp/tools/__pycache__/__init__.cpython-311.pyc b/src/cisco_switch_mcp/tools/__pycache__/__init__.cpython-311.pyc new file mode 100644 index 0000000..e22c05f Binary files /dev/null and b/src/cisco_switch_mcp/tools/__pycache__/__init__.cpython-311.pyc differ diff --git a/src/cisco_switch_mcp/tools/__pycache__/diagnostics.cpython-311.pyc b/src/cisco_switch_mcp/tools/__pycache__/diagnostics.cpython-311.pyc new file mode 100644 index 0000000..42b3a27 Binary files /dev/null and b/src/cisco_switch_mcp/tools/__pycache__/diagnostics.cpython-311.pyc differ diff --git a/src/cisco_switch_mcp/tools/__pycache__/interfaces.cpython-311.pyc b/src/cisco_switch_mcp/tools/__pycache__/interfaces.cpython-311.pyc new file mode 100644 index 0000000..440e225 Binary files /dev/null and b/src/cisco_switch_mcp/tools/__pycache__/interfaces.cpython-311.pyc differ diff --git a/src/cisco_switch_mcp/tools/__pycache__/l2.cpython-311.pyc b/src/cisco_switch_mcp/tools/__pycache__/l2.cpython-311.pyc new file mode 100644 index 0000000..f79759a Binary files /dev/null and b/src/cisco_switch_mcp/tools/__pycache__/l2.cpython-311.pyc differ diff --git a/src/cisco_switch_mcp/tools/__pycache__/system.cpython-311.pyc b/src/cisco_switch_mcp/tools/__pycache__/system.cpython-311.pyc new file mode 100644 index 0000000..b0f66bf Binary files /dev/null and b/src/cisco_switch_mcp/tools/__pycache__/system.cpython-311.pyc differ diff --git a/src/cisco_switch_mcp/tools/__pycache__/vlan.cpython-311.pyc b/src/cisco_switch_mcp/tools/__pycache__/vlan.cpython-311.pyc new file mode 100644 index 0000000..42ea213 Binary files /dev/null and b/src/cisco_switch_mcp/tools/__pycache__/vlan.cpython-311.pyc differ diff --git a/src/cisco_switch_mcp/tools/__pycache__/write.cpython-311.pyc b/src/cisco_switch_mcp/tools/__pycache__/write.cpython-311.pyc new file mode 100644 index 0000000..d3d497f Binary files /dev/null and b/src/cisco_switch_mcp/tools/__pycache__/write.cpython-311.pyc differ diff --git a/src/cisco_switch_mcp/tools/diagnostics.py b/src/cisco_switch_mcp/tools/diagnostics.py new file mode 100644 index 0000000..9df6175 --- /dev/null +++ b/src/cisco_switch_mcp/tools/diagnostics.py @@ -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") diff --git a/src/cisco_switch_mcp/tools/interfaces.py b/src/cisco_switch_mcp/tools/interfaces.py new file mode 100644 index 0000000..c5e8158 --- /dev/null +++ b/src/cisco_switch_mcp/tools/interfaces.py @@ -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") diff --git a/src/cisco_switch_mcp/tools/l2.py b/src/cisco_switch_mcp/tools/l2.py new file mode 100644 index 0000000..43cf0c4 --- /dev/null +++ b/src/cisco_switch_mcp/tools/l2.py @@ -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") diff --git a/src/cisco_switch_mcp/tools/system.py b/src/cisco_switch_mcp/tools/system.py new file mode 100644 index 0000000..6a53ca1 --- /dev/null +++ b/src/cisco_switch_mcp/tools/system.py @@ -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") diff --git a/src/cisco_switch_mcp/tools/vlan.py b/src/cisco_switch_mcp/tools/vlan.py new file mode 100644 index 0000000..9ed0073 --- /dev/null +++ b/src/cisco_switch_mcp/tools/vlan.py @@ -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") diff --git a/src/cisco_switch_mcp/tools/write.py b/src/cisco_switch_mcp/tools/write.py new file mode 100644 index 0000000..c3c567d --- /dev/null +++ b/src/cisco_switch_mcp/tools/write.py @@ -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