hass-mcp-plus

Instructions

Implementation Reference

• app/server.py:407-422 (handler)

  The MCP tool handler for 'call_service'. Decorated with @mcp.tool() to register it as an MCP tool, and @async_handler('call_service') for logging. This function accepts domain, service, and optional data parameters and delegates to the underlying hass_call_service implementation.

  @mcp.tool()
  @async_handler("call_service")
  async def call_service(domain: str, service: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
      """
      Call any Home Assistant service directly (low-level API).
  
      Args:
          domain: Service domain (e.g. 'light', 'automation')
          service: Service name (e.g. 'turn_on', 'reload')
          data: Service data (e.g. {'entity_id': 'light.x', 'brightness': 255})
  
      Examples: call_service("light", "turn_on", {"entity_id": "light.x", "brightness": 255})
                call_service("automation", "reload")
      """
      logger.info(f"Calling Home Assistant service: {domain}.{service} with data: {sanitize_for_logging(data)}")
      return await hass_call_service(domain, service, data or {})

• app/hass/services.py:18-45 (helper)

  The core implementation of call_service that makes the actual HTTP POST request to Home Assistant's /api/services/{domain}/{service} endpoint. Handles validation, rate limiting, and response processing.

  @handle_api_errors
  async def call_service(domain: str, service: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
      """Call a Home Assistant service"""
      validate_ha_identifier(domain, "domain")
      validate_ha_identifier(service, "service")
  
      if data is None:
          data = {}
      else:
          validate_service_payload(data)
  
      await _rate_limiter.acquire()
      client = await get_client()
      response = await client.post(
          f"{HA_URL}/api/services/{safe_url_path_segment(domain)}/{safe_url_path_segment(service)}",
          headers=get_ha_headers(),
          json=data
      )
      response.raise_for_status()
  
      result = response.json()
  
      # Handle list responses from Home Assistant
      # Service calls return lists (empty or with changed entity data)
      if isinstance(result, list):
          result = {"result": result} if result else {}
  
      return result

• app/hass/validation.py:59-82 (schema)

  Schema validation function validate_ha_identifier that validates domain and service name parameters. Ensures identifiers contain only lowercase letters, numbers, and underscores to prevent injection attacks.

  def validate_ha_identifier(value: str, name: str = "identifier") -> str:
      """
      Validate a Home Assistant identifier (domain, service name, etc.).
  
      Args:
          value: The identifier to validate
          name: Human-readable name for error messages (e.g., 'domain', 'service')
  
      Returns:
          The validated identifier
  
      Raises:
          ValidationError: If the identifier format is invalid
      """
      if not value or not isinstance(value, str):
          raise ValidationError(f"{name.capitalize()} must be a non-empty string")
  
      if not HA_IDENTIFIER_PATTERN.match(value):
          raise ValidationError(
              f"Invalid {name}: '{value}'. "
              f"Must contain only lowercase letters, numbers, and underscores."
          )
  
      return value

• app/hass/validation.py:156-187 (schema)

  Schema validation function validate_service_payload that validates the service data payload. Checks JSON serializability and enforces a maximum payload size (1 MB by default) to prevent resource exhaustion.

  def validate_service_payload(
      data: Optional[dict],
      max_bytes: int = MAX_SERVICE_PAYLOAD_BYTES,
  ) -> Optional[dict]:
      """
      Validate service call payload size.
  
      Args:
          data: The service call data payload
          max_bytes: Maximum allowed size in bytes
  
      Returns:
          The validated data
  
      Raises:
          ValidationError: If payload exceeds size limit or is not serializable
      """
      if data is None:
          return data
  
      try:
          payload_size = len(json.dumps(data))
      except (TypeError, ValueError):
          raise ValidationError("Service data must be JSON-serializable")
  
      if payload_size > max_bytes:
          raise ValidationError(
              f"Service data payload too large: {payload_size:,} bytes "
              f"(maximum: {max_bytes:,} bytes)"
          )
  
      return data

• app/hass/__init__.py:90-96 (registration)

  Module-level registration that exports call_service from app.hass.services, making it available for import by the server module.

  # --- services ---
  from app.hass.services import (  # noqa: F401
      call_service,
      get_hass_version,
      reload_automations,
      restart_home_assistant,
  )