Some checks failed
Build and Push Docker Image / build (push) Failing after 12s
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
216 lines
6.5 KiB
Markdown
216 lines
6.5 KiB
Markdown
# 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}"
|
|
)
|
|
```
|