get_device_activity
Retrieve connected clients and traffic data for a UniFi network device using its MAC address to monitor network activity.
Instructions
Get activity for a specific device including connected clients and their traffic
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| mac | Yes | MAC address of the device (AP or switch) |
Implementation Reference
- src/unifi_mcp/unifi_client.py:350-387 (handler)Core handler function that retrieves device information, finds connected clients, calculates total traffic, and returns structured activity data.async def get_device_activity(self, device_mac: str) -> dict[str, Any]: """Get activity summary for a specific device. This includes the device info, connected clients, and their traffic. Args: device_mac: MAC address of the device. Returns: Dictionary with device info and connected clients. """ # Get device info device = await self.get_device(device_mac) if not device: # Try getting from all devices devices = await self.get_devices() device_mac_lower = device_mac.lower().replace(":", "").replace("-", "") for d in devices: d_mac = d.get("mac", "").lower().replace(":", "") if d_mac == device_mac_lower: device = d break # Get clients connected to this device clients = await self.get_device_clients(device_mac) # Calculate totals total_tx = sum(c.get("tx_bytes", 0) for c in clients) total_rx = sum(c.get("rx_bytes", 0) for c in clients) return { "device": device, "clients": clients, "client_count": len(clients), "total_tx_bytes": total_tx, "total_rx_bytes": total_rx, }
- src/unifi_mcp/server.py:131-144 (registration)Registers the 'get_device_activity' tool with the MCP server, including description and input schema requiring a 'mac' parameter.Tool( name="get_device_activity", description="Get activity for a specific device including connected clients and their traffic", inputSchema={ "type": "object", "properties": { "mac": { "type": "string", "description": "MAC address of the device (AP or switch)", } }, "required": ["mac"], }, ),
- src/unifi_mcp/server.py:230-235 (handler)MCP server tool dispatch handler that extracts the MAC from arguments, calls the client implementation, and returns formatted text content.case "get_device_activity": mac = arguments.get("mac", "") activity = await client.get_device_activity(mac) return [ TextContent(type="text", text=format_device_activity(activity)) ]
- src/unifi_mcp/server.py:393-460 (helper)Helper function to format the raw device activity data into a readable text report for the MCP response.def format_device_activity(activity: dict[str, Any]) -> str: """Format device activity for display.""" lines = [] device = activity.get("device") clients = activity.get("clients", []) client_count = activity.get("client_count", 0) total_tx = activity.get("total_tx_bytes", 0) total_rx = activity.get("total_rx_bytes", 0) # Device info if device: name = device.get("name", "Unknown") mac = device.get("mac", "Unknown") model = device.get("model", "Unknown") device_type = device.get("type", "Unknown") state = device.get("state", 0) state_str = "Online" if state == 1 else "Offline" lines.append(f"Device: {name}") lines.append(f" MAC: {mac}") lines.append(f" Model: {model} ({device_type})") lines.append(f" Status: {state_str}") lines.append("") else: lines.append("Device: Not found") lines.append("") # Summary lines.append(f"Connected Clients: {client_count}") lines.append( f"Total Traffic: TX {format_bytes(total_tx)} / RX {format_bytes(total_rx)}" ) lines.append("") # Client details if clients: lines.append("Client Activity:") for c in clients: hostname = c.get("hostname") or c.get("name") or "Unknown" client_mac = c.get("mac", "Unknown") ip = c.get("ip", "N/A") is_wired = c.get("is_wired", False) conn_type = "Wired" if is_wired else "Wireless" essid = c.get("essid", "") tx_bytes = c.get("tx_bytes", 0) rx_bytes = c.get("rx_bytes", 0) signal = c.get("signal", None) uptime = c.get("uptime", 0) lines.append(f" - {hostname}") lines.append(f" MAC: {client_mac}") lines.append(f" IP: {ip}") lines.append(f" Connection: {conn_type}") if essid: lines.append(f" SSID: {essid}") if signal is not None: lines.append(f" Signal: {signal} dBm") if uptime > 0: lines.append(f" Uptime: {format_uptime(uptime)}") lines.append( f" Traffic: TX {format_bytes(tx_bytes)} / RX {format_bytes(rx_bytes)}" ) lines.append("") else: lines.append("No clients currently connected to this device.") return "\n".join(lines)
- Supporting helper that filters currently connected clients by device MAC (checking ap_mac or sw_mac).async def get_device_clients(self, device_mac: str) -> list[dict[str, Any]]: """Get clients connected to a specific device (AP or switch). Args: device_mac: MAC address of the device. Returns: List of client dictionaries connected to this device. """ all_clients = await self.get_clients() device_mac_lower = device_mac.lower().replace(":", "").replace("-", "") connected_clients = [] for client in all_clients: # Check if client is connected to this AP (wireless) ap_mac = client.get("ap_mac", "").lower().replace(":", "") # Check if client is connected to this switch (wired) sw_mac = client.get("sw_mac", "").lower().replace(":", "") if device_mac_lower == ap_mac or device_mac_lower == sw_mac: connected_clients.append(client) return connected_clients