commit cb57b8f537aa66030e87af4d805badb17b029705 Author: Ben Date: Fri Jan 2 02:21:10 2026 +0000 Initial implementation of UniFi MCP Light server Implements Hybrid MCP Light pattern with: - 4 specific tools: list_clients, list_devices, get_system_info, get_network_health - Meta tools: tool_index, api_call (raw API pass-through) - API documentation resource at unifi://api-reference - Starlette wrapper with /health endpoint - Write protection (UNIFI_ALLOW_WRITES env var) - UniFi OS auto-detection (proxy vs direct paths) - Docker multi-stage build - Gitea CI workflow Closes #1, #2, #3, #4, #5, #7 diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..7315423 --- /dev/null +++ b/.env.example @@ -0,0 +1,12 @@ +# UniFi Controller Configuration +UNIFI_HOST=192.168.1.1 +UNIFI_USERNAME=admin +UNIFI_PASSWORD=your_password_here +UNIFI_PORT=443 +UNIFI_SITE=default +UNIFI_VERIFY_SSL=false + +# Security Settings +# Enable write operations (POST, PUT, DELETE, PATCH) via unifi_api_call +# Default: false (read-only mode) +UNIFI_ALLOW_WRITES=false diff --git a/.gitea/workflows/build.yaml b/.gitea/workflows/build.yaml new file mode 100644 index 0000000..56d4a7a --- /dev/null +++ b/.gitea/workflows/build.yaml @@ -0,0 +1,30 @@ +name: Build and Push Docker Image + +on: + push: + branches: + - main + - master + +jobs: + build: + runs-on: docker + container: + image: catthehacker/ubuntu:act-latest + steps: + - name: Checkout + uses: actions/checkout@v3 + + - name: Login to Gitea Container Registry + uses: docker/login-action@v2 + with: + registry: gitea.ext.ben.io + username: ${{ gitea.actor }} + password: ${{ secrets.CR_PAT }} + + - name: Build and Push + uses: docker/build-push-action@v4 + with: + context: . + push: true + tags: gitea.ext.ben.io/${{ gitea.repository }}:latest diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0a3fd58 --- /dev/null +++ b/.gitignore @@ -0,0 +1,48 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# Virtual environments +.venv/ +venv/ +ENV/ +env/ + +# Environment files +.env +.env.local +.env.*.local + +# IDE +.idea/ +.vscode/ +*.swp +*.swo +*~ + +# Testing +.pytest_cache/ +.coverage +htmlcov/ + +# OS +.DS_Store +Thumbs.db diff --git a/BLUEPRINT.md b/BLUEPRINT.md new file mode 120000 index 0000000..5a9cc4e --- /dev/null +++ b/BLUEPRINT.md @@ -0,0 +1 @@ +../BLUEPRINT.md \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..fe2f86a --- /dev/null +++ b/Dockerfile @@ -0,0 +1,39 @@ +# Build stage +FROM python:3.12-slim AS builder + +WORKDIR /app + +# Install build dependencies +RUN pip install --no-cache-dir uv + +# Copy dependency files +COPY pyproject.toml ./ + +# Create virtual environment and install dependencies +RUN uv venv /app/.venv +ENV PATH="/app/.venv/bin:$PATH" +RUN uv pip install --no-cache . + +# Runtime stage +FROM python:3.12-slim + +WORKDIR /app + +# Copy virtual environment from builder +COPY --from=builder /app/.venv /app/.venv +ENV PATH="/app/.venv/bin:$PATH" + +# Copy application code +COPY server.py unifi_client.py api_docs.py ./ + +# Create non-root user +RUN useradd --create-home --shell /bin/bash appuser +USER appuser + +# Health check +HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ + CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1 + +EXPOSE 8000 + +CMD ["python", "server.py"] diff --git a/IMPLEMENTATION.md b/IMPLEMENTATION.md new file mode 100644 index 0000000..f7cf2df --- /dev/null +++ b/IMPLEMENTATION.md @@ -0,0 +1,215 @@ +# Implementation Details + +Technical documentation for the UniFi MCP Light server architecture. + +## Architecture Overview + +``` +┌─────────────────────────────────────────────────────────┐ +│ Starlette ASGI App │ +├─────────────────────────────────────────────────────────┤ +│ /health │ / (MCP via FastMCP) │ +│ Health Check │ ┌────────────────────────────┐ │ +│ │ │ MCP Protocol │ │ +│ │ │ - Tools (6) │ │ +│ │ │ - Resources (1) │ │ +│ │ └────────────────────────────┘ │ +├─────────────────────────────────────────────────────────┤ +│ UnifiClient │ +│ - Session management │ +│ - UniFi OS detection │ +│ - Write protection │ +│ - Cookie-based auth │ +├─────────────────────────────────────────────────────────┤ +│ aiohttp │ +│ - HTTPS connections │ +│ - SSL handling │ +│ - Cookie jar │ +└─────────────────────────────────────────────────────────┘ +``` + +## File Structure + +``` +unifi-mcp-light/ +├── server.py # MCP server + Starlette wrapper + tools +├── unifi_client.py # UniFi API client abstraction +├── api_docs.py # Embedded API documentation +├── pyproject.toml # Python package configuration +├── Dockerfile # Multi-stage container build +├── docker-compose.yml # Container orchestration +└── .gitea/workflows/ + └── build.yaml # CI pipeline +``` + +## UniFi Client Design + +### Session Management + +The `UnifiClient` class manages a single `aiohttp.ClientSession` with: +- Cookie jar for session persistence +- SSL context (verification optional) +- CSRF token extraction (UniFi OS) + +```python +async with UnifiClient(host, username, password) as client: + clients = await client.get_clients() +``` + +### UniFi OS Detection + +Modern UniFi OS devices (Cloud Gateway, UDM-Pro, etc.) use a proxy path structure: + +| Controller Type | API Path | +|-----------------|----------| +| UniFi OS | `/proxy/network/api/s/{site}/...` | +| Standalone | `/api/s/{site}/...` | + +Detection probes both paths during initialization: + +```python +async def _detect_controller_type(self): + # Try UniFi OS path first (more common) + try: + resp = await session.get(f"{base}/proxy/network/api/s/{site}/stat/health") + if resp.status in (200, 401, 403): + self._is_unifi_os = True + return + except: + pass + + # Fall back to direct path + ... +``` + +### Write Protection + +Write operations require explicit opt-in: + +```python +WRITE_METHODS = frozenset({'POST', 'PUT', 'DELETE', 'PATCH'}) + +async def request(self, method, endpoint, payload=None): + if method.upper() in WRITE_METHODS and not self.allow_writes: + raise UnifiWriteBlockedError( + "Write operation blocked. Set UNIFI_ALLOW_WRITES=true" + ) +``` + +## API Pass-through Pattern + +The `unifi_api_call` tool provides the "escape hatch": + +1. Accepts any endpoint path +2. Automatically prepends UniFi OS prefix if needed +3. Parses JSON params for request body +4. Enforces write protection +5. Returns raw API response + +```python +@mcp.tool() +async def unifi_api_call(endpoint: str, method: str = "GET", params: str = "{}"): + payload = json.loads(params) if params != "{}" else None + client = await get_client() + result = await client.request(method, endpoint, payload) + return json.dumps(result) +``` + +## Error Handling + +All errors are returned as JSON objects: + +```json +{ + "error": "Error message here", + "hint": "Optional suggestion for resolution" +} +``` + +Exception hierarchy: +- `UnifiClientError`: Base class for all client errors +- `UnifiAuthError`: Authentication failures +- `UnifiWriteBlockedError`: Write operation attempted without permission + +## Health Check + +The `/health` endpoint verifies: + +1. Client is connected +2. Controller is reachable +3. Authentication is valid + +Response format: +```json +{ + "status": "ok", + "controller": "192.168.1.1", + "site": "default", + "write_enabled": false +} +``` + +Error response (503): +```json +{ + "status": "error", + "error": "Connection refused" +} +``` + +## MCP Resources + +The `unifi://api-reference` resource provides: +- Endpoint documentation +- Request/response examples +- Query parameter options +- Write operation requirements + +This enables AI agents to discover API capabilities without external documentation. + +## Docker Build + +Multi-stage build for minimal image size: + +1. **Builder stage**: Install dependencies with UV +2. **Runtime stage**: Copy venv, run as non-root user + +```dockerfile +FROM python:3.12-slim AS builder +# Install dependencies +RUN uv venv && uv pip install . + +FROM python:3.12-slim +# Copy venv, set non-root user +COPY --from=builder /app/.venv /app/.venv +USER appuser +CMD ["python", "server.py"] +``` + +## Extending + +### Adding a New Tool + +1. Add tool function to `server.py`: +```python +@mcp.tool() +async def unifi_new_tool() -> str: + client = await get_client() + result = await client.request("GET", f"/api/s/{client.site}/...") + return json.dumps(result) +``` + +2. Add to `unifi_tool_index` catalog + +3. Consider if it should count toward the 4-tool limit + +### Adding Client Methods + +Add convenience methods to `UnifiClient`: +```python +async def get_events(self, limit: int = 100) -> list: + return await self.request( + "GET", + f"/api/s/{self.site}/stat/event?_limit={limit}" + ) +``` diff --git a/README.md b/README.md new file mode 100644 index 0000000..01f9d79 --- /dev/null +++ b/README.md @@ -0,0 +1,158 @@ +# UniFi MCP Light + +A **Hybrid MCP Light Server** for UniFi Network Controller following the minimal-surface-area pattern with raw API pass-through. + +## Features + +- **4 Specific Tools**: Common read operations for clients, devices, system info, and health +- **Raw API Pass-through**: Access any UniFi API endpoint via `unifi_api_call` +- **Embedded Documentation**: API reference available as MCP Resource +- **Write Protection**: Mutations disabled by default, enable with `UNIFI_ALLOW_WRITES=true` +- **UniFi OS Support**: Auto-detects Cloud Gateway, UDM, and standalone controllers +- **Health Endpoint**: `/health` for Docker health checks + +## Quick Start + +### Docker (Recommended) + +```bash +docker run -d \ + -p 8100:8000 \ + -e UNIFI_HOST=192.168.1.1 \ + -e UNIFI_USERNAME=admin \ + -e UNIFI_PASSWORD=your_password \ + gitea.ext.ben.io/b3nw/unifi-mcp-light:latest +``` + +### Docker Compose + +```bash +cp .env.example .env +# Edit .env with your UniFi credentials +docker-compose up -d +``` + +### Python / UV + +```bash +# Install UV +curl -fsSL https://astral.sh/uv/install.sh | bash + +# Clone and install +git clone gitea@git.local.ben.io:b3nw/unifi-mcp-light.git +cd unifi-mcp-light +uv venv && source .venv/bin/activate +uv pip install -e . + +# Configure +cp .env.example .env +# Edit .env with your credentials + +# Run +python server.py +``` + +## Configuration + +| Variable | Description | Default | +|----------|-------------|---------| +| `UNIFI_HOST` | Controller IP/hostname | **required** | +| `UNIFI_USERNAME` | Admin username | **required** | +| `UNIFI_PASSWORD` | Admin password | **required** | +| `UNIFI_PORT` | HTTPS port | `443` | +| `UNIFI_SITE` | Site name | `default` | +| `UNIFI_VERIFY_SSL` | Verify SSL certificate | `false` | +| `UNIFI_ALLOW_WRITES` | Enable write operations | `false` | +| `PORT` | Host port (docker-compose) | `8100` | + +## Tools + +### Specific Tools (4) + +| Tool | Description | +|------|-------------| +| `unifi_list_clients` | List connected network clients (phones, laptops, IoT) | +| `unifi_list_devices` | List UniFi devices (APs, switches, gateways) | +| `unifi_get_system_info` | Get controller version and configuration | +| `unifi_get_network_health` | Get health status per subsystem (WAN, LAN, WLAN) | + +### Meta Tools + +| Tool | Description | +|------|-------------| +| `unifi_tool_index` | Get catalog of all available tools with schemas | +| `unifi_api_call` | Raw API pass-through (the "escape hatch") | + +### Resources + +| URI | Description | +|-----|-------------| +| `unifi://api-reference` | Curated UniFi API documentation | + +## Usage Examples + +### List Connected Clients + +``` +Use the unifi_list_clients tool +``` + +### Raw API Call (Read) + +``` +Use unifi_api_call with: +- endpoint: /api/s/default/stat/event +- method: GET +``` + +### Raw API Call (Write - requires UNIFI_ALLOW_WRITES=true) + +``` +Use unifi_api_call with: +- endpoint: /api/s/default/cmd/stamgr +- method: POST +- params: {"cmd": "block-sta", "mac": "aa:bb:cc:dd:ee:ff"} +``` + +## Claude Desktop Integration + +Add to your `claude_desktop_config.json`: + +```json +{ + "mcpServers": { + "unifi": { + "command": "docker", + "args": [ + "run", "--rm", "-i", + "-e", "UNIFI_HOST=192.168.1.1", + "-e", "UNIFI_USERNAME=admin", + "-e", "UNIFI_PASSWORD=your_password", + "gitea.ext.ben.io/b3nw/unifi-mcp-light:latest" + ] + } + } +} +``` + +## Security + +- **Write operations are disabled by default** +- Set `UNIFI_ALLOW_WRITES=true` only when needed +- Use read-only credentials when possible +- Don't expose the server to the internet without authentication + +## Architecture + +This server follows the **Hybrid MCP Light** pattern: + +1. **Minimal Surface Area**: Only 4 specific tools for common operations +2. **Escape Hatch**: Raw API pass-through for any other operation +3. **Embedded Documentation**: API docs as MCP Resource +4. **Starlette Wrapper**: Health checks for container orchestration + +See [IMPLEMENTATION.md](IMPLEMENTATION.md) for technical details. + +## License + +MIT diff --git a/api_docs.py b/api_docs.py new file mode 100644 index 0000000..f2e1e6c --- /dev/null +++ b/api_docs.py @@ -0,0 +1,357 @@ +""" +UniFi API Documentation Resource + +Curated API reference for the UniFi Network Controller. +Exposed as an MCP Resource to help AI agents use the raw API pass-through tool. +""" + +API_DOCS = """ +# UniFi Network Controller API Reference + +This document provides a curated reference for the UniFi Network Controller API. +Use this with the `unifi_api_call` tool to execute raw API requests. + +## Important Notes + +- All endpoints use the site variable `{site}` which defaults to `default` +- The server automatically handles UniFi OS path prefixing (`/proxy/network`) +- Authentication is handled automatically via session cookies +- Write operations (POST, PUT, DELETE) require `UNIFI_ALLOW_WRITES=true` + +## Response Format + +All UniFi API responses follow this structure: +```json +{ + "meta": {"rc": "ok"}, + "data": [...] +} +``` + +The `unifi_api_call` tool returns only the `data` array. + +--- + +## Clients (Stations) + +### GET /api/s/{site}/stat/sta +List all currently connected clients. + +**Response fields:** +- `mac`: Client MAC address +- `ip`: IP address +- `hostname`: Client hostname +- `name`: Friendly name (if set) +- `oui`: Manufacturer OUI +- `is_wired`: Boolean, true if wired connection +- `is_guest`: Boolean, true if on guest network +- `rx_bytes`, `tx_bytes`: Traffic counters +- `uptime`: Connection uptime in seconds +- `last_seen`: Unix timestamp of last activity +- `essid`: Connected SSID (wireless only) +- `channel`: WiFi channel (wireless only) +- `signal`: Signal strength in dBm (wireless only) + +### GET /api/s/{site}/stat/alluser +List all known clients (including offline/historical). + +### GET /api/s/{site}/stat/user/{mac} +Get details for a specific client by MAC address. + +### POST /api/s/{site}/cmd/stamgr +Client management commands. Requires `UNIFI_ALLOW_WRITES=true`. + +**Block a client:** +```json +{"cmd": "block-sta", "mac": "aa:bb:cc:dd:ee:ff"} +``` + +**Unblock a client:** +```json +{"cmd": "unblock-sta", "mac": "aa:bb:cc:dd:ee:ff"} +``` + +**Force reconnect:** +```json +{"cmd": "kick-sta", "mac": "aa:bb:cc:dd:ee:ff"} +``` + +--- + +## Devices + +### GET /api/s/{site}/stat/device +List all UniFi network devices (APs, switches, gateways). + +**Response fields:** +- `mac`: Device MAC address +- `ip`: Management IP address +- `name`: Device name +- `model`: Model identifier (e.g., "U7PG2") +- `model_in_lts`: Long-term support model name +- `type`: Device type (uap, usw, ugw, udm) +- `version`: Firmware version +- `adopted`: Boolean, true if adopted +- `state`: Device state (1=connected, 0=disconnected) +- `uptime`: Uptime in seconds +- `num_sta`: Number of connected clients (APs) +- `port_table`: Port status array (switches) + +### GET /api/s/{site}/stat/device/{mac} +Get details for a specific device by MAC address. + +### POST /api/s/{site}/cmd/devmgr +Device management commands. Requires `UNIFI_ALLOW_WRITES=true`. + +**Restart device:** +```json +{"cmd": "restart", "mac": "aa:bb:cc:dd:ee:ff"} +``` + +**Adopt device:** +```json +{"cmd": "adopt", "mac": "aa:bb:cc:dd:ee:ff"} +``` + +**Upgrade firmware:** +```json +{"cmd": "upgrade", "mac": "aa:bb:cc:dd:ee:ff"} +``` + +**Locate device (flash LED):** +```json +{"cmd": "set-locate", "mac": "aa:bb:cc:dd:ee:ff"} +``` + +**Stop locating:** +```json +{"cmd": "unset-locate", "mac": "aa:bb:cc:dd:ee:ff"} +``` + +--- + +## Networks & VLANs + +### GET /api/s/{site}/rest/networkconf +List all network configurations. + +**Response fields:** +- `_id`: Network ID +- `name`: Network name +- `purpose`: Network purpose (corporate, guest, vlan-only) +- `vlan`: VLAN ID +- `subnet`: Network subnet (CIDR) +- `dhcpd_enabled`: DHCP server enabled +- `dhcpd_start`, `dhcpd_stop`: DHCP range +- `domain_name`: DHCP domain name +- `igmp_snooping`: IGMP snooping enabled + +### GET /api/s/{site}/rest/networkconf/{id} +Get specific network by ID. + +### POST /api/s/{site}/rest/networkconf +Create new network. Requires `UNIFI_ALLOW_WRITES=true`. + +### PUT /api/s/{site}/rest/networkconf/{id} +Update network. Requires `UNIFI_ALLOW_WRITES=true`. + +### DELETE /api/s/{site}/rest/networkconf/{id} +Delete network. Requires `UNIFI_ALLOW_WRITES=true`. + +--- + +## Wireless Networks (WLANs) + +### GET /api/s/{site}/rest/wlanconf +List all wireless network configurations. + +**Response fields:** +- `_id`: WLAN ID +- `name`: SSID name +- `enabled`: Boolean +- `security`: Security type (wpapsk, open) +- `wpa_mode`: WPA mode (wpa2, wpa3) +- `x_passphrase`: WiFi password +- `hide_ssid`: Hidden network +- `is_guest`: Guest network +- `vlan`: VLAN ID +- `schedule`: Scheduling configuration + +### POST /api/s/{site}/rest/wlanconf +Create new WLAN. Requires `UNIFI_ALLOW_WRITES=true`. + +### PUT /api/s/{site}/rest/wlanconf/{id} +Update WLAN. Requires `UNIFI_ALLOW_WRITES=true`. + +--- + +## Firewall + +### GET /api/s/{site}/rest/firewallrule +List firewall rules (legacy format). + +### GET /api/s/{site}/rest/firewallgroup +List firewall groups (IP groups, port groups). + +**Response fields (group):** +- `_id`: Group ID +- `name`: Group name +- `group_type`: Type (address-group, port-group) +- `group_members`: Array of IPs or ports + +--- + +## Port Forwarding + +### GET /api/s/{site}/rest/portforward +List port forwarding rules. + +**Response fields:** +- `_id`: Rule ID +- `name`: Rule name +- `enabled`: Boolean +- `src`: Source restriction (any, limited) +- `dst_port`: Destination port(s) +- `fwd`: Forward to IP address +- `fwd_port`: Forward to port +- `proto`: Protocol (tcp, udp, tcp_udp) + +### POST /api/s/{site}/rest/portforward +Create port forward. Requires `UNIFI_ALLOW_WRITES=true`. + +--- + +## Routing + +### GET /api/s/{site}/rest/routing +List static routes. + +**Response fields:** +- `_id`: Route ID +- `name`: Route name +- `enabled`: Boolean +- `type`: Route type (static, interface) +- `network`: Destination network +- `gateway`: Next hop gateway +- `interface`: Interface name + +--- + +## Statistics & Health + +### GET /api/s/{site}/stat/health +Get overall network health status. + +**Response fields:** +- `subsystem`: Subsystem name (wan, lan, wlan, vpn) +- `status`: Status (ok, warning, error) +- `num_user`: Connected user count +- `num_guest`: Connected guest count +- `tx_bytes-r`, `rx_bytes-r`: Transfer rates + +### GET /api/s/{site}/stat/sysinfo +Get system information. + +**Response fields:** +- `version`: Controller version +- `build`: Build number +- `hostname`: Controller hostname +- `name`: Site name +- `timezone`: Timezone + +### GET /api/s/{site}/stat/event +List recent events. + +**Query parameters:** +- `within`: Time range in hours (default: 24) +- `_limit`: Maximum events to return + +**Example:** `/api/s/default/stat/event?within=24&_limit=100` + +### GET /api/s/{site}/stat/alarm +List active alarms. + +--- + +## VPN + +### GET /api/s/{site}/rest/vpnclient +List VPN client configurations. + +### GET /api/s/{site}/rest/vpnserver +List VPN server configurations. + +--- + +## Hotspot & Vouchers + +### GET /api/s/{site}/stat/voucher +List hotspot vouchers. + +**Response fields:** +- `code`: Voucher code +- `create_time`: Creation timestamp +- `duration`: Duration in minutes +- `quota`: Usage quota (0=unlimited) +- `used`: Times used +- `qos_overwrite`: QoS override enabled + +### POST /api/s/{site}/cmd/hotspot +Hotspot commands. Requires `UNIFI_ALLOW_WRITES=true`. + +**Create vouchers:** +```json +{ + "cmd": "create-voucher", + "n": 5, + "quota": 1, + "expire": 1440, + "note": "Event guests" +} +``` + +**Revoke voucher:** +```json +{"cmd": "delete-voucher", "_id": "voucher_id_here"} +``` + +--- + +## User Groups (Bandwidth Profiles) + +### GET /api/s/{site}/rest/usergroup +List user groups / bandwidth profiles. + +**Response fields:** +- `_id`: Group ID +- `name`: Group name +- `qos_rate_max_up`: Upload limit (kbps, -1=unlimited) +- `qos_rate_max_down`: Download limit (kbps, -1=unlimited) + +--- + +## DPI (Deep Packet Inspection) + +### GET /api/s/{site}/stat/dpi +Get DPI statistics (application usage). + +**Response fields:** +- `by_app`: Usage by application +- `by_cat`: Usage by category + +--- + +## Common Query Parameters + +Many list endpoints support: +- `_limit`: Maximum results +- `_start`: Offset for pagination +- `_sort`: Sort field (prefix with - for descending) + +**Example:** `/api/s/default/stat/sta?_limit=50&_sort=-rx_bytes` +""" + + +def get_api_docs() -> str: + """Return the curated UniFi API documentation.""" + return API_DOCS diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..4a1f675 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,24 @@ +services: + unifi-mcp-light: + build: . + image: gitea.ext.ben.io/b3nw/unifi-mcp-light:latest + container_name: unifi-mcp-light + restart: unless-stopped + ports: + - "${PORT:-8100}:8000" + environment: + - UNIFI_HOST=${UNIFI_HOST} + - UNIFI_USERNAME=${UNIFI_USERNAME} + - UNIFI_PASSWORD=${UNIFI_PASSWORD} + - UNIFI_PORT=${UNIFI_PORT:-443} + - UNIFI_SITE=${UNIFI_SITE:-default} + - UNIFI_VERIFY_SSL=${UNIFI_VERIFY_SSL:-false} + - UNIFI_ALLOW_WRITES=${UNIFI_ALLOW_WRITES:-false} + env_file: + - .env + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 10s diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..be55670 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,50 @@ +[project] +name = "unifi-mcp-light" +version = "0.1.0" +description = "Hybrid MCP Light Server for UniFi Network Controller - minimal tools with raw API pass-through" +readme = "README.md" +license = {text = "MIT"} +requires-python = ">=3.11" +authors = [ + {name = "b3nw", email = "admin@ben.io"} +] +keywords = ["mcp", "unifi", "network", "api", "llm"] +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", +] + +dependencies = [ + "fastmcp>=2.0.0", + "aiounifi>=80", + "aiohttp>=3.9.0", + "starlette>=0.40.0", + "uvicorn>=0.30.0", + "python-dotenv>=1.0.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0.0", + "pytest-asyncio>=0.23.0", + "aioresponses>=0.7.0", +] + +[project.scripts] +unifi-mcp-light = "server:main" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["."] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] diff --git a/server.py b/server.py new file mode 100644 index 0000000..f8c8ee6 --- /dev/null +++ b/server.py @@ -0,0 +1,427 @@ +""" +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 asyncio +import json +import logging +import os +from contextlib import asynccontextmanager +from typing import Any, Optional + +from dotenv import load_dotenv +from fastmcp import FastMCP +from starlette.applications import Starlette +from starlette.responses import JSONResponse +from starlette.routing import Mount, Route + +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)}) + + +# ============================================================================= +# Starlette Wrapper with Health Endpoint +# ============================================================================= + + +async def health_endpoint(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, + ) + + +@asynccontextmanager +async def lifespan(app): + """Application lifespan manager.""" + 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'}" + ) + + yield + + logger.info("Shutting down UniFi MCP Light server...") + await close_client() + + +def create_app() -> Starlette: + """Create the Starlette ASGI application.""" + mcp_app = mcp.http_app() + + return Starlette( + routes=[ + Route("/health", health_endpoint), + Mount("/", app=mcp_app), + ], + lifespan=lifespan, + ) + + +def main(): + """Entry point for the server.""" + import uvicorn + + app = create_app() + uvicorn.run(app, host="0.0.0.0", port=8000) + + +if __name__ == "__main__": + main() diff --git a/unifi_client.py b/unifi_client.py new file mode 100644 index 0000000..ca79545 --- /dev/null +++ b/unifi_client.py @@ -0,0 +1,299 @@ +""" +UniFi Controller Client + +A clean abstraction layer wrapping aiohttp for UniFi API communication. +Handles authentication, UniFi OS detection, and write protection. +""" + +import asyncio +import json +import logging +import ssl +from typing import Any, Optional + +import aiohttp + +logger = logging.getLogger(__name__) + +# HTTP methods that modify state +WRITE_METHODS = frozenset({"POST", "PUT", "DELETE", "PATCH"}) + + +class UnifiClientError(Exception): + """Base exception for UniFi client errors.""" + + pass + + +class UnifiAuthError(UnifiClientError): + """Authentication failed.""" + + pass + + +class UnifiWriteBlockedError(UnifiClientError): + """Write operation blocked by configuration.""" + + pass + + +class UnifiClient: + """ + Async client for UniFi Network Controller API. + + Features: + - Auto-detects UniFi OS vs standalone controller paths + - Cookie-based session authentication + - Write protection (configurable) + - Proper async context manager support + """ + + def __init__( + self, + host: str, + username: str, + password: str, + port: int = 443, + site: str = "default", + verify_ssl: bool = False, + allow_writes: bool = False, + ): + self.host = host + self.username = username + self.password = password + self.port = port + self.site = site + self.verify_ssl = verify_ssl + self.allow_writes = allow_writes + + self._session: Optional[aiohttp.ClientSession] = None + self._is_unifi_os: Optional[bool] = None + self._csrf_token: Optional[str] = None + self._authenticated = False + + @property + def base_url(self) -> str: + return f"https://{self.host}:{self.port}" + + @property + def api_prefix(self) -> str: + """Return the correct API prefix based on controller type.""" + if self._is_unifi_os: + return "/proxy/network" + return "" + + async def __aenter__(self) -> "UnifiClient": + await self.connect() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: + await self.close() + + async def connect(self) -> None: + """Initialize session, detect controller type, and authenticate.""" + if self._session is not None: + return + + # Create SSL context + if self.verify_ssl: + ssl_context = ssl.create_default_context() + else: + ssl_context = ssl.create_default_context() + ssl_context.check_hostname = False + ssl_context.verify_mode = ssl.CERT_NONE + + connector = aiohttp.TCPConnector(ssl=ssl_context) + self._session = aiohttp.ClientSession( + connector=connector, + cookie_jar=aiohttp.CookieJar(unsafe=True), + ) + + # Detect controller type and authenticate + await self._detect_controller_type() + await self._authenticate() + + async def close(self) -> None: + """Close the HTTP session.""" + if self._session: + await self._session.close() + self._session = None + self._authenticated = False + + async def _detect_controller_type(self) -> None: + """ + Detect if this is a UniFi OS controller (UDM, Cloud Gateway, etc.) + or a standalone Network Controller. + + UniFi OS uses /proxy/network prefix for Network API calls. + """ + # Try UniFi OS path first (more common in modern setups) + try: + async with self._session.get( + f"{self.base_url}/proxy/network/api/s/{self.site}/stat/health", + timeout=aiohttp.ClientTimeout(total=5), + ) as resp: + if resp.status in (200, 401, 403): + # Path exists, this is UniFi OS + self._is_unifi_os = True + logger.info("Detected UniFi OS controller (proxy path)") + return + except Exception: + pass + + # Try direct path (standalone controller) + try: + async with self._session.get( + f"{self.base_url}/api/s/{self.site}/stat/health", + timeout=aiohttp.ClientTimeout(total=5), + ) as resp: + if resp.status in (200, 401, 403): + self._is_unifi_os = False + logger.info("Detected standalone UniFi controller (direct path)") + return + except Exception: + pass + + # Default to UniFi OS if detection fails + self._is_unifi_os = True + logger.warning("Controller type detection failed, defaulting to UniFi OS") + + async def _authenticate(self) -> None: + """Authenticate with the UniFi controller.""" + login_url = f"{self.base_url}{self.api_prefix}/api/login" + + payload = { + "username": self.username, + "password": self.password, + } + + try: + async with self._session.post( + login_url, + json=payload, + timeout=aiohttp.ClientTimeout(total=10), + ) as resp: + if resp.status == 200: + self._authenticated = True + # Extract CSRF token if present (UniFi OS) + if "x-csrf-token" in resp.headers: + self._csrf_token = resp.headers["x-csrf-token"] + logger.info(f"Successfully authenticated to {self.host}") + elif resp.status == 401: + raise UnifiAuthError("Invalid username or password") + else: + text = await resp.text() + raise UnifiAuthError( + f"Authentication failed: {resp.status} - {text}" + ) + except aiohttp.ClientError as e: + raise UnifiClientError(f"Connection error during authentication: {e}") + + async def ping(self) -> bool: + """ + Check if the connection to the controller is alive. + Returns True if connected, raises exception otherwise. + """ + if not self._authenticated: + raise UnifiClientError("Not authenticated") + + try: + result = await self.request("GET", f"/api/s/{self.site}/stat/health") + return True + except Exception as e: + raise UnifiClientError(f"Controller ping failed: {e}") + + async def request( + self, + method: str, + endpoint: str, + payload: Optional[dict] = None, + ) -> Any: + """ + Make an authenticated request to the UniFi API. + + Args: + method: HTTP method (GET, POST, PUT, DELETE, PATCH) + endpoint: API endpoint (e.g., /api/s/default/stat/sta) + payload: Optional request body for POST/PUT requests + + Returns: + Parsed JSON response data + + Raises: + UnifiWriteBlockedError: If write operation attempted without permission + UnifiClientError: For other API errors + """ + if not self._session: + raise UnifiClientError( + "Client not connected. Use 'async with' or call connect()" + ) + + # Check write protection + method_upper = method.upper() + if method_upper in WRITE_METHODS and not self.allow_writes: + raise UnifiWriteBlockedError( + f"Write operation ({method_upper}) blocked. " + "Set UNIFI_ALLOW_WRITES=true to enable write operations." + ) + + # Build full URL with correct prefix + url = f"{self.base_url}{self.api_prefix}{endpoint}" + + # Build headers + headers = {} + if self._csrf_token: + headers["x-csrf-token"] = self._csrf_token + + try: + async with self._session.request( + method_upper, + url, + json=payload, + headers=headers, + timeout=aiohttp.ClientTimeout(total=30), + ) as resp: + text = await resp.text() + + if resp.status == 401: + # Session expired, try to re-authenticate + await self._authenticate() + return await self.request(method, endpoint, payload) + + if resp.status >= 400: + raise UnifiClientError(f"API error {resp.status}: {text}") + + try: + data = json.loads(text) + except json.JSONDecodeError: + return {"raw": text} + + # UniFi API returns {"meta": {"rc": "ok"}, "data": [...]} + if isinstance(data, dict): + meta = data.get("meta", {}) + if meta.get("rc") == "error": + raise UnifiClientError( + f"API error: {meta.get('msg', 'Unknown error')}" + ) + if "data" in data: + return data["data"] + + return data + + except aiohttp.ClientError as e: + raise UnifiClientError(f"Request failed: {e}") + + async def get_clients(self) -> list: + """Get all connected clients.""" + return await self.request("GET", f"/api/s/{self.site}/stat/sta") + + async def get_devices(self) -> list: + """Get all network devices.""" + return await self.request("GET", f"/api/s/{self.site}/stat/device") + + async def get_system_info(self) -> list: + """Get system information.""" + return await self.request("GET", f"/api/s/{self.site}/stat/sysinfo") + + async def get_health(self) -> list: + """Get network health status.""" + return await self.request("GET", f"/api/s/{self.site}/stat/health")