Apple Health MCP Server

Instructions

Implementation Reference

• app/mcp/v1/tools/es_reader.py:42-74 (handler)

  FastMCP tool handler for 'search_health_records_es' that delegates to the core Elasticsearch search logic.

  @es_reader_router.tool
  def search_health_records_es(params: HealthRecordSearchParams) -> list[dict[str, Any]]:
      """
      Search health records in Elasticsearch with flexible query building.
  
      Parameters:
      - params: HealthRecordSearchParams object containing all search/filter parameters.
  
      Notes for LLMs:
      - This function should return a list of health record documents (dicts)
        matching the search criteria.
      - Each document in the list should represent a single health record as stored in Elasticsearch.
      - If an error occurs, the function should return a list with a single dict
        containing an 'error' key and the error message.
      - Use this to retrieve structured health data for further analysis, filtering, or display.
      - Example source_name: "Rob’s iPhone", "Polar Flow", "Sync Solver".
      - Example date_from/date_to: "2020-01-01T00:00:00+00:00"
      - Example value_min/value_max: "10", "100.5"
      - 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 search_health_records_logic(params)
      except Exception as e:
          return [{"error": f"Failed to search health records: {str(e)}"}]

• app/schemas/record.py:39-49 (schema)

  Pydantic input schema defining parameters for searching health records (record_type, source_name, dates, value ranges, limit).

  class HealthRecordSearchParams(BaseModel):
      record_type: RecordType | WorkoutType | str | None = None
      source_name: str | None = None
      date_from: str | None = None
      date_to: str | None = None
      min_workout_duration: str | None = None
      max_workout_duration: str | None = None
      value_min: str | None = None
      value_max: str | None = None
      limit: int = 10

• app/services/health/elasticsearch.py:48-69 (helper)

  Core helper function implementing the Elasticsearch query construction and execution for health record search.

  def search_health_records_logic(params: HealthRecordSearchParams) -> list[dict[str, Any]]:
      must_conditions = []
      if params.record_type:
          must_conditions.append(_build_match_condition("type", params.record_type))
      if params.source_name:
          must_conditions.append(_build_match_condition("sourceName", params.source_name))
      if params.date_from or params.date_to:
          range_cond = _build_range_condition("dateComponents", params.date_from, params.date_to)
          if range_cond:
              must_conditions.append(range_cond)
      if params.value_min is not None or params.value_max is not None:
          range_cond = _build_range_condition("value", params.value_min, params.value_max)
          if range_cond:
              must_conditions.append(range_cond)
      query: dict[str, Any]
      if must_conditions:
          query = {"query": {"bool": {"must": must_conditions}}}
      else:
          query = {"query": {"match_all": {}}}
      query["size"] = params.limit
      response = _run_es_query(query)
      return [hit["_source"] for hit in response["hits"]["hits"]]

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

  Mounts the es_reader_router (containing search_health_records_es tool) into the main MCP router, making the tool available.

  from app.mcp.v1.tools import duckdb_reader, es_reader, xml_reader
  
  mcp_router = FastMCP(name="Main MCP")
  
  mcp_router.mount(duckdb_reader.duckdb_reader_router)
  # mcp_router.mount(ch_reader.ch_reader_router)
  mcp_router.mount(es_reader.es_reader_router)