Apple Health MCP Server

Instructions

Implementation Reference

• app/mcp/v1/tools/es_reader.py:127-182 (handler)

  MCP tool handler for get_trend_data_es. Registers the tool with the ES Reader MCP router and delegates execution to the underlying logic function.

  @es_reader_router.tool
  def get_trend_data_es(
      record_type: RecordType | str,
      interval: IntervalType = "month",
      date_from: str | None = None,
      date_to: str | None = None,
  ) -> dict[str, Any]:
      """
      Get trend data for a specific health record type over time using
      Elasticsearch date histogram aggregation.
  
      Parameters:
      - record_type: The type of health record to analyze (e.g., "HKQuantityTypeIdentifierStepCount")
      - interval: Time interval for aggregation.
      - date_from, date_to: Optional ISO8601 date strings for filtering date range
  
      Returns:
      - record_type: The analyzed record type
      - device: The device on which the data was recorded
      - interval: The time interval used
      - trend_data: List of time buckets with statistics for each period:
        * date: The time period (ISO string)
        * value_sum: Sum of values for the period
        * avg_value: Average value for the period
        * min_value: Minimum value for the period
        * max_value: Maximum value for the period
        * count: Number of records in the period
  
      Notes for LLMs:
      - Use this to analyze trends, patterns, and seasonal variations in health data
      - Keep in mind that when there is data from multiple devices spanning the same
        time period, there is a possibility of data being duplicated. Inform the user
        of this possibility if you see multiple devices in the same time period.
      - If a user asks you to sum up some values from their health records, DO NOT
        search for records and write a script to sum them, instead, use this tool:
        if they ask to sum data from a year, use this tool with date_from set as the
        beginning of the year and date_to as the end of the year, with an interval
        of 'year'
      - The function automatically handles date filtering if date_from/date_to are provided
      - IMPORTANT - interval must be one of: "day", "week", "month", or "year".
        Do not use other values.
      - Do not guess, autofill, or assume any missing data.
      - If there are multiple databases available (DuckDB, ClickHouse, Elasticsearch):
        first, ask the user which one he wants to use. DO NOT call any tools before
        the user specifies his intent.
      - If the user decides on an option, only use tools from this database,
        do not switch over to another until the user specifies that he wants
        to use a different one. You do not have to keep asking whether
        the user wants to use the same database that he used before.
      - If there is only one database available (DuckDB, ClickHouse, Elasticsearch):
        you can use the tools from this database without the user specifying it.
      """
      try:
          return get_trend_data_logic(record_type, interval, date_from, date_to)
      except Exception as e:
          return {"error": f"Failed to get trend data: {str(e)}"}

• app/services/health/elasticsearch.py:88-129 (helper)

  Core implementation logic for retrieving trend data using Elasticsearch date_histogram aggregation, including query building, execution, and response processing.

  def get_trend_data_logic(
      record_type: RecordType | str,
      interval: IntervalType = "month",
      date_from: str | None = None,
      date_to: str | None = None,
  ) -> dict[str, Any]:
      must_conditions = [{"match": {"type": record_type}}]
      if date_from or date_to:
          range_cond = _build_range_condition("dateComponents", date_from, date_to)
          if range_cond:
              must_conditions.append(range_cond)
      query = {
          "size": 0,
          "query": {"bool": {"must": must_conditions}},
          "aggs": {
              "trend_over_time": {
                  "date_histogram": {"field": "dateComponents", "calendar_interval": interval},
                  "aggs": {
                      "avg_value": {"avg": {"field": "value"}},
                      "min_value": {"min": {"field": "value"}},
                      "max_value": {"max": {"field": "value"}},
                      "value_sum": {"sum": {"field": "value"}},
                      "count": {"value_count": {"field": "value"}},
                  },
              },
          },
      }
      response = _run_es_query(query)
      buckets = response["aggregations"]["trend_over_time"]["buckets"]
      trend_data = []
      for bucket in buckets:
          trend_data.append(
              {
                  "date": bucket["key_as_string"],
                  "avg_value": bucket["avg_value"]["value"],
                  "min_value": bucket["min_value"]["value"],
                  "max_value": bucket["max_value"]["value"],
                  "value_sum": bucket["value_sum"]["value"],
                  "count": bucket["count"]["value"],
              },
          )
      return {"record_type": record_type, "interval": interval, "trend_data": trend_data}