""" MCP Camera Tool Server — JSON-RPC style interface for camera control. Exposes camera management, VQA, alert rules, and alert history as callable tools that can be consumed by any MCP-compatible client. Tools: - list_cameras: Enumerate all registered cameras and their status - get_camera_status: Detailed health info for a single camera - get_latest_frame: Retrieve the most recent frame as base64 JPEG - ask_camera: Visual question answering on the latest frame - set_alert_rule: Register an alert rule on a camera - get_alert_history: Retrieve recent alert events for a camera """ from __future__ import annotations import base64 import logging import time from typing import Any, Callable logger = logging.getLogger(__name__) class CameraToolServer: """Camera control tools exposed via MCP protocol. Implements a JSON-RPC style request handler that routes method calls to individual tool functions. Designed to work with any camera manager (edge.ingest.CameraManager) and model manager for VQA inference. Args: camera_manager: CameraManager instance (from edge/ingest.py). model_manager: Object with an ``ask(image, question)`` method for VQA. """ def __init__(self, camera_manager=None, model_manager=None): self.cameras = camera_manager self.model = model_manager # Alert rules stored per camera: {camera_id: [rule_dicts]} self._alert_rules: dict[str, list[dict[str, Any]]] = {} # Alert event history: list of alert dicts (most recent last) self._alert_history: list[dict[str, Any]] = [] # Tool registry self.tools: dict[str, Callable] = { "list_cameras": self.list_cameras, "get_camera_status": self.get_camera_status, "get_latest_frame": self.get_latest_frame, "ask_camera": self.ask_camera, "set_alert_rule": self.set_alert_rule, "get_alert_history": self.get_alert_history, } # ------------------------------------------------------------------ # Tool descriptors (for MCP tool listing) # ------------------------------------------------------------------ def get_tool_definitions(self) -> list[dict[str, Any]]: """Return MCP-compatible tool definitions.""" return [ { "name": "list_cameras", "description": "List all registered cameras and their current status.", "parameters": {}, }, { "name": "get_camera_status", "description": "Get detailed status for a specific camera.", "parameters": { "camera_id": {"type": "string", "description": "Camera identifier."}, }, }, { "name": "get_latest_frame", "description": "Get the latest frame from a camera as base64 JPEG.", "parameters": { "camera_id": {"type": "string", "description": "Camera identifier."}, }, }, { "name": "ask_camera", "description": "Ask a visual question about the latest frame from a camera.", "parameters": { "camera_id": {"type": "string", "description": "Camera identifier."}, "question": {"type": "string", "description": "Natural language question."}, }, }, { "name": "set_alert_rule", "description": "Register an alert rule on a camera.", "parameters": { "camera_id": {"type": "string", "description": "Camera identifier."}, "rule": {"type": "object", "description": "Alert rule definition."}, }, }, { "name": "get_alert_history", "description": "Get recent alert events for a camera.", "parameters": { "camera_id": {"type": "string", "description": "Camera identifier."}, "limit": {"type": "integer", "description": "Max events to return.", "default": 20}, }, }, ] # ------------------------------------------------------------------ # Tools # ------------------------------------------------------------------ def list_cameras(self, **kwargs) -> dict[str, Any]: """List all registered cameras with status summary.""" if self.cameras is None: return {"cameras": [], "count": 0} camera_ids = self.cameras.camera_ids() status_map = self.cameras.status() cameras = [] for cid in camera_ids: health = status_map.get(cid, {}) cameras.append({ "camera_id": cid, "state": health.get("state", "unknown"), "fps": health.get("fps_actual", 0.0), "frames_captured": health.get("frames_captured", 0), }) return {"cameras": cameras, "count": len(cameras)} def get_camera_status(self, camera_id: str, **kwargs) -> dict[str, Any]: """Get detailed health info for a single camera.""" if self.cameras is None: return {"error": "No camera manager configured"} status_map = self.cameras.status() health = status_map.get(camera_id) if health is None: return {"error": f"Camera '{camera_id}' not found"} return { "camera_id": camera_id, **health, } def get_latest_frame(self, camera_id: str, **kwargs) -> dict[str, Any]: """Return the latest frame from a camera as base64-encoded JPEG.""" if self.cameras is None: return {"error": "No camera manager configured"} result = self.cameras.get_frame(camera_id) if result is None: return {"error": f"No frame available for camera '{camera_id}'"} frame, timestamp = result # Encode frame to JPEG then base64 try: import cv2 _, jpeg_buf = cv2.imencode(".jpg", frame) b64 = base64.b64encode(jpeg_buf.tobytes()).decode("utf-8") except ImportError: # Fallback: raw numpy bytes (unlikely in production) b64 = base64.b64encode(frame.tobytes()).decode("utf-8") return { "camera_id": camera_id, "frame_base64": b64, "format": "jpeg", "timestamp": timestamp, } def ask_camera(self, camera_id: str, question: str, **kwargs) -> dict[str, Any]: """Run visual question answering on the latest frame from a camera.""" if self.cameras is None: return {"error": "No camera manager configured"} if self.model is None: return {"error": "No model manager configured"} result = self.cameras.get_frame(camera_id) if result is None: return {"error": f"No frame available for camera '{camera_id}'"} frame, timestamp = result try: answer = self.model.ask(frame, question) except Exception as exc: logger.error("ask_camera failed for %s: %s", camera_id, exc) return {"error": f"Model inference failed: {exc}"} return { "camera_id": camera_id, "question": question, "answer": answer, "timestamp": timestamp, } def set_alert_rule(self, camera_id: str, rule: dict, **kwargs) -> dict[str, Any]: """Register an alert rule for a camera.""" rule_id = rule.get("rule_id", f"rule-{int(time.time() * 1000)}") rule["rule_id"] = rule_id if camera_id not in self._alert_rules: self._alert_rules[camera_id] = [] self._alert_rules[camera_id].append(rule) logger.info("Set alert rule '%s' on camera '%s'", rule_id, camera_id) return { "camera_id": camera_id, "rule_id": rule_id, "status": "created", "total_rules": len(self._alert_rules[camera_id]), } def get_alert_history(self, camera_id: str, limit: int = 20, **kwargs) -> dict[str, Any]: """Get recent alert events for a camera.""" events = [ e for e in self._alert_history if e.get("camera_id") == camera_id ] # Most recent first, limited events = list(reversed(events))[:limit] return { "camera_id": camera_id, "events": events, "count": len(events), } def record_alert(self, event: dict[str, Any]) -> None: """Record an alert event to history (called by alert processing pipeline).""" if "timestamp" not in event: event["timestamp"] = time.time() self._alert_history.append(event) # ------------------------------------------------------------------ # JSON-RPC style request handler # ------------------------------------------------------------------ def handle_request(self, method: str, params: dict | None = None) -> dict[str, Any]: """ JSON-RPC style request handler. Args: method: Tool name to invoke. params: Keyword arguments for the tool. Returns: Dict with either ``result`` key on success or ``error`` key on failure. """ if params is None: params = {} tool_fn = self.tools.get(method) if tool_fn is None: return { "error": { "code": -32601, "message": f"Method not found: {method}", }, } try: result = tool_fn(**params) return {"result": result} except Exception as exc: logger.exception("Error executing tool '%s'", method) return { "error": { "code": -32000, "message": str(exc), }, }