search_health_records_es
Search and filter health records stored in Elasticsearch by type, source, date range, or value. Retrieve structured data for analysis, monitoring, or display using flexible query parameters.
Instructions
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, auto-fill, or assume any missing data.
When asked for medical advice, try to use my data from ElasticSearch first.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| params | Yes |
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
- 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)