Ben
487f5355a0
All checks were successful
Build and Push Docker Image / build (push) Successful in 15s
- Use /api/auth/login for UniFi OS controllers (UDM, Cloud Gateway) - Use /api/login for standalone controllers - Refactor to use FastMCP's native mcp.run() with custom_route for /health - Switch to network_mode: host for local network access - Include README.md in Dockerfile for hatchling build
403 lines
12 KiB
Python
403 lines
12 KiB
Python
"""
|
|
UniFi MCP Light Server
|
|
|
|
A Hybrid MCP Light server for UniFi Network Controller.
|
|
Provides minimal specific tools with a raw API pass-through escape hatch.
|
|
|
|
Following the BLUEPRINT.md pattern:
|
|
- 4 specific tools for common operations
|
|
- Meta tools (tool_index, api_call) for flexibility
|
|
- Embedded API documentation resource
|
|
- Starlette wrapper with health endpoint
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
from typing import Optional
|
|
|
|
from dotenv import load_dotenv
|
|
from fastmcp import FastMCP
|
|
from starlette.responses import JSONResponse
|
|
|
|
from api_docs import get_api_docs
|
|
from unifi_client import UnifiClient, UnifiClientError, UnifiWriteBlockedError
|
|
|
|
# Load environment variables
|
|
load_dotenv()
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
|
)
|
|
logger = logging.getLogger("unifi-mcp-light")
|
|
|
|
# Configuration from environment
|
|
CONFIG = {
|
|
"host": os.getenv("UNIFI_HOST", ""),
|
|
"username": os.getenv("UNIFI_USERNAME", ""),
|
|
"password": os.getenv("UNIFI_PASSWORD", ""),
|
|
"port": int(os.getenv("UNIFI_PORT", "443")),
|
|
"site": os.getenv("UNIFI_SITE", "default"),
|
|
"verify_ssl": os.getenv("UNIFI_VERIFY_SSL", "false").lower() == "true",
|
|
"allow_writes": os.getenv("UNIFI_ALLOW_WRITES", "false").lower() == "true",
|
|
}
|
|
|
|
# Global client instance
|
|
_client: Optional[UnifiClient] = None
|
|
|
|
|
|
async def get_client() -> UnifiClient:
|
|
"""Get or create the UniFi client singleton."""
|
|
global _client
|
|
if _client is None:
|
|
if not CONFIG["host"]:
|
|
raise UnifiClientError("UNIFI_HOST environment variable not set")
|
|
if not CONFIG["username"]:
|
|
raise UnifiClientError("UNIFI_USERNAME environment variable not set")
|
|
if not CONFIG["password"]:
|
|
raise UnifiClientError("UNIFI_PASSWORD environment variable not set")
|
|
|
|
_client = UnifiClient(
|
|
host=CONFIG["host"],
|
|
username=CONFIG["username"],
|
|
password=CONFIG["password"],
|
|
port=CONFIG["port"],
|
|
site=CONFIG["site"],
|
|
verify_ssl=CONFIG["verify_ssl"],
|
|
allow_writes=CONFIG["allow_writes"],
|
|
)
|
|
await _client.connect()
|
|
return _client
|
|
|
|
|
|
async def close_client() -> None:
|
|
"""Close the UniFi client connection."""
|
|
global _client
|
|
if _client:
|
|
await _client.close()
|
|
_client = None
|
|
|
|
|
|
# Create MCP server
|
|
mcp = FastMCP(
|
|
name="unifi-mcp-light",
|
|
instructions="""
|
|
UniFi Network Controller MCP Server (Light Edition)
|
|
|
|
This server provides access to your UniFi Network Controller with:
|
|
- 4 specific tools for common read operations
|
|
- Raw API pass-through for any endpoint
|
|
- API documentation resource for reference
|
|
|
|
Use 'unifi://api-reference' resource for API documentation.
|
|
Write operations require UNIFI_ALLOW_WRITES=true.
|
|
""",
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# MCP Resources
|
|
# =============================================================================
|
|
|
|
|
|
@mcp.resource("unifi://api-reference")
|
|
def api_reference() -> str:
|
|
"""
|
|
UniFi API Reference Documentation.
|
|
|
|
Returns curated documentation for UniFi Network Controller API endpoints.
|
|
Use this to understand how to construct raw API calls with unifi_api_call.
|
|
"""
|
|
return get_api_docs()
|
|
|
|
|
|
# =============================================================================
|
|
# Specific Tools (4 total - don't exceed BLUEPRINT limit)
|
|
# =============================================================================
|
|
|
|
|
|
@mcp.tool()
|
|
async def unifi_list_clients() -> str:
|
|
"""
|
|
List all currently connected network clients.
|
|
|
|
Returns information about all devices (phones, laptops, IoT devices, etc.)
|
|
currently connected to your UniFi network, including:
|
|
- MAC and IP addresses
|
|
- Hostname and friendly name
|
|
- Connection type (wired/wireless)
|
|
- Traffic statistics
|
|
- Signal strength (wireless clients)
|
|
- Connected SSID (wireless clients)
|
|
|
|
Returns:
|
|
JSON array of client objects
|
|
"""
|
|
try:
|
|
client = await get_client()
|
|
result = await client.get_clients()
|
|
return json.dumps(result, indent=2)
|
|
except UnifiClientError as e:
|
|
return json.dumps({"error": str(e)})
|
|
|
|
|
|
@mcp.tool()
|
|
async def unifi_list_devices() -> str:
|
|
"""
|
|
List all UniFi network devices.
|
|
|
|
Returns information about all adopted UniFi devices including:
|
|
- Access Points (UAP)
|
|
- Switches (USW)
|
|
- Gateways/Routers (UGW, UDM)
|
|
- Other UniFi devices
|
|
|
|
Each device includes:
|
|
- Model and firmware version
|
|
- IP address and status
|
|
- Uptime and connected clients
|
|
- Port information (switches)
|
|
|
|
Returns:
|
|
JSON array of device objects
|
|
"""
|
|
try:
|
|
client = await get_client()
|
|
result = await client.get_devices()
|
|
return json.dumps(result, indent=2)
|
|
except UnifiClientError as e:
|
|
return json.dumps({"error": str(e)})
|
|
|
|
|
|
@mcp.tool()
|
|
async def unifi_get_system_info() -> str:
|
|
"""
|
|
Get UniFi controller system information.
|
|
|
|
Returns controller details including:
|
|
- Controller version and build
|
|
- Hostname and site name
|
|
- Timezone configuration
|
|
|
|
Returns:
|
|
JSON array with system info object
|
|
"""
|
|
try:
|
|
client = await get_client()
|
|
result = await client.get_system_info()
|
|
return json.dumps(result, indent=2)
|
|
except UnifiClientError as e:
|
|
return json.dumps({"error": str(e)})
|
|
|
|
|
|
@mcp.tool()
|
|
async def unifi_get_network_health() -> str:
|
|
"""
|
|
Get overall network health status.
|
|
|
|
Returns health metrics for each network subsystem:
|
|
- WAN: Internet connectivity status
|
|
- LAN: Local network status
|
|
- WLAN: Wireless network status
|
|
- VPN: VPN tunnel status
|
|
|
|
Each subsystem includes:
|
|
- Status (ok, warning, error)
|
|
- Connected user/guest counts
|
|
- Transfer rates
|
|
|
|
Returns:
|
|
JSON array of health status objects per subsystem
|
|
"""
|
|
try:
|
|
client = await get_client()
|
|
result = await client.get_health()
|
|
return json.dumps(result, indent=2)
|
|
except UnifiClientError as e:
|
|
return json.dumps({"error": str(e)})
|
|
|
|
|
|
# =============================================================================
|
|
# Meta Tools (don't count toward BLUEPRINT limit)
|
|
# =============================================================================
|
|
|
|
|
|
@mcp.tool()
|
|
async def unifi_tool_index() -> str:
|
|
"""
|
|
Get the complete index of available UniFi tools.
|
|
|
|
Returns a machine-readable catalog of all registered tools with their:
|
|
- Name and description
|
|
- Input parameter schemas
|
|
- Return type information
|
|
|
|
Use this for programmatic tool discovery and dynamic invocation.
|
|
|
|
Returns:
|
|
JSON object with tools array containing tool schemas
|
|
"""
|
|
tools = []
|
|
|
|
# Get all registered tools from MCP
|
|
tool_definitions = [
|
|
{
|
|
"name": "unifi_list_clients",
|
|
"description": "List all currently connected network clients",
|
|
"parameters": {},
|
|
},
|
|
{
|
|
"name": "unifi_list_devices",
|
|
"description": "List all UniFi network devices (APs, switches, gateways)",
|
|
"parameters": {},
|
|
},
|
|
{
|
|
"name": "unifi_get_system_info",
|
|
"description": "Get UniFi controller system information",
|
|
"parameters": {},
|
|
},
|
|
{
|
|
"name": "unifi_get_network_health",
|
|
"description": "Get overall network health status",
|
|
"parameters": {},
|
|
},
|
|
{
|
|
"name": "unifi_api_call",
|
|
"description": "Execute raw UniFi API call (escape hatch)",
|
|
"parameters": {
|
|
"endpoint": {
|
|
"type": "string",
|
|
"description": "API path (e.g., /api/s/default/stat/sta)",
|
|
"required": True,
|
|
},
|
|
"method": {
|
|
"type": "string",
|
|
"description": "HTTP method (GET, POST, PUT, DELETE)",
|
|
"default": "GET",
|
|
},
|
|
"params": {
|
|
"type": "string",
|
|
"description": "JSON string of request body/parameters",
|
|
"default": "{}",
|
|
},
|
|
},
|
|
},
|
|
]
|
|
|
|
return json.dumps({"tools": tool_definitions}, indent=2)
|
|
|
|
|
|
@mcp.tool()
|
|
async def unifi_api_call(
|
|
endpoint: str,
|
|
method: str = "GET",
|
|
params: str = "{}",
|
|
) -> str:
|
|
"""
|
|
Execute a raw UniFi API call (escape hatch).
|
|
|
|
This tool provides direct access to any UniFi API endpoint.
|
|
Use the 'unifi://api-reference' resource for API documentation.
|
|
|
|
IMPORTANT: Write operations (POST, PUT, DELETE, PATCH) require
|
|
UNIFI_ALLOW_WRITES=true environment variable.
|
|
|
|
Args:
|
|
endpoint: API path (e.g., '/api/s/default/stat/sta')
|
|
The server automatically handles UniFi OS path prefixing.
|
|
method: HTTP method - GET, POST, PUT, DELETE, PATCH (default: GET)
|
|
params: JSON string of request body/parameters (default: "{}")
|
|
|
|
Returns:
|
|
JSON response from the UniFi API, or error object
|
|
|
|
Examples:
|
|
# List all events from last 24 hours
|
|
endpoint: "/api/s/default/stat/event"
|
|
method: "GET"
|
|
params: "{}"
|
|
|
|
# Get specific client by MAC
|
|
endpoint: "/api/s/default/stat/user/aa:bb:cc:dd:ee:ff"
|
|
method: "GET"
|
|
params: "{}"
|
|
|
|
# Block a client (requires UNIFI_ALLOW_WRITES=true)
|
|
endpoint: "/api/s/default/cmd/stamgr"
|
|
method: "POST"
|
|
params: '{"cmd": "block-sta", "mac": "aa:bb:cc:dd:ee:ff"}'
|
|
"""
|
|
try:
|
|
# Parse params JSON
|
|
try:
|
|
payload = json.loads(params) if params and params != "{}" else None
|
|
except json.JSONDecodeError as e:
|
|
return json.dumps({"error": f"Invalid params JSON: {e}"})
|
|
|
|
client = await get_client()
|
|
result = await client.request(method, endpoint, payload)
|
|
return json.dumps(result, indent=2)
|
|
|
|
except UnifiWriteBlockedError as e:
|
|
return json.dumps(
|
|
{
|
|
"error": str(e),
|
|
"hint": "Set UNIFI_ALLOW_WRITES=true environment variable to enable write operations",
|
|
}
|
|
)
|
|
except UnifiClientError as e:
|
|
return json.dumps({"error": str(e)})
|
|
|
|
|
|
# =============================================================================
|
|
# Health Check Endpoint (using FastMCP custom_route)
|
|
# =============================================================================
|
|
|
|
from starlette.requests import Request
|
|
|
|
|
|
@mcp.custom_route("/health", methods=["GET"])
|
|
async def health_endpoint(request: Request) -> JSONResponse:
|
|
"""
|
|
Health check endpoint for Docker/Kubernetes.
|
|
|
|
Returns controller connectivity status and configuration.
|
|
"""
|
|
try:
|
|
client = await get_client()
|
|
await client.ping()
|
|
return JSONResponse(
|
|
{
|
|
"status": "ok",
|
|
"controller": CONFIG["host"],
|
|
"site": CONFIG["site"],
|
|
"write_enabled": CONFIG["allow_writes"],
|
|
}
|
|
)
|
|
except Exception as e:
|
|
return JSONResponse(
|
|
{
|
|
"status": "error",
|
|
"error": str(e),
|
|
},
|
|
status_code=503,
|
|
)
|
|
|
|
|
|
def main():
|
|
"""Entry point for the server."""
|
|
logger.info("Starting UniFi MCP Light server...")
|
|
logger.info(f"Controller: {CONFIG['host']}:{CONFIG['port']}")
|
|
logger.info(f"Site: {CONFIG['site']}")
|
|
logger.info(
|
|
f"Write operations: {'enabled' if CONFIG['allow_writes'] else 'disabled'}"
|
|
)
|
|
|
|
# Run with HTTP transport - MCP endpoint at /mcp, health at /health
|
|
mcp.run(transport="http", host="0.0.0.0", port=8000)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|