get_cameras

Instructions

Implementation Reference

• src/domoticz_mcp/server.py:900-905 (handler)

  The `get_cameras` tool handler function. It is decorated with @mcp.tool(), calls the Domoticz API endpoint with 'param=getcameras', and returns the JSON response as text.

  @mcp.tool()
  async def get_cameras() -> str:
      """Get all configured cameras in Domoticz."""
      async with create_client() as client:
          response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=getcameras")
          return response.text

• src/domoticz_mcp/server.py:900-901 (registration)

  The `@mcp.tool()` decorator on line 900 registers `get_cameras` as an MCP tool with the FastMCP server.

  @mcp.tool()
  async def get_cameras() -> str:

• src/domoticz_mcp/server.py:269-296 (helper)

  The `_do_request` helper performs HTTP requests with retry logic for expired OAuth tokens. Used by `get_cameras` (and all other tools) to communicate with the Domoticz API.

  async def _do_request(client: httpx.AsyncClient, method: str, url: str, **kwargs) -> httpx.Response:
      """Perform a request with a single retry on 401 Unauthorized to handle expired tokens."""
      global _oauth_token_cache
      
      try:
          resp = await client.request(method, url, **kwargs)
          if resp.status_code == 401:
              # Token might be expired. Clear cache and retry once.
              _oauth_token_cache = None
              
              # Re-fetch token (this will trigger OAuth flow if needed)
              new_token = await _fetch_oauth_token(force_refresh=True)
              if new_token:
                  # Update headers for the retry
                  if "headers" not in kwargs:
                      kwargs["headers"] = {}
                  kwargs["headers"]["Authorization"] = f"Bearer {new_token}"
                  
                  # Retry the request
                  resp = await client.request(method, url, **kwargs)
          
          resp.raise_for_status()
          return resp
      except httpx.HTTPStatusError as e:
          if e.response.status_code == 401:
              raise Exception("Authentication failed. Please check your credentials or re-authenticate.")
          raise e

• src/domoticz_mcp/server.py:298-336 (helper)

  The `DomoticzClient` context manager class handles authentication (OAuth/basic auth) for all API requests. Used via `create_client()` in `get_cameras`.

  class DomoticzClient:
      def __init__(self, own_client: bool = False):
          self._own_client = own_client
          if own_client or _global_http_client is None:
              self.client: httpx.AsyncClient = httpx.AsyncClient(timeout=30.0)
              self._owns_client = True
          else:
              self.client = _global_http_client
              self._owns_client = False
  
      async def __aenter__(self) -> "httpx.AsyncClient":
          oauth_token = None
  
          if DOMOTICZ_CLIENT_ID:
              oauth_token = await _fetch_oauth_token()
  
          if oauth_token:
              self.client.headers["Authorization"] = f"Bearer {oauth_token}"
          elif DOMOTICZ_USERNAME and DOMOTICZ_PASSWORD:
              self.client.auth = (DOMOTICZ_USERNAME, DOMOTICZ_PASSWORD)
          else:
              self.client.headers.pop("Authorization", None)
              self.client.auth = None
  
          return self.client
  
      async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
          if self._owns_client:
              await self.client.aclose()
  
  
  def create_client(own_client: bool = False) -> DomoticzClient:
      """Create a DomoticzClient instance.
  
      Args:
          own_client: If True, creates a dedicated client that will be closed on exit.
                      If False (default), uses a shared client for connection pooling.
      """
      return DomoticzClient(own_client=own_client)

• src/domoticz_mcp/server.py:70-327 (helper)

  The FastMCP server instance created at module level, used by the @mcp.tool() decorator to register `get_cameras`.

  mcp = FastMCP("Domoticz", transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False), stateless_http=True)
  
  # Configuration defaults
  DOMOTICZ_BASE_URL = "https://xmpp.vanadrighem.eu/domoticz"
  DOMOTICZ_API_URL = f"{DOMOTICZ_BASE_URL}/json.htm"
  DOMOTICZ_USERNAME = None
  DOMOTICZ_PASSWORD = None
  DOMOTICZ_CLIENT_ID = None
  DOMOTICZ_CLIENT_SECRET = None
  DOMOTICZ_OAUTH_TOKEN = None
  TOKEN_FILE = os.path.expanduser("~/.config/domoticz-mcp/token.json")
  _oauth_token_cache: Optional[str] = None
  _global_http_client: Optional[httpx.AsyncClient] = None
  
  # Cache settings
  CACHE_TTL = 300 # 5 minutes
  _device_cache = {"data": None, "timestamp": 0}
  _scene_cache = {"data": None, "timestamp": 0}
  _user_variable_cache = {"data": None, "timestamp": 0}
  _plans_cache = {"data": None, "timestamp": 0}
  
  
  def _format_response(data: Dict[str, Any]) -> str:
      """Format a dictionary as a JSON string response."""
      return json.dumps(data)
  
  
  def _error_response(message: str, status: str = "error") -> str:
      """Format an error response as a JSON string."""
      return json.dumps({"status": status, "message": message})
  
  def _do_interactive_oauth_flow():
  # ... (rest of the file remains the same until _resolve_device_idx)
      code = None
      state_received = None
      
      class CallbackHandler(BaseHTTPRequestHandler):
          def log_message(self, format, *args):
              pass # Suppress logging to stderr
              
          def do_GET(self):
              nonlocal code, state_received
              parsed_path = urllib.parse.urlparse(self.path)
              if parsed_path.path == '/callback':
                  query = urllib.parse.parse_qs(parsed_path.query)
                  if 'code' in query:
                      code = query['code'][0]
                  if 'state' in query:
                      state_received = query['state'][0]
                  self.send_response(200)
                  self.send_header('Content-type', 'text/html')
                  self.end_headers()
                  self.wfile.write(b"<html><body><h1>Authentication successful!</h1><p>You can close this window now and return to the application.</p></body></html>")
                  threading.Thread(target=self.server.shutdown).start()
              else:
                  self.send_response(404)
                  self.end_headers()
  
      server = HTTPServer(('127.0.0.1', 0), CallbackHandler)
      port = server.server_port
      
      state = secrets.token_urlsafe(16)
      code_verifier = secrets.token_urlsafe(32)
      code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode('ascii')).digest()).decode('ascii').rstrip('=')
      
      redirect_uri = f"http://127.0.0.1:{port}/callback"
      auth_url = f"{DOMOTICZ_BASE_URL}/oauth2/v1/authorize?response_type=code&client_id={DOMOTICZ_CLIENT_ID}&redirect_uri={urllib.parse.quote(redirect_uri)}&state={state}&code_challenge={code_challenge}&code_challenge_method=S256"
      
      sys.stderr.write("\n==========================================================\n")
      sys.stderr.write("Authentication required for Domoticz MCP Server.\n")
      sys.stderr.write("Please open this URL in your browser to authenticate:\n\n")
      sys.stderr.write(f"{auth_url}\n\n")
      sys.stderr.write(f"Waiting for authentication on {redirect_uri}...\n")
      sys.stderr.write("==========================================================\n\n")
      sys.stderr.flush()
      
      try:
          webbrowser.open(auth_url)
      except Exception:
          pass
          
      server.serve_forever()
      
      if code and state_received == state:
          return code, code_verifier, redirect_uri
      return None, None, None
  
  async def _fetch_oauth_token(force_refresh: bool = False) -> Optional[str]:
      global _oauth_token_cache
      
      if not force_refresh and _oauth_token_cache:
          return _oauth_token_cache
          
      # Try loading from file
      existing_token_data = None
      if os.path.exists(TOKEN_FILE):
          try:
              with open(TOKEN_FILE, 'r') as f:
                  existing_token_data = json.load(f)
                  if not force_refresh and existing_token_data and "access_token" in existing_token_data:
                      _oauth_token_cache = existing_token_data["access_token"]
                      return _oauth_token_cache
          except Exception as e:
              sys.stderr.write(f"Failed to load token file: {e}\n")
  
      if not DOMOTICZ_CLIENT_ID:
          return None
  
      try:
          async with httpx.AsyncClient() as client:
              discovery_url = f"{DOMOTICZ_BASE_URL}/.well-known/openid-configuration"
              resp = await client.get(discovery_url)
              resp.raise_for_status()
              config = resp.json()
              
              token_endpoint = config.get("token_endpoint")
              if not token_endpoint:
                  return None
                  
              if "127.0.0.1" in token_endpoint or "localhost" in token_endpoint:
                  parsed = urllib.parse.urlparse(token_endpoint)
                  token_endpoint = f"{DOMOTICZ_BASE_URL}{parsed.path}"
  
              auth = (DOMOTICZ_CLIENT_ID, DOMOTICZ_CLIENT_SECRET) if DOMOTICZ_CLIENT_SECRET else None
              token_resp = None
  
              # Try refresh token first
              if force_refresh and existing_token_data and "refresh_token" in existing_token_data:
                  data = {
                      "grant_type": "refresh_token",
                      "refresh_token": existing_token_data["refresh_token"],
                      "client_id": DOMOTICZ_CLIENT_ID
                  }
                  
                  try:
                      refresh_resp = await client.post(
                          token_endpoint,
                          auth=auth,
                          data=data
                      )
                      if refresh_resp.status_code == 200:
                          token_resp = refresh_resp
                      else:
                          sys.stderr.write(f"Failed to refresh token (Status: {refresh_resp.status_code}), re-authenticating...\n")
                  except Exception as e:
                      sys.stderr.write(f"Exception during token refresh: {e}\n")
              
              # If no token_resp yet, perform initial flow
              if token_resp is None:
                  if DOMOTICZ_USERNAME and DOMOTICZ_PASSWORD:
                      data = {
                          "grant_type": "password",
                          "client_id": DOMOTICZ_CLIENT_ID,
                          "client_secret": DOMOTICZ_CLIENT_SECRET
                      }
                      token_resp = await client.post(
                          token_endpoint, 
                          auth=(DOMOTICZ_USERNAME, DOMOTICZ_PASSWORD),
                          data=data
                      )
                  else:
                      code, code_verifier, redirect_uri = await asyncio.to_thread(_do_interactive_oauth_flow)
                      if not code:
                          return None
                          
                      data = {
                          "grant_type": "authorization_code",
                          "code": code,
                          "redirect_uri": redirect_uri,
                          "client_id": DOMOTICZ_CLIENT_ID,
                          "code_verifier": code_verifier
                      }
                      
                      token_resp = await client.post(
                          token_endpoint,
                          auth=auth,
                          data=data
                      )
  
              token_resp.raise_for_status()
              token_data = token_resp.json()
              
              # Persist the old refresh token if the new response doesn't provide one
              if "refresh_token" not in token_data and existing_token_data and "refresh_token" in existing_token_data:
                  token_data["refresh_token"] = existing_token_data["refresh_token"]
  
              _oauth_token_cache = token_data.get("access_token")
              
              # Save token to file
              os.makedirs(os.path.dirname(TOKEN_FILE), exist_ok=True)
              with open(TOKEN_FILE, 'w') as f:
                  json.dump(token_data, f)
                  
              return _oauth_token_cache
      except Exception as e:
          import logging
          logging.error(f"Failed to fetch OAuth token: {e}")
          return None
  
  async def _do_request(client: httpx.AsyncClient, method: str, url: str, **kwargs) -> httpx.Response:
      """Perform a request with a single retry on 401 Unauthorized to handle expired tokens."""
      global _oauth_token_cache
      
      try:
          resp = await client.request(method, url, **kwargs)
          if resp.status_code == 401:
              # Token might be expired. Clear cache and retry once.
              _oauth_token_cache = None
              
              # Re-fetch token (this will trigger OAuth flow if needed)
              new_token = await _fetch_oauth_token(force_refresh=True)
              if new_token:
                  # Update headers for the retry
                  if "headers" not in kwargs:
                      kwargs["headers"] = {}
                  kwargs["headers"]["Authorization"] = f"Bearer {new_token}"
                  
                  # Retry the request
                  resp = await client.request(method, url, **kwargs)
          
          resp.raise_for_status()
          return resp
      except httpx.HTTPStatusError as e:
          if e.response.status_code == 401:
              raise Exception("Authentication failed. Please check your credentials or re-authenticate.")
          raise e
  
  # Custom AsyncClient wrapper that ensures the token is added
  class DomoticzClient:
      def __init__(self, own_client: bool = False):
          self._own_client = own_client
          if own_client or _global_http_client is None:
              self.client: httpx.AsyncClient = httpx.AsyncClient(timeout=30.0)
              self._owns_client = True
          else:
              self.client = _global_http_client
              self._owns_client = False
  
      async def __aenter__(self) -> "httpx.AsyncClient":
          oauth_token = None
  
          if DOMOTICZ_CLIENT_ID:
              oauth_token = await _fetch_oauth_token()
  
          if oauth_token:
              self.client.headers["Authorization"] = f"Bearer {oauth_token}"
          elif DOMOTICZ_USERNAME and DOMOTICZ_PASSWORD:
              self.client.auth = (DOMOTICZ_USERNAME, DOMOTICZ_PASSWORD)
          else:
              self.client.headers.pop("Authorization", None)
              self.client.auth = None
  
          return self.client
  
      async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
          if self._owns_client:
              await self.client.aclose()