Hass-MCP

Instructions

Implementation Reference

• app/server.py:1124-1125 (registration)

  Registers the get_history function as an MCP tool using the @mcp.tool() decorator. The @async_handler decorator adds logging.
  @mcp.tool() @async_handler("get_history")
• app/server.py:1126-1204 (handler)

  The main handler function for the 'get_history' tool. It calls the helper get_entity_history, processes the raw history data (flattens lists, sorts by timestamp), extracts metadata (count, first/last changed), and returns a structured response. Includes comprehensive error handling and input validation via type hints and docstring.
  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 }
• app/hass.py:507-543 (helper)

  Supporting utility that makes the direct API call to Home Assistant's /api/history/period endpoint to fetch raw history data for the specified entity over the given hours. Handles date calculations, minimal response, and error handling via decorator.
  """ Get the history of an entity's state changes from Home Assistant. Args: entity_id: The entity ID to get history for. hours: Number of hours of history to retrieve. Returns: A list of state change objects, or an error dictionary. """ client = await get_client() # Calculate the end time for the history lookup end_time = datetime.now(timezone.utc) end_time_iso = end_time.strftime("%Y-%m-%dT%H:%M:%SZ") # Calculate the start time for the history lookup based on end_time start_time = end_time - timedelta(hours=hours) start_time_iso = start_time.strftime("%Y-%m-%dT%H:%M:%SZ") # Construct the API URL url = f"{HA_URL}/api/history/period/{start_time_iso}" # Set query parameters params = { "filter_entity_id": entity_id, "minimal_response": "true", "end_time": end_time_iso, } # Make the API call response = await client.get(url, headers=get_ha_headers(), params=params) response.raise_for_status() # Return the JSON response return response.json()