get_health_summary_es
Retrieve a summary of Apple Health data from Elasticsearch, including total record count, type breakdown, and optional date range aggregation for analysis.
Instructions
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, auto-fill, or assume any missing data.
When asked for medical advice, try to use my data from ElasticSearch first.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
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) - 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, }