Agentic-OS/mcp-servers/fortiswitch-mcp/server.py

"""FortiSwitch MCP server (standalone FortiSwitchOS REST API).

Read-only diagnostics. Supports token (API key) auth, with optional
username/password session login as a fallback. Writes require
FORTISWITCH_ALLOW_WRITES=true. Endpoints are best-effort and may vary by
FortiSwitchOS version.
"""
from __future__ import annotations

import json
import os

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("fortiswitch")

ALLOW_WRITES = os.environ.get("FORTISWITCH_ALLOW_WRITES", "false").lower() == "true"

_active: dict | None = None  # {"host","api_key","username","password","port"}


def _resolve(host, api_key, username, password, port) -> dict:
    if host:
        return {
            "host": host,
            "api_key": api_key,
            "username": username,
            "password": password,
            "port": port or 443,
        }
    if _active:
        return _active
    raise RuntimeError("No FortiSwitch target. Call fortiswitch_connect first.")


def _base(target: dict) -> str:
    return f"https://{target['host']}:{target.get('port', 443)}"


def _client(target: dict) -> httpx.Client:
    client = httpx.Client(verify=False, timeout=20)  # noqa: S501 (self-signed device)
    if not target.get("api_key") and target.get("username"):
        # Session login fallback
        try:
            client.post(
                f"{_base(target)}/logincheck",
                data={"username": target["username"], "secretkey": target.get("password", "")},
            )
        except Exception:  # noqa: BLE001
            pass
    return client


def _headers(target: dict) -> dict:
    if target.get("api_key"):
        return {"Authorization": f"Bearer {target['api_key']}"}
    return {}


def _get(target: dict, path: str, params: dict | None = None) -> dict:
    client = _client(target)
    try:
        r = client.get(f"{_base(target)}{path}", headers=_headers(target), params=params)
        r.raise_for_status()
        return r.json()
    finally:
        client.close()


@mcp.tool()
def fortiswitch_connect(host: str, api_key: str | None = None, username: str | None = None,
                        password: str | None = None, port: int = 443) -> str:
    """Set the active FortiSwitch target and verify connectivity."""
    global _active
    target = {"host": host, "api_key": api_key, "username": username, "password": password, "port": port}
    try:
        _get(target, "/api/v2/monitor/system/status")
    except Exception as exc:  # noqa: BLE001
        return f"connect failed: {exc}"
    _active = target
    return f"connected to FortiSwitch {host}"


@mcp.tool()
def fortiswitch_test_connection(host: str | None = None, api_key: str | None = None,
                                username: str | None = None, password: str | None = None,
                                port: int = 443) -> str:
    """Test connectivity to a FortiSwitch."""
    target = _resolve(host, api_key, username, password, port)
    try:
        _get(target, "/api/v2/monitor/system/status")
        return "ok"
    except Exception as exc:  # noqa: BLE001
        return f"failed: {exc}"


@mcp.tool()
def fortiswitch_get_system_status(host: str | None = None, api_key: str | None = None,
                                  username: str | None = None, password: str | None = None,
                                  port: int = 443) -> str:
    """System status: model, version, serial, uptime."""
    target = _resolve(host, api_key, username, password, port)
    return json.dumps(_get(target, "/api/v2/monitor/system/status"), indent=2)[:6000]


@mcp.tool()
def fortiswitch_list_ports(host: str | None = None, api_key: str | None = None,
                           username: str | None = None, password: str | None = None,
                           port: int = 443) -> str:
    """List switch ports and their status."""
    target = _resolve(host, api_key, username, password, port)
    for path in ("/api/v2/monitor/switch/port", "/api/v2/cmdb/switch/interface"):
        try:
            return json.dumps(_get(target, path), indent=2)[:8000]
        except Exception:  # noqa: BLE001
            continue
    return "could not read switch ports on known endpoints"


@mcp.tool()
def fortiswitch_get_port_stats(host: str | None = None, api_key: str | None = None,
                               username: str | None = None, password: str | None = None,
                               port: int = 443) -> str:
    """Per-port statistics / counters (errors, traffic)."""
    target = _resolve(host, api_key, username, password, port)
    for path in ("/api/v2/monitor/switch/port-statistics", "/api/v2/monitor/switch/port"):
        try:
            return json.dumps(_get(target, path), indent=2)[:8000]
        except Exception:  # noqa: BLE001
            continue
    return "could not read port statistics on known endpoints"


if __name__ == "__main__":
    mcp.run()