"""Site management tools for UniFi Network."""
from typing import Any
from mcp.server.fastmcp import Context
from unifi_mcp.clients.base import AppContext
from unifi_mcp.clients.network import UniFiNetworkClient
def _get_client(ctx: Context) -> UniFiNetworkClient:
"""Get the UniFi Network client from context."""
app_ctx: AppContext = ctx.request_context.lifespan_context
return UniFiNetworkClient(app_ctx)
async def list_sites(ctx: Context) -> list[dict[str, Any]]:
"""List all UniFi sites accessible to the current user.
Returns:
List of sites with name, description, and role information.
"""
client = _get_client(ctx)
sites = await client.get_sites()
result = []
for site in sites:
result.append({
"name": site.get("name", ""),
"desc": site.get("desc", ""),
"role": site.get("role", ""),
"role_hotspot": site.get("role_hotspot", False),
"attr_hidden_id": site.get("attr_hidden_id", ""),
"attr_no_delete": site.get("attr_no_delete", False),
})
return result
async def get_site_health(ctx: Context, site: str = "default") -> dict[str, Any]:
"""Get comprehensive health status for a site.
Args:
ctx: MCP context
site: Site name
Returns:
Health status broken down by subsystem (WAN, LAN, WLAN)
including connectivity, speed, and issues.
"""
client = _get_client(ctx)
health_data = await client.get_site_health(site)
# Organize by subsystem
health = {}
issues = []
for subsystem in health_data:
name = subsystem.get("subsystem", "unknown")
status = subsystem.get("status", "unknown")
health[name] = {
"status": status,
"num_adopted": subsystem.get("num_adopted"),
"num_pending": subsystem.get("num_pending"),
"num_disabled": subsystem.get("num_disabled"),
"num_disconnected": subsystem.get("num_disconnected"),
"num_sta": subsystem.get("num_sta"),
"num_user": subsystem.get("num_user"),
"num_guest": subsystem.get("num_guest"),
}
# WAN specific
if name == "wan":
health[name].update({
"wan_ip": subsystem.get("wan_ip"),
"gateways": subsystem.get("gateways", []),
"nameservers": subsystem.get("nameservers", []),
"isp_name": subsystem.get("isp_name"),
"isp_organization": subsystem.get("isp_organization"),
"tx_bytes-r": subsystem.get("tx_bytes-r"),
"rx_bytes-r": subsystem.get("rx_bytes-r"),
"speedtest_lastrun": subsystem.get("speedtest_lastrun"),
"speedtest_status": subsystem.get("speedtest_status"),
"xput_up": subsystem.get("xput_up"),
"xput_down": subsystem.get("xput_down"),
"latency": subsystem.get("latency"),
})
# Collect issues
if status != "ok":
issues.append({
"subsystem": name,
"status": status,
})
return {
"site": site,
"overall_status": "healthy" if not issues else "issues_detected",
"subsystems": health,
"issues": issues,
}
async def get_site_settings(ctx: Context, site: str = "default") -> dict[str, Any]:
"""Get site configuration settings.
Args:
ctx: MCP context
site: Site name
Returns:
Site settings organized by category.
"""
client = _get_client(ctx)
settings = await client.get_site_settings(site)
# Organize settings by key
organized = {}
for setting in settings:
key = setting.get("key", "unknown")
organized[key] = setting
return {
"site": site,
"settings": organized,
}
async def get_sysinfo(ctx: Context, site: str = "default") -> dict[str, Any]:
"""Get system information for the site controller.
Args:
ctx: MCP context
site: Site name
Returns:
System information including version, uptime, and build info.
"""
client = _get_client(ctx)
sysinfo = await client.get_sysinfo(site)
return {
"site": site,
"version": sysinfo.get("version", ""),
"build": sysinfo.get("build", ""),
"timezone": sysinfo.get("timezone", ""),
"hostname": sysinfo.get("hostname", ""),
"name": sysinfo.get("name", ""),
"uptime": sysinfo.get("uptime", 0),
"autobackup": sysinfo.get("autobackup", False),
"ip_addrs": sysinfo.get("ip_addrs", []),
"update_available": sysinfo.get("update_available", False),
"update_downloaded": sysinfo.get("update_downloaded", False),
}
async def get_networks(ctx: Context, site: str = "default") -> list[dict[str, Any]]:
"""Get all network/VLAN configurations.
Args:
ctx: MCP context
site: Site name
Returns:
List of network configurations including VLANs, IP ranges,
and DHCP settings.
"""
client = _get_client(ctx)
networks = await client.get_networks(site)
result = []
for network in networks:
result.append({
"name": network.get("name", ""),
"purpose": network.get("purpose", ""),
"vlan": network.get("vlan"),
"vlan_enabled": network.get("vlan_enabled", False),
"subnet": network.get("ip_subnet", ""),
"dhcp_enabled": network.get("dhcp_enabled", False),
"dhcp_start": network.get("dhcp_start"),
"dhcp_stop": network.get("dhcp_stop"),
"dhcp_lease_time": network.get("dhcp_lease_time"),
"domain_name": network.get("domain_name"),
"igmp_snooping": network.get("igmp_snooping", False),
"enabled": network.get("enabled", True),
})
return result
async def get_wlans(ctx: Context, site: str = "default") -> list[dict[str, Any]]:
"""Get all wireless network (SSID) configurations.
Args:
ctx: MCP context
site: Site name
Returns:
List of WLAN configurations including SSIDs, security settings,
and associated networks.
"""
client = _get_client(ctx)
wlans = await client.get_wlans(site)
result = []
for wlan in wlans:
result.append({
"name": wlan.get("name", ""),
"ssid": wlan.get("essid", wlan.get("name", "")),
"enabled": wlan.get("enabled", True),
"is_guest": wlan.get("is_guest", False),
"security": wlan.get("security", ""),
"wpa_mode": wlan.get("wpa_mode", ""),
"wpa_enc": wlan.get("wpa_enc", ""),
"network_id": wlan.get("networkconf_id"),
"vlan": wlan.get("vlan"),
"hide_ssid": wlan.get("hide_ssid", False),
"mac_filter_enabled": wlan.get("mac_filter_enabled", False),
"mac_filter_policy": wlan.get("mac_filter_policy"),
"schedule_enabled": wlan.get("schedule_enabled", False),
"band_steering": wlan.get("band_steering", "off"),
})
return result
async def get_port_profiles(ctx: Context, site: str = "default") -> list[dict[str, Any]]:
"""Get switch port profile configurations.
Args:
ctx: MCP context
site: Site name
Returns:
List of port profiles with VLAN and PoE settings.
"""
client = _get_client(ctx)
profiles = await client.get_port_profiles(site)
result = []
for profile in profiles:
result.append({
"name": profile.get("name", ""),
"native_networkconf_id": profile.get("native_networkconf_id"),
"forward": profile.get("forward", ""),
"poe_mode": profile.get("poe_mode", "auto"),
"stormctrl_enabled": profile.get("stormctrl_enabled", False),
"stp_port_mode": profile.get("stp_port_mode", True),
"lldpmed_enabled": profile.get("lldpmed_enabled", True),
"tagged_vlan_mgmt": profile.get("tagged_vlan_mgmt"),
})
return result
async def get_firewall_rules(ctx: Context, site: str = "default") -> list[dict[str, Any]]:
"""Get firewall rule configurations.
Args:
ctx: MCP context
site: Site name
Returns:
List of firewall rules with action, protocol, and port details.
"""
client = _get_client(ctx)
rules = await client.get_firewall_rules(site)
result = []
for rule in rules:
result.append({
"name": rule.get("name", ""),
"enabled": rule.get("enabled", True),
"ruleset": rule.get("ruleset", ""),
"rule_index": rule.get("rule_index"),
"action": rule.get("action", ""),
"protocol": rule.get("protocol", "all"),
"src_firewallgroup_ids": rule.get("src_firewallgroup_ids", []),
"dst_firewallgroup_ids": rule.get("dst_firewallgroup_ids", []),
"dst_port": rule.get("dst_port"),
"logging": rule.get("logging", False),
})
return result
async def get_routing_table(ctx: Context, site: str = "default") -> list[dict[str, Any]]:
"""Get the current routing table.
Args:
ctx: MCP context
site: Site name
Returns:
List of routes with destination, gateway, and interface.
"""
client = _get_client(ctx)
routes = await client.get_routing(site)
result = []
for route in routes:
result.append({
"destination": route.get("pfx", ""),
"gateway": route.get("nh", []),
"type": route.get("type", ""),
"interface": route.get("intf", ""),
"metric": route.get("metric"),
"static": route.get("static", False),
})
return result
