Hass-MCP

Instructions

Implementation Reference

• app/server.py:1124-1204 (handler)

  The handler function that implements the 'get_history' tool logic. It fetches entity history data using get_entity_history helper, processes and formats the state changes, handles errors, and returns a structured dictionary with history information.

  @mcp.tool()
  @async_handler("get_history")
  async def get_history(entity_id: str, hours: int = 24) -> Dict[str, Any]:
      """
      Get the history of an entity's state changes
      
      Args:
          entity_id: The entity ID to get history for
          hours: Number of hours of history to retrieve (default: 24)
      
      Returns:
          A dictionary containing:
          - entity_id: The entity ID requested
          - states: List of state objects with timestamps
          - count: Number of state changes found
          - first_changed: Timestamp of earliest state change
          - last_changed: Timestamp of most recent state change
          
      Examples:
          entity_id="light.living_room" - get 24h history
          entity_id="sensor.temperature", hours=168 - get 7 day history
      Best Practices:
          - Keep hours reasonable (24-72) for token efficiency
          - Use for entities with discrete state changes rather than continuously changing sensors
          - Consider the state distribution rather than every individual state    
      """
      logger.info(f"Getting history for entity: {entity_id}, hours: {hours}")
      
      try:
          # Call the new hass function to get history
          history_data = await get_entity_history(entity_id, hours)
          
          # Check for errors from the API call
          if isinstance(history_data, dict) and "error" in history_data:
              return {
                  "entity_id": entity_id,
                  "error": history_data["error"],
                  "states": [],
                  "count": 0
              }
          
          # The result from the API is a list of lists of state changes
          # We need to flatten it and process it
          states = []
          if history_data and isinstance(history_data, list):
              for state_list in history_data:
                  states.extend(state_list)
          
          if not states:
              return {
                  "entity_id": entity_id,
                  "states": [],
                  "count": 0,
                  "first_changed": None,
                  "last_changed": None,
                  "note": "No state changes found in the specified timeframe."
              }
          
          # Sort states by last_changed timestamp
          states.sort(key=lambda x: x.get("last_changed", ""))
          
          # Extract first and last changed timestamps
          first_changed = states[0].get("last_changed")
          last_changed = states[-1].get("last_changed")
          
          return {
              "entity_id": entity_id,
              "states": states,
              "count": len(states),
              "first_changed": first_changed,
              "last_changed": last_changed
          }
      except Exception as e:
          logger.error(f"Error processing history for {entity_id}: {str(e)}")
          return {
              "entity_id": entity_id,
              "error": f"Error processing history: {str(e)}",
              "states": [],
              "count": 0
          }