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}