diff --git a/README.md b/README.md index 6f250e5..453cbc2 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Agent Experiments -Experiments with Google Agent Development Kit (ADK) for building AI agents with local Ollama models and Prefect workflows. +Experiments with Google Agent Development Kit (ADK) for building AI agents with local Ollama models and Prefect workflows. Also includes MCP (Model Context Protocol) servers for home automation. ## Quick Start @@ -16,10 +16,19 @@ make web # Run Prefect workflows make prefect-server + +# Run Home Security MCP Server +python -m src.mcp_servers.home_security.server ``` ## Components +### MCP Servers +- **`home_security`**: MCP server for home security automation + - **Cameras**: Manage security cameras (view, record, snapshots, motion detection) + - **Smart Locks**: Control front door lock (lock/unlock, access codes, logs) + - Supports stdio and SSE transports for LLM integration + ### Agents - **`doc_agent`**: Technical writing assistant with Google Docs integration - Uses local Ollama models (`gpt-oss:20b`) via LiteLLM @@ -61,12 +70,93 @@ src/ │ ├── agent.py # Agent definition │ └── tools/ │ └── google_docs_tool.py # Google Docs integration +├── mcp_servers/ +│ └── home_security/ # Home security MCP server +│ ├── server.py # FastMCP server with tools & resources +│ ├── camera.py # Camera management +│ ├── lock.py # Smart lock management +│ └── models.py # Data models └── workflows/ ├── pipeline.py # Main Prefect workflow ├── serve.py # Flow serving entry point └── discover.py # Flow discovery utility ``` +## Home Security MCP Server + +The home security MCP server provides tools and resources for managing security cameras and smart locks. + +### Running the Server + +```bash +# Run with stdio transport (for local LLM clients like Claude Desktop) +python -m src.mcp_servers.home_security.server + +# Run with SSE transport (for remote access) +python -m src.mcp_servers.home_security.server --transport sse --port 8000 +``` + +### Available Resources + +| Resource URI | Description | +|-------------|-------------| +| `security://cameras` | List all cameras | +| `security://cameras/{id}` | Get camera details | +| `security://cameras/{id}/status` | Get camera status | +| `security://cameras/{id}/storage` | Get storage info | +| `security://locks` | List all locks | +| `security://locks/{id}` | Get lock details | +| `security://locks/{id}/status` | Get lock status | +| `security://locks/{id}/battery` | Get battery status | +| `security://alerts` | Get all security alerts | +| `security://system/status` | Get overall system status | + +### Available Tools + +#### Camera Tools +- `get_camera_status` - Get camera operational status +- `start_camera_recording` - Start recording +- `stop_camera_recording` - Stop recording +- `get_camera_stream_url` - Get RTSP stream URL +- `capture_camera_snapshot` - Capture a snapshot +- `configure_motion_detection` - Enable/disable motion detection +- `get_motion_events` - Get recent motion events +- `acknowledge_motion_event` - Acknowledge an event +- `get_camera_recordings` - List available recordings +- `update_camera_settings` - Update camera settings + +#### Lock Tools +- `get_lock_status` - Get lock status +- `lock_door` - Lock the door +- `unlock_door` - Unlock the door (with optional PIN) +- `add_lock_access_code` - Add a new PIN code +- `remove_lock_access_code` - Remove a PIN code +- `list_lock_access_codes` - List all codes +- `deactivate_lock_access_code` - Deactivate a code +- `get_lock_access_logs` - Get access history +- `configure_auto_lock` - Configure auto-lock +- `clear_lock_tamper_alert` - Clear tamper alert + +#### Alert Tools +- `get_security_alerts` - Get all alerts +- `acknowledge_security_alert` - Acknowledge an alert + +### Claude Desktop Integration + +Add to your Claude Desktop config (`~/.cursor/mcp.json` or similar): + +```json +{ + "mcpServers": { + "home-security": { + "command": "python", + "args": ["-m", "src.mcp_servers.home_security.server"], + "cwd": "/path/to/this/project" + } + } +} +``` + ## Configuration - **Ollama**: Set `OLLAMA_API_BASE` in `.env` or environment diff --git a/pyproject.toml b/pyproject.toml index 4a955ba..4919ddd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,11 +7,12 @@ authors = [ ] classifiers = [ "Programming Language :: Python", + "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", ] keywords = ['agent-catalog', 'adk', 'agent'] readme = "README.md" -requires-python = ">=3.13" +requires-python = ">=3.12" dependencies = [ "google-adk>=1.10.0", "litellm>=1.0.0", @@ -22,6 +23,7 @@ dependencies = [ "google-api-python-client>=2.100.0", "prefect>=3.0.0", "markdown-it-py>=3.0.0", + "mcp>=1.0.0", ] [build-system] diff --git a/src/mcp_servers/__init__.py b/src/mcp_servers/__init__.py new file mode 100644 index 0000000..89e26ba --- /dev/null +++ b/src/mcp_servers/__init__.py @@ -0,0 +1 @@ +"""MCP Servers for home automation and security.""" diff --git a/src/mcp_servers/home_security/__init__.py b/src/mcp_servers/home_security/__init__.py new file mode 100644 index 0000000..06aa210 --- /dev/null +++ b/src/mcp_servers/home_security/__init__.py @@ -0,0 +1,7 @@ +"""Home Security MCP Server - Cameras and Smart Lock integration.""" + +from .server import mcp +from .camera import CameraManager +from .lock import SmartLockManager + +__all__ = ["mcp", "CameraManager", "SmartLockManager"] diff --git a/src/mcp_servers/home_security/camera.py b/src/mcp_servers/home_security/camera.py new file mode 100644 index 0000000..d99ca27 --- /dev/null +++ b/src/mcp_servers/home_security/camera.py @@ -0,0 +1,324 @@ +"""Camera management module for the Home Security MCP Server.""" + +import uuid +from datetime import datetime, timedelta +from typing import Any + +from .models import ( + Camera, + CameraStatus, + MotionEvent, + MotionSensitivity, + SecurityAlert, +) + + +class CameraManager: + """Manages security cameras and their operations.""" + + def __init__(self) -> None: + self._cameras: dict[str, Camera] = {} + self._motion_events: dict[str, list[MotionEvent]] = {} + self._alerts: list[SecurityAlert] = [] + self._initialize_demo_cameras() + + def _initialize_demo_cameras(self) -> None: + """Initialize demo cameras for testing.""" + demo_cameras = [ + Camera( + id="cam_front_door", + name="Front Door Camera", + location="Front Entrance", + status=CameraStatus.ONLINE, + resolution="4K", + ip_address="192.168.1.101", + ), + Camera( + id="cam_backyard", + name="Backyard Camera", + location="Backyard", + status=CameraStatus.ONLINE, + resolution="1080p", + ip_address="192.168.1.102", + ), + Camera( + id="cam_garage", + name="Garage Camera", + location="Garage", + status=CameraStatus.ONLINE, + resolution="1080p", + night_vision=True, + ip_address="192.168.1.103", + ), + Camera( + id="cam_driveway", + name="Driveway Camera", + location="Driveway", + status=CameraStatus.ONLINE, + resolution="4K", + ip_address="192.168.1.104", + ), + ] + for camera in demo_cameras: + self._cameras[camera.id] = camera + self._motion_events[camera.id] = [] + + def list_cameras(self) -> list[dict[str, Any]]: + """List all registered cameras.""" + return [camera.to_dict() for camera in self._cameras.values()] + + def get_camera(self, camera_id: str) -> dict[str, Any] | None: + """Get details for a specific camera.""" + camera = self._cameras.get(camera_id) + return camera.to_dict() if camera else None + + def get_camera_status(self, camera_id: str) -> dict[str, Any]: + """Get the current status of a camera.""" + camera = self._cameras.get(camera_id) + if not camera: + return {"error": f"Camera {camera_id} not found"} + return { + "camera_id": camera_id, + "name": camera.name, + "status": camera.status.value, + "recording": camera.recording, + "motion_detection": camera.motion_detection, + "last_motion": ( + camera.last_motion_detected.isoformat() if camera.last_motion_detected else None + ), + } + + def start_recording(self, camera_id: str) -> dict[str, Any]: + """Start recording on a camera.""" + camera = self._cameras.get(camera_id) + if not camera: + return {"success": False, "error": f"Camera {camera_id} not found"} + if camera.status == CameraStatus.OFFLINE: + return {"success": False, "error": "Camera is offline"} + camera.recording = True + camera.status = CameraStatus.RECORDING + return { + "success": True, + "camera_id": camera_id, + "message": f"Recording started on {camera.name}", + } + + def stop_recording(self, camera_id: str) -> dict[str, Any]: + """Stop recording on a camera.""" + camera = self._cameras.get(camera_id) + if not camera: + return {"success": False, "error": f"Camera {camera_id} not found"} + camera.recording = False + if camera.status == CameraStatus.RECORDING: + camera.status = CameraStatus.ONLINE + return { + "success": True, + "camera_id": camera_id, + "message": f"Recording stopped on {camera.name}", + } + + def get_stream_url(self, camera_id: str) -> dict[str, Any]: + """Get the streaming URL for a camera.""" + camera = self._cameras.get(camera_id) + if not camera: + return {"error": f"Camera {camera_id} not found"} + if camera.status == CameraStatus.OFFLINE: + return {"error": "Camera is offline"} + stream_url = f"rtsp://{camera.ip_address}:554/stream" + return { + "camera_id": camera_id, + "name": camera.name, + "stream_url": stream_url, + "protocol": "rtsp", + } + + def capture_snapshot(self, camera_id: str) -> dict[str, Any]: + """Capture a snapshot from a camera.""" + camera = self._cameras.get(camera_id) + if not camera: + return {"success": False, "error": f"Camera {camera_id} not found"} + if camera.status == CameraStatus.OFFLINE: + return {"success": False, "error": "Camera is offline"} + timestamp = datetime.now() + snapshot_url = f"/snapshots/{camera_id}/{timestamp.strftime('%Y%m%d_%H%M%S')}.jpg" + return { + "success": True, + "camera_id": camera_id, + "snapshot_url": snapshot_url, + "timestamp": timestamp.isoformat(), + "resolution": camera.resolution, + } + + def set_motion_detection( + self, camera_id: str, enabled: bool, sensitivity: str | None = None + ) -> dict[str, Any]: + """Enable or disable motion detection on a camera.""" + camera = self._cameras.get(camera_id) + if not camera: + return {"success": False, "error": f"Camera {camera_id} not found"} + camera.motion_detection = enabled + if sensitivity: + try: + camera.motion_sensitivity = MotionSensitivity(sensitivity.lower()) + except ValueError: + return {"success": False, "error": f"Invalid sensitivity: {sensitivity}"} + return { + "success": True, + "camera_id": camera_id, + "motion_detection": camera.motion_detection, + "sensitivity": camera.motion_sensitivity.value, + } + + def get_motion_events( + self, + camera_id: str | None = None, + since: datetime | None = None, + limit: int = 50, + ) -> list[dict[str, Any]]: + """Get motion events, optionally filtered by camera and time.""" + events: list[MotionEvent] = [] + if camera_id: + events = self._motion_events.get(camera_id, []) + else: + for camera_events in self._motion_events.values(): + events.extend(camera_events) + + if since: + events = [e for e in events if e.timestamp >= since] + + events.sort(key=lambda x: x.timestamp, reverse=True) + return [e.to_dict() for e in events[:limit]] + + def simulate_motion_event( + self, camera_id: str, zone: str | None = None + ) -> dict[str, Any]: + """Simulate a motion detection event (for testing).""" + camera = self._cameras.get(camera_id) + if not camera: + return {"success": False, "error": f"Camera {camera_id} not found"} + if not camera.motion_detection: + return {"success": False, "error": "Motion detection is disabled"} + + event = MotionEvent( + id=f"motion_{uuid.uuid4().hex[:8]}", + camera_id=camera_id, + timestamp=datetime.now(), + duration_seconds=5.0, + confidence=0.85, + zone=zone or "main", + ) + self._motion_events[camera_id].append(event) + camera.last_motion_detected = event.timestamp + + alert = SecurityAlert( + id=f"alert_{uuid.uuid4().hex[:8]}", + device_type="camera", + device_id=camera_id, + alert_type="motion_detected", + severity="info", + message=f"Motion detected on {camera.name}", + timestamp=event.timestamp, + ) + self._alerts.append(alert) + + return { + "success": True, + "event": event.to_dict(), + "alert": alert.to_dict(), + } + + def acknowledge_motion_event(self, event_id: str) -> dict[str, Any]: + """Acknowledge a motion event.""" + for events in self._motion_events.values(): + for event in events: + if event.id == event_id: + event.acknowledged = True + return {"success": True, "event_id": event_id} + return {"success": False, "error": f"Event {event_id} not found"} + + def get_recordings( + self, + camera_id: str, + start_time: datetime | None = None, + end_time: datetime | None = None, + ) -> list[dict[str, Any]]: + """Get list of recordings for a camera.""" + camera = self._cameras.get(camera_id) + if not camera: + return [] + + if not start_time: + start_time = datetime.now() - timedelta(days=1) + if not end_time: + end_time = datetime.now() + + recordings = [] + current = start_time + while current < end_time: + recording = { + "camera_id": camera_id, + "start_time": current.isoformat(), + "end_time": (current + timedelta(hours=1)).isoformat(), + "duration_seconds": 3600, + "size_mb": 450, + "url": f"/recordings/{camera_id}/{current.strftime('%Y%m%d_%H%M%S')}.mp4", + } + recordings.append(recording) + current += timedelta(hours=1) + return recordings[:24] + + def get_storage_info(self, camera_id: str) -> dict[str, Any]: + """Get storage information for a camera.""" + camera = self._cameras.get(camera_id) + if not camera: + return {"error": f"Camera {camera_id} not found"} + return { + "camera_id": camera_id, + "storage_used_gb": camera.storage_used_gb, + "storage_limit_gb": camera.storage_limit_gb, + "storage_available_gb": camera.storage_limit_gb - camera.storage_used_gb, + "usage_percent": (camera.storage_used_gb / camera.storage_limit_gb) * 100, + } + + def set_camera_settings( + self, + camera_id: str, + resolution: str | None = None, + night_vision: bool | None = None, + ) -> dict[str, Any]: + """Update camera settings.""" + camera = self._cameras.get(camera_id) + if not camera: + return {"success": False, "error": f"Camera {camera_id} not found"} + if resolution: + camera.resolution = resolution + if night_vision is not None: + camera.night_vision = night_vision + return { + "success": True, + "camera_id": camera_id, + "settings": { + "resolution": camera.resolution, + "night_vision": camera.night_vision, + }, + } + + def get_alerts( + self, unacknowledged_only: bool = False, limit: int = 50 + ) -> list[dict[str, Any]]: + """Get security alerts.""" + alerts = self._alerts + if unacknowledged_only: + alerts = [a for a in alerts if not a.acknowledged] + alerts.sort(key=lambda x: x.timestamp, reverse=True) + return [a.to_dict() for a in alerts[:limit]] + + def acknowledge_alert(self, alert_id: str, acknowledged_by: str) -> dict[str, Any]: + """Acknowledge a security alert.""" + for alert in self._alerts: + if alert.id == alert_id: + alert.acknowledged = True + alert.acknowledged_by = acknowledged_by + alert.acknowledged_at = datetime.now() + return {"success": True, "alert_id": alert_id} + return {"success": False, "error": f"Alert {alert_id} not found"} diff --git a/src/mcp_servers/home_security/lock.py b/src/mcp_servers/home_security/lock.py new file mode 100644 index 0000000..faf0d8f --- /dev/null +++ b/src/mcp_servers/home_security/lock.py @@ -0,0 +1,442 @@ +"""Smart Lock management module for the Home Security MCP Server.""" + +import hashlib +import uuid +from datetime import datetime, timedelta +from typing import Any + +from .models import ( + AccessCode, + AccessMethod, + LockAccessLog, + LockStatus, + SecurityAlert, + SmartLock, +) + + +class SmartLockManager: + """Manages smart locks and their operations.""" + + def __init__(self) -> None: + self._locks: dict[str, SmartLock] = {} + self._access_codes: dict[str, list[AccessCode]] = {} + self._access_logs: dict[str, list[LockAccessLog]] = {} + self._alerts: list[SecurityAlert] = [] + self._initialize_demo_lock() + + def _initialize_demo_lock(self) -> None: + """Initialize demo front door lock for testing.""" + front_door = SmartLock( + id="lock_front_door", + name="Front Door Lock", + location="Front Entrance", + status=LockStatus.LOCKED, + battery_level=85, + auto_lock_enabled=True, + auto_lock_delay_seconds=30, + connected=True, + firmware_version="2.1.3", + ) + self._locks[front_door.id] = front_door + self._access_codes[front_door.id] = [] + self._access_logs[front_door.id] = [] + + master_code = AccessCode( + id="code_master", + lock_id=front_door.id, + name="Master Code", + code_hash=self._hash_code("1234"), + active=True, + ) + self._access_codes[front_door.id].append(master_code) + + guest_code = AccessCode( + id="code_guest", + lock_id=front_door.id, + name="Guest Code", + code_hash=self._hash_code("5678"), + active=True, + valid_until=datetime.now() + timedelta(days=7), + ) + self._access_codes[front_door.id].append(guest_code) + + def _hash_code(self, code: str) -> str: + """Hash an access code for storage.""" + return hashlib.sha256(code.encode()).hexdigest() + + def _log_access( + self, + lock_id: str, + action: str, + method: AccessMethod, + success: bool = True, + user_name: str | None = None, + access_code_id: str | None = None, + failure_reason: str | None = None, + ) -> LockAccessLog: + """Log an access attempt.""" + log = LockAccessLog( + id=f"log_{uuid.uuid4().hex[:8]}", + lock_id=lock_id, + timestamp=datetime.now(), + action=action, + method=method, + user_name=user_name, + access_code_id=access_code_id, + success=success, + failure_reason=failure_reason, + ) + self._access_logs[lock_id].append(log) + + lock = self._locks.get(lock_id) + if lock: + lock.last_activity = log.timestamp + + return log + + def _create_alert( + self, + lock_id: str, + alert_type: str, + severity: str, + message: str, + ) -> SecurityAlert: + """Create a security alert.""" + alert = SecurityAlert( + id=f"alert_{uuid.uuid4().hex[:8]}", + device_type="lock", + device_id=lock_id, + alert_type=alert_type, + severity=severity, + message=message, + timestamp=datetime.now(), + ) + self._alerts.append(alert) + return alert + + def list_locks(self) -> list[dict[str, Any]]: + """List all registered smart locks.""" + return [lock.to_dict() for lock in self._locks.values()] + + def get_lock(self, lock_id: str) -> dict[str, Any] | None: + """Get details for a specific lock.""" + lock = self._locks.get(lock_id) + return lock.to_dict() if lock else None + + def get_lock_status(self, lock_id: str) -> dict[str, Any]: + """Get the current status of a lock.""" + lock = self._locks.get(lock_id) + if not lock: + return {"error": f"Lock {lock_id} not found"} + return { + "lock_id": lock_id, + "name": lock.name, + "status": lock.status.value, + "battery_level": lock.battery_level, + "connected": lock.connected, + "tamper_alert": lock.tamper_alert, + "last_activity": lock.last_activity.isoformat() if lock.last_activity else None, + } + + def lock(self, lock_id: str, user: str = "system") -> dict[str, Any]: + """Lock the door.""" + lock = self._locks.get(lock_id) + if not lock: + return {"success": False, "error": f"Lock {lock_id} not found"} + if not lock.connected: + return {"success": False, "error": "Lock is not connected"} + if lock.status == LockStatus.JAMMED: + return {"success": False, "error": "Lock is jammed"} + + lock.status = LockStatus.LOCKED + log = self._log_access( + lock_id=lock_id, + action="lock", + method=AccessMethod.REMOTE, + user_name=user, + ) + + return { + "success": True, + "lock_id": lock_id, + "status": lock.status.value, + "message": f"{lock.name} is now locked", + "log_id": log.id, + } + + def unlock( + self, + lock_id: str, + code: str | None = None, + user: str = "system", + method: str = "remote", + ) -> dict[str, Any]: + """Unlock the door with optional code verification.""" + lock = self._locks.get(lock_id) + if not lock: + return {"success": False, "error": f"Lock {lock_id} not found"} + if not lock.connected: + return {"success": False, "error": "Lock is not connected"} + if lock.status == LockStatus.JAMMED: + return {"success": False, "error": "Lock is jammed"} + + access_method = AccessMethod.REMOTE + access_code_id = None + + if code: + access_method = AccessMethod.PIN_CODE + code_hash = self._hash_code(code) + valid_code = self._validate_code(lock_id, code_hash) + if not valid_code: + log = self._log_access( + lock_id=lock_id, + action="unlock", + method=access_method, + success=False, + failure_reason="Invalid code", + ) + failed_attempts = sum( + 1 for log_entry in self._access_logs[lock_id][-10:] + if not log_entry.success and log_entry.method == AccessMethod.PIN_CODE + ) + if failed_attempts >= 3: + self._create_alert( + lock_id=lock_id, + alert_type="multiple_failed_attempts", + severity="warning", + message=f"Multiple failed unlock attempts on {lock.name}", + ) + return { + "success": False, + "error": "Invalid access code", + "log_id": log.id, + } + access_code_id = valid_code.id + valid_code.use_count += 1 + valid_code.last_used = datetime.now() + user = valid_code.name + elif method: + try: + access_method = AccessMethod(method.lower()) + except ValueError: + access_method = AccessMethod.REMOTE + + lock.status = LockStatus.UNLOCKED + log = self._log_access( + lock_id=lock_id, + action="unlock", + method=access_method, + user_name=user, + access_code_id=access_code_id, + ) + + return { + "success": True, + "lock_id": lock_id, + "status": lock.status.value, + "message": f"{lock.name} is now unlocked", + "log_id": log.id, + "auto_lock_in_seconds": ( + lock.auto_lock_delay_seconds if lock.auto_lock_enabled else None + ), + } + + def _validate_code(self, lock_id: str, code_hash: str) -> AccessCode | None: + """Validate an access code.""" + codes = self._access_codes.get(lock_id, []) + now = datetime.now() + + for code in codes: + if code.code_hash != code_hash: + continue + if not code.active: + continue + if code.valid_from and now < code.valid_from: + continue + if code.valid_until and now > code.valid_until: + continue + if code.max_uses and code.use_count >= code.max_uses: + continue + if code.allowed_days and now.strftime("%A").lower() not in [ + d.lower() for d in code.allowed_days + ]: + continue + if code.allowed_hours_start is not None and code.allowed_hours_end is not None: + if not (code.allowed_hours_start <= now.hour < code.allowed_hours_end): + continue + return code + + return None + + def add_access_code( + self, + lock_id: str, + name: str, + code: str, + one_time: bool = False, + valid_days: int | None = None, + allowed_days: list[str] | None = None, + allowed_hours_start: int | None = None, + allowed_hours_end: int | None = None, + ) -> dict[str, Any]: + """Add a new access code.""" + lock = self._locks.get(lock_id) + if not lock: + return {"success": False, "error": f"Lock {lock_id} not found"} + + if len(code) < 4: + return {"success": False, "error": "Code must be at least 4 digits"} + + access_code = AccessCode( + id=f"code_{uuid.uuid4().hex[:8]}", + lock_id=lock_id, + name=name, + code_hash=self._hash_code(code), + one_time=one_time, + max_uses=1 if one_time else None, + valid_until=datetime.now() + timedelta(days=valid_days) if valid_days else None, + allowed_days=allowed_days or [], + allowed_hours_start=allowed_hours_start, + allowed_hours_end=allowed_hours_end, + ) + self._access_codes[lock_id].append(access_code) + + return { + "success": True, + "code_id": access_code.id, + "name": name, + "lock_id": lock_id, + } + + def remove_access_code(self, lock_id: str, code_id: str) -> dict[str, Any]: + """Remove an access code.""" + if lock_id not in self._access_codes: + return {"success": False, "error": f"Lock {lock_id} not found"} + + codes = self._access_codes[lock_id] + for i, code in enumerate(codes): + if code.id == code_id: + codes.pop(i) + return {"success": True, "message": f"Code {code.name} removed"} + + return {"success": False, "error": f"Code {code_id} not found"} + + def list_access_codes(self, lock_id: str) -> list[dict[str, Any]]: + """List all access codes for a lock.""" + codes = self._access_codes.get(lock_id, []) + return [code.to_dict() for code in codes] + + def deactivate_access_code(self, lock_id: str, code_id: str) -> dict[str, Any]: + """Deactivate an access code.""" + if lock_id not in self._access_codes: + return {"success": False, "error": f"Lock {lock_id} not found"} + + for code in self._access_codes[lock_id]: + if code.id == code_id: + code.active = False + return {"success": True, "message": f"Code {code.name} deactivated"} + + return {"success": False, "error": f"Code {code_id} not found"} + + def get_access_logs( + self, + lock_id: str, + since: datetime | None = None, + limit: int = 50, + ) -> list[dict[str, Any]]: + """Get access logs for a lock.""" + logs = self._access_logs.get(lock_id, []) + if since: + logs = [log for log in logs if log.timestamp >= since] + logs.sort(key=lambda x: x.timestamp, reverse=True) + return [log.to_dict() for log in logs[:limit]] + + def set_auto_lock( + self, + lock_id: str, + enabled: bool, + delay_seconds: int | None = None, + ) -> dict[str, Any]: + """Configure auto-lock settings.""" + lock = self._locks.get(lock_id) + if not lock: + return {"success": False, "error": f"Lock {lock_id} not found"} + + lock.auto_lock_enabled = enabled + if delay_seconds is not None: + lock.auto_lock_delay_seconds = delay_seconds + + return { + "success": True, + "lock_id": lock_id, + "auto_lock_enabled": lock.auto_lock_enabled, + "auto_lock_delay_seconds": lock.auto_lock_delay_seconds, + } + + def get_battery_status(self, lock_id: str) -> dict[str, Any]: + """Get battery status for a lock.""" + lock = self._locks.get(lock_id) + if not lock: + return {"error": f"Lock {lock_id} not found"} + + status = "good" + if lock.battery_level < 20: + status = "critical" + elif lock.battery_level < 40: + status = "low" + + return { + "lock_id": lock_id, + "battery_level": lock.battery_level, + "status": status, + "needs_replacement": lock.battery_level < 20, + } + + def simulate_tamper(self, lock_id: str) -> dict[str, Any]: + """Simulate a tamper alert (for testing).""" + lock = self._locks.get(lock_id) + if not lock: + return {"success": False, "error": f"Lock {lock_id} not found"} + + lock.tamper_alert = True + alert = self._create_alert( + lock_id=lock_id, + alert_type="tamper_detected", + severity="critical", + message=f"Tamper detected on {lock.name}!", + ) + + return { + "success": True, + "alert": alert.to_dict(), + } + + def clear_tamper_alert(self, lock_id: str) -> dict[str, Any]: + """Clear a tamper alert.""" + lock = self._locks.get(lock_id) + if not lock: + return {"success": False, "error": f"Lock {lock_id} not found"} + + lock.tamper_alert = False + return {"success": True, "message": "Tamper alert cleared"} + + def get_alerts( + self, unacknowledged_only: bool = False, limit: int = 50 + ) -> list[dict[str, Any]]: + """Get security alerts.""" + alerts = self._alerts + if unacknowledged_only: + alerts = [a for a in alerts if not a.acknowledged] + alerts.sort(key=lambda x: x.timestamp, reverse=True) + return [a.to_dict() for a in alerts[:limit]] + + def acknowledge_alert(self, alert_id: str, acknowledged_by: str) -> dict[str, Any]: + """Acknowledge a security alert.""" + for alert in self._alerts: + if alert.id == alert_id: + alert.acknowledged = True + alert.acknowledged_by = acknowledged_by + alert.acknowledged_at = datetime.now() + return {"success": True, "alert_id": alert_id} + return {"success": False, "error": f"Alert {alert_id} not found"} diff --git a/src/mcp_servers/home_security/models.py b/src/mcp_servers/home_security/models.py new file mode 100644 index 0000000..65c663f --- /dev/null +++ b/src/mcp_servers/home_security/models.py @@ -0,0 +1,264 @@ +"""Data models for the Home Security MCP Server.""" + +from dataclasses import dataclass, field +from datetime import datetime +from enum import Enum +from typing import Any + + +class CameraStatus(Enum): + """Camera operational status.""" + + ONLINE = "online" + OFFLINE = "offline" + RECORDING = "recording" + STREAMING = "streaming" + ERROR = "error" + + +class LockStatus(Enum): + """Smart lock status.""" + + LOCKED = "locked" + UNLOCKED = "unlocked" + JAMMED = "jammed" + UNKNOWN = "unknown" + + +class MotionSensitivity(Enum): + """Motion detection sensitivity levels.""" + + LOW = "low" + MEDIUM = "medium" + HIGH = "high" + + +class AccessMethod(Enum): + """How the lock was accessed.""" + + PIN_CODE = "pin_code" + FINGERPRINT = "fingerprint" + RFID_CARD = "rfid_card" + MOBILE_APP = "mobile_app" + PHYSICAL_KEY = "physical_key" + AUTO_LOCK = "auto_lock" + REMOTE = "remote" + + +@dataclass +class Camera: + """Represents a security camera.""" + + id: str + name: str + location: str + status: CameraStatus = CameraStatus.OFFLINE + resolution: str = "1080p" + night_vision: bool = True + motion_detection: bool = True + motion_sensitivity: MotionSensitivity = MotionSensitivity.MEDIUM + recording: bool = False + streaming_url: str | None = None + last_motion_detected: datetime | None = None + storage_used_gb: float = 0.0 + storage_limit_gb: float = 100.0 + firmware_version: str = "1.0.0" + ip_address: str | None = None + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary representation.""" + return { + "id": self.id, + "name": self.name, + "location": self.location, + "status": self.status.value, + "resolution": self.resolution, + "night_vision": self.night_vision, + "motion_detection": self.motion_detection, + "motion_sensitivity": self.motion_sensitivity.value, + "recording": self.recording, + "streaming_url": self.streaming_url, + "last_motion_detected": ( + self.last_motion_detected.isoformat() if self.last_motion_detected else None + ), + "storage_used_gb": self.storage_used_gb, + "storage_limit_gb": self.storage_limit_gb, + "firmware_version": self.firmware_version, + "ip_address": self.ip_address, + "metadata": self.metadata, + } + + +@dataclass +class MotionEvent: + """Represents a motion detection event.""" + + id: str + camera_id: str + timestamp: datetime + duration_seconds: float + confidence: float + zone: str | None = None + thumbnail_url: str | None = None + video_clip_url: str | None = None + acknowledged: bool = False + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary representation.""" + return { + "id": self.id, + "camera_id": self.camera_id, + "timestamp": self.timestamp.isoformat(), + "duration_seconds": self.duration_seconds, + "confidence": self.confidence, + "zone": self.zone, + "thumbnail_url": self.thumbnail_url, + "video_clip_url": self.video_clip_url, + "acknowledged": self.acknowledged, + "metadata": self.metadata, + } + + +@dataclass +class SmartLock: + """Represents a smart door lock.""" + + id: str + name: str + location: str + status: LockStatus = LockStatus.UNKNOWN + battery_level: int = 100 + auto_lock_enabled: bool = True + auto_lock_delay_seconds: int = 30 + tamper_alert: bool = False + firmware_version: str = "1.0.0" + last_activity: datetime | None = None + connected: bool = False + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary representation.""" + return { + "id": self.id, + "name": self.name, + "location": self.location, + "status": self.status.value, + "battery_level": self.battery_level, + "auto_lock_enabled": self.auto_lock_enabled, + "auto_lock_delay_seconds": self.auto_lock_delay_seconds, + "tamper_alert": self.tamper_alert, + "firmware_version": self.firmware_version, + "last_activity": self.last_activity.isoformat() if self.last_activity else None, + "connected": self.connected, + "metadata": self.metadata, + } + + +@dataclass +class AccessCode: + """Represents a PIN access code for a smart lock.""" + + id: str + lock_id: str + name: str + code_hash: str + active: bool = True + one_time: bool = False + valid_from: datetime | None = None + valid_until: datetime | None = None + allowed_days: list[str] = field(default_factory=list) + allowed_hours_start: int | None = None + allowed_hours_end: int | None = None + use_count: int = 0 + max_uses: int | None = None + created_at: datetime = field(default_factory=datetime.now) + last_used: datetime | None = None + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary representation.""" + return { + "id": self.id, + "lock_id": self.lock_id, + "name": self.name, + "active": self.active, + "one_time": self.one_time, + "valid_from": self.valid_from.isoformat() if self.valid_from else None, + "valid_until": self.valid_until.isoformat() if self.valid_until else None, + "allowed_days": self.allowed_days, + "allowed_hours_start": self.allowed_hours_start, + "allowed_hours_end": self.allowed_hours_end, + "use_count": self.use_count, + "max_uses": self.max_uses, + "created_at": self.created_at.isoformat(), + "last_used": self.last_used.isoformat() if self.last_used else None, + } + + +@dataclass +class LockAccessLog: + """Represents an access log entry for a smart lock.""" + + id: str + lock_id: str + timestamp: datetime + action: str + method: AccessMethod + user_name: str | None = None + access_code_id: str | None = None + success: bool = True + failure_reason: str | None = None + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary representation.""" + return { + "id": self.id, + "lock_id": self.lock_id, + "timestamp": self.timestamp.isoformat(), + "action": self.action, + "method": self.method.value, + "user_name": self.user_name, + "access_code_id": self.access_code_id, + "success": self.success, + "failure_reason": self.failure_reason, + "metadata": self.metadata, + } + + +@dataclass +class SecurityAlert: + """Represents a security alert from any device.""" + + id: str + device_type: str + device_id: str + alert_type: str + severity: str + message: str + timestamp: datetime + acknowledged: bool = False + acknowledged_by: str | None = None + acknowledged_at: datetime | None = None + resolved: bool = False + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary representation.""" + return { + "id": self.id, + "device_type": self.device_type, + "device_id": self.device_id, + "alert_type": self.alert_type, + "severity": self.severity, + "message": self.message, + "timestamp": self.timestamp.isoformat(), + "acknowledged": self.acknowledged, + "acknowledged_by": self.acknowledged_by, + "acknowledged_at": ( + self.acknowledged_at.isoformat() if self.acknowledged_at else None + ), + "resolved": self.resolved, + "metadata": self.metadata, + } diff --git a/src/mcp_servers/home_security/server.py b/src/mcp_servers/home_security/server.py new file mode 100644 index 0000000..b304866 --- /dev/null +++ b/src/mcp_servers/home_security/server.py @@ -0,0 +1,646 @@ +""" +Home Security MCP Server + +An MCP (Model Context Protocol) server for managing home security devices +including cameras and smart locks. This server exposes tools and resources +that can be consumed by LLMs and AI agents. + +Usage: + # Run with stdio transport (for local LLM clients) + python -m src.mcp_servers.home_security.server + + # Run with SSE transport (for remote access) + python -m src.mcp_servers.home_security.server --transport sse --port 8000 +""" + +from datetime import datetime, timedelta +from typing import Any + +from mcp.server.fastmcp import FastMCP + +from .camera import CameraManager +from .lock import SmartLockManager + +mcp = FastMCP( + "Home Security", + instructions="MCP server for managing home security cameras and smart locks", +) + +camera_manager = CameraManager() +lock_manager = SmartLockManager() + + +# ============================================================================= +# RESOURCES - Read-only data endpoints +# ============================================================================= + + +@mcp.resource("security://cameras") +def list_all_cameras() -> list[dict[str, Any]]: + """List all registered security cameras with their current status.""" + return camera_manager.list_cameras() + + +@mcp.resource("security://cameras/{camera_id}") +def get_camera_details(camera_id: str) -> dict[str, Any]: + """Get detailed information about a specific camera.""" + result = camera_manager.get_camera(camera_id) + if result is None: + return {"error": f"Camera {camera_id} not found"} + return result + + +@mcp.resource("security://cameras/{camera_id}/status") +def get_camera_status_resource(camera_id: str) -> dict[str, Any]: + """Get the current operational status of a camera.""" + return camera_manager.get_camera_status(camera_id) + + +@mcp.resource("security://cameras/{camera_id}/storage") +def get_camera_storage(camera_id: str) -> dict[str, Any]: + """Get storage usage information for a camera.""" + return camera_manager.get_storage_info(camera_id) + + +@mcp.resource("security://locks") +def list_all_locks() -> list[dict[str, Any]]: + """List all registered smart locks with their current status.""" + return lock_manager.list_locks() + + +@mcp.resource("security://locks/{lock_id}") +def get_lock_details(lock_id: str) -> dict[str, Any]: + """Get detailed information about a specific lock.""" + result = lock_manager.get_lock(lock_id) + if result is None: + return {"error": f"Lock {lock_id} not found"} + return result + + +@mcp.resource("security://locks/{lock_id}/status") +def get_lock_status_resource(lock_id: str) -> dict[str, Any]: + """Get the current status of a lock.""" + return lock_manager.get_lock_status(lock_id) + + +@mcp.resource("security://locks/{lock_id}/battery") +def get_lock_battery(lock_id: str) -> dict[str, Any]: + """Get battery status for a lock.""" + return lock_manager.get_battery_status(lock_id) + + +@mcp.resource("security://alerts") +def get_all_alerts() -> dict[str, Any]: + """Get all security alerts from all devices.""" + camera_alerts = camera_manager.get_alerts() + lock_alerts = lock_manager.get_alerts() + all_alerts = camera_alerts + lock_alerts + all_alerts.sort(key=lambda x: x["timestamp"], reverse=True) + return { + "total": len(all_alerts), + "alerts": all_alerts[:100], + } + + +@mcp.resource("security://system/status") +def get_system_status() -> dict[str, Any]: + """Get overall system status including all devices and alerts.""" + cameras = camera_manager.list_cameras() + locks = lock_manager.list_locks() + camera_alerts = camera_manager.get_alerts(unacknowledged_only=True) + lock_alerts = lock_manager.get_alerts(unacknowledged_only=True) + + online_cameras = sum(1 for c in cameras if c["status"] != "offline") + locked_doors = sum(1 for lock in locks if lock["status"] == "locked") + low_battery_locks = [lock for lock in locks if lock["battery_level"] < 20] + + return { + "timestamp": datetime.now().isoformat(), + "cameras": { + "total": len(cameras), + "online": online_cameras, + "offline": len(cameras) - online_cameras, + }, + "locks": { + "total": len(locks), + "locked": locked_doors, + "unlocked": len(locks) - locked_doors, + "low_battery": [lock["id"] for lock in low_battery_locks], + }, + "alerts": { + "unacknowledged": len(camera_alerts) + len(lock_alerts), + }, + "overall_status": "secure" if locked_doors == len(locks) else "attention_needed", + } + + +# ============================================================================= +# CAMERA TOOLS - Actions for camera management +# ============================================================================= + + +@mcp.tool() +def get_camera_status(camera_id: str) -> dict[str, Any]: + """ + Get the current status of a security camera. + + Args: + camera_id: The unique identifier of the camera (e.g., 'cam_front_door') + + Returns: + Camera status including online state, recording status, and motion detection. + """ + return camera_manager.get_camera_status(camera_id) + + +@mcp.tool() +def start_camera_recording(camera_id: str) -> dict[str, Any]: + """ + Start recording on a security camera. + + Args: + camera_id: The unique identifier of the camera + + Returns: + Result indicating success or failure of the recording start. + """ + return camera_manager.start_recording(camera_id) + + +@mcp.tool() +def stop_camera_recording(camera_id: str) -> dict[str, Any]: + """ + Stop recording on a security camera. + + Args: + camera_id: The unique identifier of the camera + + Returns: + Result indicating success or failure of the recording stop. + """ + return camera_manager.stop_recording(camera_id) + + +@mcp.tool() +def get_camera_stream_url(camera_id: str) -> dict[str, Any]: + """ + Get the live streaming URL for a camera. + + Args: + camera_id: The unique identifier of the camera + + Returns: + The RTSP streaming URL for the camera. + """ + return camera_manager.get_stream_url(camera_id) + + +@mcp.tool() +def capture_camera_snapshot(camera_id: str) -> dict[str, Any]: + """ + Capture a snapshot image from a camera. + + Args: + camera_id: The unique identifier of the camera + + Returns: + URL of the captured snapshot and metadata. + """ + return camera_manager.capture_snapshot(camera_id) + + +@mcp.tool() +def configure_motion_detection( + camera_id: str, + enabled: bool, + sensitivity: str | None = None, +) -> dict[str, Any]: + """ + Configure motion detection settings for a camera. + + Args: + camera_id: The unique identifier of the camera + enabled: Whether to enable motion detection + sensitivity: Detection sensitivity - 'low', 'medium', or 'high' + + Returns: + Updated motion detection configuration. + """ + return camera_manager.set_motion_detection(camera_id, enabled, sensitivity) + + +@mcp.tool() +def get_motion_events( + camera_id: str | None = None, + hours_back: int = 24, + limit: int = 50, +) -> list[dict[str, Any]]: + """ + Get recent motion detection events. + + Args: + camera_id: Optional camera ID to filter events + hours_back: Number of hours to look back (default 24) + limit: Maximum number of events to return (default 50) + + Returns: + List of motion events with timestamps and details. + """ + since = datetime.now() - timedelta(hours=hours_back) + return camera_manager.get_motion_events(camera_id, since, limit) + + +@mcp.tool() +def acknowledge_motion_event(event_id: str) -> dict[str, Any]: + """ + Acknowledge a motion detection event. + + Args: + event_id: The unique identifier of the motion event + + Returns: + Result indicating success or failure. + """ + return camera_manager.acknowledge_motion_event(event_id) + + +@mcp.tool() +def get_camera_recordings( + camera_id: str, + hours_back: int = 24, +) -> list[dict[str, Any]]: + """ + Get list of recordings for a camera. + + Args: + camera_id: The unique identifier of the camera + hours_back: Number of hours to look back (default 24) + + Returns: + List of available recordings with URLs. + """ + start_time = datetime.now() - timedelta(hours=hours_back) + return camera_manager.get_recordings(camera_id, start_time) + + +@mcp.tool() +def update_camera_settings( + camera_id: str, + resolution: str | None = None, + night_vision: bool | None = None, +) -> dict[str, Any]: + """ + Update camera settings. + + Args: + camera_id: The unique identifier of the camera + resolution: Video resolution (e.g., '1080p', '4K') + night_vision: Enable or disable night vision + + Returns: + Updated camera settings. + """ + return camera_manager.set_camera_settings(camera_id, resolution, night_vision) + + +# ============================================================================= +# LOCK TOOLS - Actions for smart lock management +# ============================================================================= + + +@mcp.tool() +def get_lock_status(lock_id: str) -> dict[str, Any]: + """ + Get the current status of a smart lock. + + Args: + lock_id: The unique identifier of the lock (e.g., 'lock_front_door') + + Returns: + Lock status including locked/unlocked state, battery level, and connectivity. + """ + return lock_manager.get_lock_status(lock_id) + + +@mcp.tool() +def lock_door(lock_id: str, user: str = "assistant") -> dict[str, Any]: + """ + Lock a smart door lock. + + Args: + lock_id: The unique identifier of the lock + user: Name of the user/agent performing the action + + Returns: + Result indicating success or failure of the lock operation. + """ + return lock_manager.lock(lock_id, user) + + +@mcp.tool() +def unlock_door( + lock_id: str, + code: str | None = None, + user: str = "assistant", +) -> dict[str, Any]: + """ + Unlock a smart door lock. + + SECURITY NOTE: This action unlocks a physical door. Use with caution. + + Args: + lock_id: The unique identifier of the lock + code: Optional PIN code for verification + user: Name of the user/agent performing the action + + Returns: + Result indicating success or failure, plus auto-lock timing if enabled. + """ + return lock_manager.unlock(lock_id, code, user) + + +@mcp.tool() +def add_lock_access_code( + lock_id: str, + name: str, + code: str, + one_time: bool = False, + valid_days: int | None = None, +) -> dict[str, Any]: + """ + Add a new access code to a smart lock. + + Args: + lock_id: The unique identifier of the lock + name: Friendly name for this code (e.g., 'House Cleaner') + code: The PIN code (minimum 4 digits) + one_time: Whether the code can only be used once + valid_days: Number of days the code is valid (None = permanent) + + Returns: + The created access code details (code ID, not the actual code). + """ + return lock_manager.add_access_code(lock_id, name, code, one_time, valid_days) + + +@mcp.tool() +def remove_lock_access_code(lock_id: str, code_id: str) -> dict[str, Any]: + """ + Remove an access code from a smart lock. + + Args: + lock_id: The unique identifier of the lock + code_id: The unique identifier of the access code to remove + + Returns: + Result indicating success or failure. + """ + return lock_manager.remove_access_code(lock_id, code_id) + + +@mcp.tool() +def list_lock_access_codes(lock_id: str) -> list[dict[str, Any]]: + """ + List all access codes for a smart lock. + + Note: Actual code values are not returned for security. + + Args: + lock_id: The unique identifier of the lock + + Returns: + List of access codes with metadata (names, validity, usage). + """ + return lock_manager.list_access_codes(lock_id) + + +@mcp.tool() +def deactivate_lock_access_code(lock_id: str, code_id: str) -> dict[str, Any]: + """ + Deactivate an access code without removing it. + + Args: + lock_id: The unique identifier of the lock + code_id: The unique identifier of the access code + + Returns: + Result indicating success or failure. + """ + return lock_manager.deactivate_access_code(lock_id, code_id) + + +@mcp.tool() +def get_lock_access_logs( + lock_id: str, + hours_back: int = 24, + limit: int = 50, +) -> list[dict[str, Any]]: + """ + Get access logs for a smart lock. + + Args: + lock_id: The unique identifier of the lock + hours_back: Number of hours to look back (default 24) + limit: Maximum number of log entries to return + + Returns: + List of access log entries with timestamps, methods, and outcomes. + """ + since = datetime.now() - timedelta(hours=hours_back) + return lock_manager.get_access_logs(lock_id, since, limit) + + +@mcp.tool() +def configure_auto_lock( + lock_id: str, + enabled: bool, + delay_seconds: int | None = None, +) -> dict[str, Any]: + """ + Configure auto-lock settings for a smart lock. + + Args: + lock_id: The unique identifier of the lock + enabled: Whether to enable auto-lock + delay_seconds: Seconds to wait before auto-locking (default 30) + + Returns: + Updated auto-lock configuration. + """ + return lock_manager.set_auto_lock(lock_id, enabled, delay_seconds) + + +@mcp.tool() +def clear_lock_tamper_alert(lock_id: str) -> dict[str, Any]: + """ + Clear a tamper alert on a smart lock after investigation. + + Args: + lock_id: The unique identifier of the lock + + Returns: + Result indicating success or failure. + """ + return lock_manager.clear_tamper_alert(lock_id) + + +# ============================================================================= +# ALERT TOOLS - Actions for security alerts +# ============================================================================= + + +@mcp.tool() +def get_security_alerts( + unacknowledged_only: bool = False, + limit: int = 50, +) -> dict[str, Any]: + """ + Get security alerts from all devices. + + Args: + unacknowledged_only: Only return alerts that haven't been acknowledged + limit: Maximum number of alerts to return + + Returns: + List of security alerts sorted by timestamp. + """ + camera_alerts = camera_manager.get_alerts(unacknowledged_only, limit) + lock_alerts = lock_manager.get_alerts(unacknowledged_only, limit) + all_alerts = camera_alerts + lock_alerts + all_alerts.sort(key=lambda x: x["timestamp"], reverse=True) + return { + "total": len(all_alerts), + "alerts": all_alerts[:limit], + } + + +@mcp.tool() +def acknowledge_security_alert( + alert_id: str, + acknowledged_by: str = "assistant", +) -> dict[str, Any]: + """ + Acknowledge a security alert. + + Args: + alert_id: The unique identifier of the alert + acknowledged_by: Name of the user/agent acknowledging + + Returns: + Result indicating success or failure. + """ + result = camera_manager.acknowledge_alert(alert_id, acknowledged_by) + if result.get("success"): + return result + return lock_manager.acknowledge_alert(alert_id, acknowledged_by) + + +# ============================================================================= +# TESTING TOOLS - Simulation tools for development +# ============================================================================= + + +@mcp.tool() +def simulate_motion_detected(camera_id: str, zone: str = "main") -> dict[str, Any]: + """ + [TEST ONLY] Simulate a motion detection event. + + Args: + camera_id: The camera to simulate motion on + zone: The detection zone name + + Returns: + The generated motion event and alert. + """ + return camera_manager.simulate_motion_event(camera_id, zone) + + +@mcp.tool() +def simulate_tamper_alert(lock_id: str) -> dict[str, Any]: + """ + [TEST ONLY] Simulate a tamper detection on a lock. + + Args: + lock_id: The lock to simulate tampering on + + Returns: + The generated tamper alert. + """ + return lock_manager.simulate_tamper(lock_id) + + +# ============================================================================= +# PROMPTS - Pre-defined prompts for common tasks +# ============================================================================= + + +@mcp.prompt() +def security_check() -> str: + """Generate a prompt for performing a full security check.""" + return """Please perform a comprehensive security check of the home: + +1. Check the status of all cameras - are they online and recording? +2. Check if all doors are locked +3. Review any recent motion events +4. Check for any unacknowledged security alerts +5. Verify battery levels on all smart locks + +Provide a summary of the security status and flag any concerns.""" + + +@mcp.prompt() +def prepare_for_vacation() -> str: + """Generate a prompt for vacation preparation.""" + return """Help me prepare the home security system for vacation: + +1. Ensure all cameras have motion detection enabled with HIGH sensitivity +2. Start recording on all cameras +3. Lock all doors +4. Enable auto-lock on all smart locks +5. List current access codes - disable any temporary ones +6. Provide a final security status report""" + + +@mcp.prompt() +def guest_access_setup(guest_name: str, days_valid: int = 7) -> str: + """Generate a prompt for setting up guest access.""" + return f"""Help me set up temporary access for a guest: + +Guest name: {guest_name} +Access duration: {days_valid} days + +Please: +1. Create a temporary access code for the front door +2. Make sure the code expires after {days_valid} days +3. Confirm the current lock status +4. Provide instructions I can share with the guest""" + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="Home Security MCP Server") + parser.add_argument( + "--transport", + choices=["stdio", "sse"], + default="stdio", + help="Transport method (default: stdio)", + ) + parser.add_argument( + "--host", + default="0.0.0.0", + help="Host for SSE transport (default: 0.0.0.0)", + ) + parser.add_argument( + "--port", + type=int, + default=8000, + help="Port for SSE transport (default: 8000)", + ) + + args = parser.parse_args() + + if args.transport == "sse": + mcp.run(transport="sse", host=args.host, port=args.port) + else: + mcp.run() diff --git a/tests/test_home_security_mcp.py b/tests/test_home_security_mcp.py new file mode 100644 index 0000000..db8e7b9 --- /dev/null +++ b/tests/test_home_security_mcp.py @@ -0,0 +1,352 @@ +"""Tests for the Home Security MCP Server.""" + +from datetime import datetime, timedelta + +import pytest + +from src.mcp_servers.home_security.camera import CameraManager +from src.mcp_servers.home_security.lock import SmartLockManager +from src.mcp_servers.home_security.models import ( + AccessMethod, + Camera, + CameraStatus, + LockStatus, + MotionSensitivity, + SmartLock, +) + + +class TestCameraManager: + """Tests for CameraManager class.""" + + @pytest.fixture + def manager(self) -> CameraManager: + """Create a camera manager instance for testing.""" + return CameraManager() + + def test_list_cameras(self, manager: CameraManager) -> None: + """Test listing all cameras.""" + cameras = manager.list_cameras() + assert len(cameras) == 4 + assert all(isinstance(c, dict) for c in cameras) + assert any(c["id"] == "cam_front_door" for c in cameras) + + def test_get_camera(self, manager: CameraManager) -> None: + """Test getting a specific camera.""" + camera = manager.get_camera("cam_front_door") + assert camera is not None + assert camera["name"] == "Front Door Camera" + assert camera["resolution"] == "4K" + + def test_get_camera_not_found(self, manager: CameraManager) -> None: + """Test getting a non-existent camera.""" + camera = manager.get_camera("nonexistent") + assert camera is None + + def test_get_camera_status(self, manager: CameraManager) -> None: + """Test getting camera status.""" + status = manager.get_camera_status("cam_front_door") + assert status["camera_id"] == "cam_front_door" + assert "status" in status + assert "recording" in status + + def test_start_stop_recording(self, manager: CameraManager) -> None: + """Test starting and stopping recording.""" + result = manager.start_recording("cam_front_door") + assert result["success"] is True + assert "Recording started" in result["message"] + + status = manager.get_camera_status("cam_front_door") + assert status["recording"] is True + + result = manager.stop_recording("cam_front_door") + assert result["success"] is True + status = manager.get_camera_status("cam_front_door") + assert status["recording"] is False + + def test_get_stream_url(self, manager: CameraManager) -> None: + """Test getting stream URL.""" + result = manager.get_stream_url("cam_front_door") + assert "stream_url" in result + assert "rtsp://" in result["stream_url"] + + def test_capture_snapshot(self, manager: CameraManager) -> None: + """Test capturing snapshot.""" + result = manager.capture_snapshot("cam_front_door") + assert result["success"] is True + assert "snapshot_url" in result + assert result["resolution"] == "4K" + + def test_set_motion_detection(self, manager: CameraManager) -> None: + """Test configuring motion detection.""" + result = manager.set_motion_detection("cam_front_door", True, "high") + assert result["success"] is True + assert result["motion_detection"] is True + assert result["sensitivity"] == "high" + + result = manager.set_motion_detection("cam_front_door", False) + assert result["motion_detection"] is False + + def test_simulate_motion_event(self, manager: CameraManager) -> None: + """Test simulating motion event.""" + manager.set_motion_detection("cam_front_door", True) + result = manager.simulate_motion_event("cam_front_door", "entrance") + assert result["success"] is True + assert "event" in result + assert "alert" in result + assert result["event"]["zone"] == "entrance" + + def test_get_motion_events(self, manager: CameraManager) -> None: + """Test getting motion events.""" + manager.set_motion_detection("cam_front_door", True) + manager.simulate_motion_event("cam_front_door") + events = manager.get_motion_events("cam_front_door") + assert len(events) >= 1 + + def test_acknowledge_motion_event(self, manager: CameraManager) -> None: + """Test acknowledging motion event.""" + manager.set_motion_detection("cam_front_door", True) + result = manager.simulate_motion_event("cam_front_door") + event_id = result["event"]["id"] + + ack_result = manager.acknowledge_motion_event(event_id) + assert ack_result["success"] is True + + def test_get_recordings(self, manager: CameraManager) -> None: + """Test getting recordings.""" + recordings = manager.get_recordings("cam_front_door") + assert isinstance(recordings, list) + if recordings: + assert "url" in recordings[0] + assert "duration_seconds" in recordings[0] + + def test_get_storage_info(self, manager: CameraManager) -> None: + """Test getting storage info.""" + info = manager.get_storage_info("cam_front_door") + assert "storage_used_gb" in info + assert "storage_limit_gb" in info + assert "usage_percent" in info + + def test_set_camera_settings(self, manager: CameraManager) -> None: + """Test updating camera settings.""" + result = manager.set_camera_settings("cam_front_door", resolution="1080p", night_vision=False) + assert result["success"] is True + assert result["settings"]["resolution"] == "1080p" + assert result["settings"]["night_vision"] is False + + +class TestSmartLockManager: + """Tests for SmartLockManager class.""" + + @pytest.fixture + def manager(self) -> SmartLockManager: + """Create a lock manager instance for testing.""" + return SmartLockManager() + + def test_list_locks(self, manager: SmartLockManager) -> None: + """Test listing all locks.""" + locks = manager.list_locks() + assert len(locks) == 1 + assert locks[0]["id"] == "lock_front_door" + + def test_get_lock(self, manager: SmartLockManager) -> None: + """Test getting a specific lock.""" + lock = manager.get_lock("lock_front_door") + assert lock is not None + assert lock["name"] == "Front Door Lock" + + def test_get_lock_not_found(self, manager: SmartLockManager) -> None: + """Test getting a non-existent lock.""" + lock = manager.get_lock("nonexistent") + assert lock is None + + def test_get_lock_status(self, manager: SmartLockManager) -> None: + """Test getting lock status.""" + status = manager.get_lock_status("lock_front_door") + assert status["lock_id"] == "lock_front_door" + assert "status" in status + assert "battery_level" in status + + def test_lock_unlock(self, manager: SmartLockManager) -> None: + """Test locking and unlocking.""" + unlock_result = manager.unlock("lock_front_door", user="test") + assert unlock_result["success"] is True + assert unlock_result["status"] == "unlocked" + + lock_result = manager.lock("lock_front_door", user="test") + assert lock_result["success"] is True + assert lock_result["status"] == "locked" + + def test_unlock_with_code(self, manager: SmartLockManager) -> None: + """Test unlocking with PIN code.""" + result = manager.unlock("lock_front_door", code="1234") + assert result["success"] is True + assert result["status"] == "unlocked" + + def test_unlock_with_invalid_code(self, manager: SmartLockManager) -> None: + """Test unlocking with invalid PIN code.""" + result = manager.unlock("lock_front_door", code="9999") + assert result["success"] is False + assert "Invalid" in result["error"] + + def test_add_access_code(self, manager: SmartLockManager) -> None: + """Test adding an access code.""" + result = manager.add_access_code( + "lock_front_door", + name="Test Code", + code="4567", + valid_days=7, + ) + assert result["success"] is True + assert "code_id" in result + + unlock_result = manager.unlock("lock_front_door", code="4567") + assert unlock_result["success"] is True + + def test_add_one_time_code(self, manager: SmartLockManager) -> None: + """Test adding a one-time access code.""" + result = manager.add_access_code( + "lock_front_door", + name="One Time", + code="1111", + one_time=True, + ) + assert result["success"] is True + + unlock1 = manager.unlock("lock_front_door", code="1111") + assert unlock1["success"] is True + + manager.lock("lock_front_door") + unlock2 = manager.unlock("lock_front_door", code="1111") + assert unlock2["success"] is False + + def test_remove_access_code(self, manager: SmartLockManager) -> None: + """Test removing an access code.""" + add_result = manager.add_access_code( + "lock_front_door", + name="Temp Code", + code="2222", + ) + code_id = add_result["code_id"] + + remove_result = manager.remove_access_code("lock_front_door", code_id) + assert remove_result["success"] is True + + unlock_result = manager.unlock("lock_front_door", code="2222") + assert unlock_result["success"] is False + + def test_list_access_codes(self, manager: SmartLockManager) -> None: + """Test listing access codes.""" + codes = manager.list_access_codes("lock_front_door") + assert len(codes) >= 2 + assert all("name" in c for c in codes) + assert all("code_hash" not in c or "code" not in c for c in codes) + + def test_deactivate_access_code(self, manager: SmartLockManager) -> None: + """Test deactivating an access code.""" + add_result = manager.add_access_code( + "lock_front_door", + name="Deactivate Test", + code="3333", + ) + code_id = add_result["code_id"] + + deactivate_result = manager.deactivate_access_code("lock_front_door", code_id) + assert deactivate_result["success"] is True + + unlock_result = manager.unlock("lock_front_door", code="3333") + assert unlock_result["success"] is False + + def test_get_access_logs(self, manager: SmartLockManager) -> None: + """Test getting access logs.""" + manager.unlock("lock_front_door", user="test") + manager.lock("lock_front_door", user="test") + + logs = manager.get_access_logs("lock_front_door") + assert len(logs) >= 2 + assert all("timestamp" in l for l in logs) + assert all("action" in l for l in logs) + + def test_set_auto_lock(self, manager: SmartLockManager) -> None: + """Test configuring auto-lock.""" + result = manager.set_auto_lock("lock_front_door", enabled=True, delay_seconds=60) + assert result["success"] is True + assert result["auto_lock_enabled"] is True + assert result["auto_lock_delay_seconds"] == 60 + + def test_get_battery_status(self, manager: SmartLockManager) -> None: + """Test getting battery status.""" + status = manager.get_battery_status("lock_front_door") + assert "battery_level" in status + assert "status" in status + assert status["status"] in ["good", "low", "critical"] + + def test_simulate_tamper(self, manager: SmartLockManager) -> None: + """Test simulating tamper alert.""" + result = manager.simulate_tamper("lock_front_door") + assert result["success"] is True + assert "alert" in result + assert result["alert"]["alert_type"] == "tamper_detected" + + def test_clear_tamper_alert(self, manager: SmartLockManager) -> None: + """Test clearing tamper alert.""" + manager.simulate_tamper("lock_front_door") + result = manager.clear_tamper_alert("lock_front_door") + assert result["success"] is True + + status = manager.get_lock_status("lock_front_door") + assert status["tamper_alert"] is False + + def test_get_alerts(self, manager: SmartLockManager) -> None: + """Test getting alerts.""" + manager.simulate_tamper("lock_front_door") + alerts = manager.get_alerts() + assert len(alerts) >= 1 + + def test_acknowledge_alert(self, manager: SmartLockManager) -> None: + """Test acknowledging an alert.""" + tamper_result = manager.simulate_tamper("lock_front_door") + alert_id = tamper_result["alert"]["id"] + + ack_result = manager.acknowledge_alert(alert_id, "test_user") + assert ack_result["success"] is True + + +class TestModels: + """Tests for data models.""" + + def test_camera_to_dict(self) -> None: + """Test Camera serialization.""" + camera = Camera( + id="test_cam", + name="Test Camera", + location="Test Location", + status=CameraStatus.ONLINE, + ) + data = camera.to_dict() + assert data["id"] == "test_cam" + assert data["status"] == "online" + + def test_smart_lock_to_dict(self) -> None: + """Test SmartLock serialization.""" + lock = SmartLock( + id="test_lock", + name="Test Lock", + location="Test Door", + status=LockStatus.LOCKED, + ) + data = lock.to_dict() + assert data["id"] == "test_lock" + assert data["status"] == "locked" + + def test_motion_sensitivity_values(self) -> None: + """Test MotionSensitivity enum values.""" + assert MotionSensitivity.LOW.value == "low" + assert MotionSensitivity.MEDIUM.value == "medium" + assert MotionSensitivity.HIGH.value == "high" + + def test_access_method_values(self) -> None: + """Test AccessMethod enum values.""" + assert AccessMethod.PIN_CODE.value == "pin_code" + assert AccessMethod.FINGERPRINT.value == "fingerprint" + assert AccessMethod.MOBILE_APP.value == "mobile_app"