Apple Health MCP Server

Instructions

Implementation Reference

• app/mcp/v1/tools/es_reader.py:17-39 (handler)

  MCP tool handler for get_health_summary_es. Decorated with @es_reader_router.tool decorator for registration. Delegates to the helper function and handles exceptions.

  @es_reader_router.tool
  def get_health_summary_es() -> dict[str, Any]:
      """
      Get a summary of Apple Health data from Elasticsearch.
      The function returns total record count, record type breakdown, and
      (optionally) a date range aggregation.
  
      Notes for LLM:
      - IMPORTANT - 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_health_summary_from_es()
      except Exception as e:
          return {"error": f"Failed to get health summary: {str(e)}"}

• app/mcp/v1/mcp.py:9-9 (registration)

  Registration of the es_reader_router (containing the get_health_summary_es tool) by mounting it onto the main mcp_router.

  mcp_router.mount(es_reader.es_reader_router)

• app/services/health/elasticsearch.py:27-45 (helper)

  Core implementation logic that executes the Elasticsearch aggregation query to retrieve total record count and breakdown by record types.

  def get_health_summary_from_es() -> dict[str, Any]:
      query = {
          "size": 0,
          "aggs": {
              "total_records": {"value_count": {"field": "type.keyword"}},
              "record_types": {"terms": {"field": "type.keyword", "size": 50}},
          },
      }
      response = _run_es_query(query)
      total_records_value = response["aggregations"]["total_records"]["value"]
      record_types = {
          bucket["key"]: bucket["doc_count"]
          for bucket in response["aggregations"]["record_types"]["buckets"]
      }
      return {
          "total_records": total_records_value,
          "record_types": record_types,
          "index_name": settings.ES_INDEX,
      }