作者：来自 Elastic Alex Salgado
学习如何使用 FastMCP 和 Elasticsearch 构建一个 MCP 服务器来管理和搜索数据。
Elasticsearch 具备与业界领先的生成式 AI 工具和服务商的原生集成。查看我们的网络研讨会，了解如何超越 RAG 基础，或使用 Elastic 向量数据库构建可投入生产的应用。
为了构建最适合你用例的搜索解决方案，现在就开始免费的云试用，或在本地机器上试用 Elastic。
大型语言模型（LLMs）是革命性的工具，但当它们拥有实时上下文并能与外部世界交互时，其真正潜力才能被释放。模型上下文协议（Model Context Protocol - MCP）就像一座桥梁，将你的 LLM 连接到真实数据和外部工具。通过提供这些上下文，LLM 能够给出更相关、更真实的响应。
尽管传统的 RAG（检索增强生成）为 LLM 提供上下文，但它通常依赖于预先存储在数据库或私有数据集中的静态信息。相比之下，MCP 可以将来自数据库、API、传感器和其他数据流等各种系统的动态实时信息作为上下文引入。
在这篇博客中，你将学习如何使用 FastMCP 和 Elasticsearch 构建一个 MCP 服务器来管理和搜索数据。作为示例，我们将使用 Apple Watch 数据，演示如何将原始信息转化为智能洞察。
为什么选择 MCP？
MCP 是一个开放协议，用于标准化 LLMs 如何与外部数据源和工具交互。想了解详细介绍，请参阅 JD Armada 的文章： MCP（Model Context Protocol，模型上下文协议）的当前状态。
MCP 在 LLM Agent 开发中的优势
使用 MCP 为基于 LLM 的 Agent 开发带来多个优势：
职责分离：MCP 将上下文逻辑（数据和工具）与 LLM 交互分离，使开发更有条理。
标准化：遵循标准协议，开发者可以构建兼容任何 MCP 客户端的工具，无论使用哪个 LLM。
可重用性：已创建的工具和资源可以轻松复用于不同项目和上下文中。
可扩展性：通过 MCP，可以横向扩展，添加更多服务器以处理不同的知识领域或功能。
MCP 的三个核心原语
MCP 定义了三个主要组件，用于建立 LLM 与外部系统交互的标准：
1) Resources：LLM 的 “眼睛”
功能：通过直接暴露数据为 LLM 提供上下文
行为：类似 REST API 中的 GET 端点，只读取数据不做修改
示例： health://steps/latest - 获取最新的步数记录
URI 模式：使用域名命名空间模式
2) Tools：LLM 的 “手”
功能：支持执行操作和计算，是处理复杂查询和分析的关键
行为：类似 REST API 中的 POST 端点，会执行操作，可能有副作用
特点：
接受结构化参数
根据参数执行动态逻辑
主动处理和操作数据，而不仅仅是获取数据
3) Prompts：LLM 的 “地图”（MCP 特有）
功能：为常见任务提供可重用的命令模板，与传统将用户输入和任务说明混合的 Prompt 不同
行为：作为已注册模板出现，可作为斜杠命令使用
示例：/daily_report - 触发预设的分析工作流
目的：规范常见交互与工作流程
构建你自己的 MCP 服务器
解决方案架构
在开始编码之前，先了解一下我们要构建的内容：
我们的 MCP 服务器充当 LLM Agent 和你的 Elasticsearch 数据之间的桥梁，使你能够对结构化健康数据进行自然语言查询。
前提条件
本文假设你已安装以下技术：
完整项目代码托管在 GitHub
整个项目包含脚本和示例数据，均托管在这个 GitHub 仓库中。如果你愿意，可以按照 README.md 文件中的说明快速完成环境搭建。
在终端中克隆该仓库，使用辅助脚本如 ingest_data.py 来准备数据库，然后再继续后续章节。
git clone https://github.com/salgado/apple-watch-health-mcp.git
cd apple-watch-health-mcp当你列出文件夹内容（使用 ls 命令）时，你应该看到以下文件：
apple_watch_mcp.py：MCP 服务器的完整脚本，包含所有实现的 Resources、Tools 和 Prompts。
ingest_data.py：辅助脚本，用于将示例数据导入到你的 Elasticsearch 实例。
sample_data.json：包含虚构 Apple Health 步数数据的示例 JSON 文件。
README.md：完整解决方案的运行说明。
在 Python 中设置 FastMCP 框架
现在，让我们为 MCP 服务器搭建环境。FastMCP 是一个 Python 框架，可以轻松创建与 MCP 协议兼容的完整服务器。
MCP + Elasticsearch 有两种集成方式
在构建自定义服务器之前，了解两种集成 MCP 与 Elasticsearch 的方式很重要：
1) 官方方案：使用 Elastic MCP Server
通用：适用于大多数用例
配置简单：快速轻松搭建
定制有限：特定场景灵活性较低
操作受限：撰写时仅支持基本查询操作
2) 自定义方案：构建你自己的 MCP 服务器（本文重点）
领域专用：针对特定用例优化
需要编程：开发工作量较大
完全控制：功能灵活可控
首先，进入项目文件夹，使用 uv 管理依赖：
# Navigate to the repository folder
cd apple-watch-health-mcp
# Create and initialize a new project
uv init
# Create and activate a virtual environment
uv venv
source .venv/bin/activate
# Install required dependencies
uv add "mcp[cli]" "elasticsearch>=8.0.0,<9.0.0" aiohttp pydantic
# Set up environment variable for API key (after creating it in Elasticsearch)
export ES_API_KEY="your_encoded_api_key_here"注意，我们指定了与本博客中使用的 Elasticsearch v8.x 服务器兼容的库版本。
用例：基于 Apple HealthKit 数据的知识库助手
导出 Apple Watch 数据
要获取 Apple HealthKit 数据，请参阅 Apple 官方导出指南。导出会生成包含所有健康记录的 XML 文件。
使用示例数据设置 Elasticsearch
在创建 MCP 服务器之前，我们需要用示例数据准备 Elasticsearch。仓库中已包含一个苹果健康步数数据文件（sample_data.json）和一个导入脚本（ingest_data.py）。
要准备 Elasticsearch，请在终端运行以下命令：
# Set the environment variable with Elasticsearch API KEY
export ES_API_KEY="ES_API_KEY"
python ingest_data.py该脚本将会：
连接到 http://localhost:9200 的 Elasticsearch
使用正确的映射创建索引
从 JSON 文件插入数据
如果你想使用其他连接选项，比如 Elastic Cloud，请查阅相关文档。默认配置使用：
ES_HOST = "http://localhost:9200"
ES_INDEX = "apple-health-steps"
ES_API_KEY = os.getenv('ES_API_KEY')过程结束时你应该会看到成功消息。
Connecting to Elasticsearch at http://localhost:9200...
Connection successful!
Index 'apple-health-steps' found. Deleting...
Index deleted.
Creating index 'apple-health-steps' with the specified mapping...
Index created successfully.
Reading data from 'sample_data.json' for ingestion...
Ingestion complete. Documents successfully ingested: 68
Final check: The index 'apple-health-steps' now contains 68 documents.脚本执行完成后，索引将被创建并填充示例数据。该索引遵循以下模式，对应从 Apple Watch 导出的步数数据 XML 文件中的字段：
GET /apple-health-steps/_mapping
{
"apple-health-steps": {
"mappings": {
"properties": {
"creationDate": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis" },
"day": { "type": "date", "format": "yyyy-MM-dd" },
"dayOfWeek": { "type": "keyword" },
"deviceInfo": {
"properties": {
"hardware": { "type": "keyword" },
"manufacturer": { "type": "keyword" },
"model": { "type": "keyword" },
"name": { "type": "keyword" },
"software": { "type": "keyword" }
}
},
"duration": { "type": "float" },
"endDate": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis" },
"hour": { "type": "integer" },
"sourceName": { "type": "keyword" },
"startDate": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis" },
"type": { "type": "keyword" },
"unit": { "type": "keyword" },
"value": { "type": "float" }
}
}
}
}以下是插入的数据示例：
{
"type": "HKQuantityTypeIdentifierStepCount",
"sourceName": "Apple Watch",
"deviceInfo": {
"name": "Apple Watch",
"manufacturer": "Apple Inc.",
"model": "Watch",
"hardware": "Watch7,1",
"software": "11.3.1"
},
"unit": "count",
"creationDate": "2025-06-01 08:11:04",
"startDate": "2025-06-01 07:58:42",
"endDate": "2025-06-01 07:59:10",
"value": 340,
"day": "2025-06-01",
"dayOfWeek": "Sunday",
"hour": 8,
"duration": 583
}数据包含来自 Apple Watch 和 iPhone 两个来源的步数计数，记录了详细的时间戳和设备信息。
数据库创建并填充完成后，我们可以开始构建 MCP 服务器来访问这些数据。
构建健康数据的 MCP 服务器
使用 FastMCP 和 Pydantic 进行初始配置
让我们创建一个 apple_watch_mcp.py 文件，包含 MCP 服务器的基本结构：
# apple_watch_health_mcp.py
import os
from typing import Any, Optional
import json
from datetime import datetime
from pydantic import BaseModel, field_validator, ValidationError
from mcp.server.fastmcp import FastMCP
from elasticsearch import AsyncElasticsearch
from contextlib import asynccontextmanager
# Initialize FastMCP server
mcp = FastMCP("apple-watch-steps")
# Constants
ES_HOST = "http://localhost:9200"
ES_INDEX = "apple-health-steps"
# API key from environment variable or fallback for development
ES_API_KEY = os.getenv('ES_API_KEY')
# Pydantic model for parameter validation
class QueryStepDataParams(BaseModel):
start_date: Optional[str] = None
end_date: Optional[str] = None
aggregation: Optional[str] = None
device: Optional[str] = None
@field_validator('start_date', 'end_date')
def validate_date_format(cls, value):
if value is None:
return value
try:
datetime.strptime(value, "%Y-%m-%d")
return value
except ValueError:
raise ValueError("Invalid date format. Use YYYY-MM-DD")
@field_validator('aggregation')
def validate_aggregation(cls, value):
valid_aggregations = ["hourly", "daily", "weekly", "monthly", None]
if value not in valid_aggregations:
raise ValueError(f"Invalid aggregation. Use one of: {valid_aggregations[:-1]}")
return value
# Main function to run the server
if __name__ == "__main__":
mcp.run()这段初始代码设置了一个名为 “apple-watch-steps” 的 MCP 服务器，并定义了一个 Pydantic 模型来验证工具中使用的参数。服务器将连接到本地 Elasticsearch 集群，使用 “apple-health-steps” 索引。
连接 Elasticsearch：我们的数据桥梁
为了与 Elasticsearch 交互，让我们在代码中添加一个辅助函数：
@asynccontextmanager
async def get_es_client():
"""Context manager for Elasticsearch client."""
client = AsyncElasticsearch([ES_HOST], api_key=ES_API_KEY)
try:
yield client
finally:
await client.close()
# Elasticsearch helper function
async def query_elasticsearch(query: dict) -> dict[str, Any] | None:
"""Makes a request to Elasticsearch with proper error handling."""
print(f"Sending query to Elasticsearch: {json.dumps(query)}")
# Use context manager
async with get_es_client() as client:
try:
response = await client.search(
index=ES_INDEX,
body=query
)
return response
except Exception as e:
print(f"Error querying Elasticsearch: {e}")
return None该实现使用官方 Elasticsearch Python 客户端中的 AsyncElasticsearch 类，支持完整的 async/await 异步操作。采用异步方式替代传统同步客户端操作，可以避免长时间任务阻塞主线程，使服务器响应更快。上下文管理器模式（context manager pattern）确保每次操作时连接正确打开和关闭，防止资源泄漏和连接警告。函数接收一个查询字典，发送到 Elasticsearch，并返回 JSON 格式结果；如果出错则返回 None。
MCP 的三大支柱：Resources、Tools 和 Prompts
现在我们来实现 MCP 的 Resources、Tools 和 Prompts，每个组件有其特定角色。
示例 Resource：
@mcp.resource("health://steps/latest")
async def get_latest_steps() -> str:
"""Gets the most recent step records"""
query = {
"query": {
"match_all": {}
},
"sort": [
{"endDate": {"order": "desc"}}
],
"size": 10
}
data = await query_elasticsearch(query)
if not data:
return json.dumps({"error": "Unable to query Elasticsearch"}, indent=2)
results = []
for hit in data["hits"]["hits"]:
source = hit["_source"]
results.append({
"startDate": source.get("startDate"),
"endDate": source.get("endDate"),
"value": source.get("value"),
"device": source.get("device"),
"sourceName": source.get("sourceName"),
"dayOfWeek": source.get("dayOfWeek"),
"hour": source.get("hour")
})
return json.dumps({
"latest_steps": results
}, indent=2)该 Resource 会查询 Elasticsearch，获取按结束时间排序的最近 10 条步数记录。本文后面会通过实际 Claude Desktop 交互示例展示输出格式和数据效果。
健康数据智能助手
我们的用例是创建一个知识库助手，使 LLM Agent 能与 Apple HealthKit 健康数据交互。重点是步数数据，但同样方法可扩展到其他健康数据类型。
该助手允许用户：
用对话方式查询步数数据
获取个性化分析和可视化
比较不同设备的数据
识别长期趋势和模式
实现 Resources：步数数据的读取端点
我们的 MCP 服务器将实现三个主要 Resource：
数据类型列表：显示数据库中可用的数据类型
最近数据获取：获取最新的步数记录
统计汇总：提供数据概览
让我们开始实现这些 Resources：
# Resources
@mcp.resource("health://steps/types")
async def list_step_types() -> str:
"""List all available step types in the database"""
query = {
"size": 0,
"aggs": {
"step_types": {
"terms": {
"field": "type",
"size": 100
}
}
}
}
data = await query_elasticsearch(query)
if not data:
return json.dumps({"error": "Unable to query Elasticsearch"}, indent=2)
step_types = [bucket["key"] for bucket in
data["aggregations"]["step_types"]["buckets"]]
return json.dumps({
"available_types": step_types,
"count": len(step_types)
}, indent=2)
@mcp.resource("health://steps/summary")
async def get_steps_summary() -> str:
"""Get summary statistics for step counts"""
query = {
"aggs": {
"all_time": {
"stats": {
"field": "value"
}
}
},
"size": 0
}
data = await query_elasticsearch(query)
if not data:
return json.dumps({"error": "Unable to query Elasticsearch"}, indent=2)
return json.dumps(data["aggregations"], indent=2)与智能 RAG 应用集成
Tools：将 MCP 连接到 RAG 工作流程
为了将我们的 MCP 服务器与 RAG（检索增强生成）工作流程集成，我们需要实现支持自定义复杂查询的工具。下图展示了 MCP 如何融入完整的 RAG 流程：
该图展示了我们的 MCP 服务器如何完美融入 RAG 流程：
用户提出关于健康数据的问题
查询被 RAG 系统处理
检索阶段使用 MCP 的 query_step_data 工具在 Elasticsearch 中搜索相关数据
检索到的数据用于构建上下文并丰富回复内容
LLM 根据数据和上下文生成最终回复
让我们实现这个流程所需的工具：
# Tools
@mcp.tool()
async def query_step_data(params: QueryStepDataParams) -> str:
"""
Query step data with customizable parameters
Args:
start_date: Start date in YYYY-MM-DD format
end_date: End date in YYYY-MM-DD format
aggregation: Aggregation interval (hourly, daily, weekly, monthly)
device: Filter by specific device name
"""
# Extract parameters from model
start_date = params.start_date or ""
end_date = params.end_date or ""
aggregation = params.aggregation or ""
device = params.device or ""
query = {"query": {"match_all": {}}}
filters = []
# Date filters
date_ranges = []
if start_date:
date_ranges.append({"gte": start_date})
if end_date:
date_ranges.append({"lte": end_date})
if date_ranges:
filters.append({
"range": {
"day": {**{k: v for d in date_ranges for k, v in d.items()}}
}
})
# Device filter
if device:
filters.append({
"wildcard": {
"device": f"*{device}*"
}
})
if filters:
query["query"] = {"bool": {"must": filters}}
# Aggregation handling
if aggregation:
interval_mapping = {
"hourly": "1h",
"daily": "1d",
"weekly": "1w",
"monthly": "1M"
}
interval = interval_mapping.get(aggregation, "1d")
date_field = "startDate" if aggregation == "hourly" else "day"
query["aggs"] = {
"time_series": {
"date_histogram": {
"field": date_field,
"calendar_interval": interval,
"min_doc_count": 0
},
"aggs": {
"total_steps": {"sum": {"field": "value"}},
"avg_steps": {"avg": {"field": "value"}},
"max_steps": {"max": {"field": "value"}},
"min_steps": {"min": {"field": "value"}}
}
}
}
query["size"] = 0
else:
query.update({
"sort": [{"startDate": "desc"}],
"size": 10
})
data = await query_elasticsearch(query)
if not data:
return json.dumps({
"error": "Unable to query Elasticsearch",
"query": query
}, indent=2)
# Process results
results = []
if aggregation and "time_series" in data.get("aggregations", {}):
for bucket in data["aggregations"]["time_series"]["buckets"]:
results.append({
"date": bucket["key_as_string"],
"total_steps": bucket["total_steps"]["value"],
"average_steps": bucket["avg_steps"]["value"],
"max_steps": bucket["max_steps"]["value"],
"min_steps": bucket["min_steps"]["value"],
"records": bucket["doc_count"]
})
else:
for hit in data["hits"]["hits"]:
source = hit["_source"]
results.append({
"startDate": source.get("startDate"),
"endDate": source.get("endDate"),
"day": source.get("day"),
"dayOfWeek": source.get("dayOfWeek"),
"hour": source.get("hour"),
"value": source.get("value"),
"device": source.get("device"),
"sourceName": source.get("sourceName")
})
return json.dumps({
"aggregation": aggregation,
"total_records": data["hits"]["total"]["value"],
"data": results,
"query": query
}, indent=2)
@mcp.tool()
async def get_all_steps() -> str:
"""Get all steps without any filtering"""
query = {
"query": {
"match_all": {}
},
"size": 10,
"sort": [{"startDate": "desc"}]
}
data = await query_elasticsearch(query)
if not data:
return json.dumps({"error": "Unable to query Elasticsearch"}, indent=2)
results = []
for hit in data["hits"]["hits"]:
source = hit["_source"]
results.append({
"startDate": source.get("startDate"),
"value": source.get("value"),
"device": source.get("device")
})
return json.dumps({
"total_records": data["hits"]["total"]["value"],
"data": results
}, indent=2)这些工具支持：
带日期、设备和聚合过滤的自定义查询
无过滤条件的快速最近数据检索
Prompts：不同 LLM 服务商的配置
我们的 MCP 服务器可与不同 LLM 服务商配合使用。为了方便集成，让我们实现一些 prompts，帮助 LLM 有效利用我们的资源和工具：
@mcp.prompt()
def daily_report(date: str = None) -> str:
"""Create a daily step report for a specific date"""
if date:
return f"""Please analyze the step data for {date}.
Provide:
1. Total steps
2. Average steps per active hour
3. Most active periods of the day
4. Comparison with weekly average
5. Graphical visualization of the data, if possible"""
else:
return """Please analyze the step data for today.
Provide:
1. Total steps so far
2. Average steps per active hour
3. Most active periods of the day
4. Comparison with weekly average
5. Graphical visualization of the data, if possible"""
@mcp.prompt()
def trend_analysis(start_date: str, end_date: str) -> str:
"""Analyze step trends over a specific period"""
return f"""Analyze step trends between {start_date} and {end_date}.
Please include:
1. Daily step trend graph
2. Identification of weekly patterns
3. Days with highest and lowest activity
4. Progression over time
5. Recommendations based on the data"""
@mcp.prompt()
def device_comparison() -> str:
"""Compare step data recorded by different devices"""
return """Compare step data recorded by Apple Watch and iPhone:
1. Which device records more steps on average?
2. Are there notable differences in usage patterns?
3. Times when each device is used more
4. Apparent accuracy of each device
5. Recommendations on which device to prioritize for tracking"""使用 MCP Inspector 测试服务器
在将服务器连接到真实的 LLM 客户端（比如 Claude）之前，最好先在受控环境中进行测试。mcp 库自带一个强大的基于网页的调试工具，叫做 MCP Inspector。它允许我们交互式测试服务器，实时查看请求和响应，验证逻辑，确保部署前无误。
如何启动 Inspector
在激活虚拟环境的终端中，运行以下命令，后接你的脚本名称：
mcp dev apple_watch_mcp.py你第一次运行时，mcp 可能会请求安装 Inspector 的网页组件。这是一次性设置。只需输入 y 然后按回车即可。
确认后，该工具会启动本地代理服务器，并自动在默认浏览器打开 MCP Inspector 界面，通常地址是 http://127.0.0.1:6274。
认证说明：新版 MCP Inspector 可能需要认证以保障安全。如果界面未自动打开，请查看终端中的提示信息，如：
Session token: xxxxx...
Open inspector with token pre-filled:
http://localhost:6274/?MCP_PROXY_AUTH_TOKEN=xxxxx...使用带有令牌的完整 URL 访问 Inspector。该令牌确保测试期间对服务器的安全访问。
界面介绍
MCP Inspector 提供了简洁的界面，方便你探索和交互服务器功能。
1）初始连接
页面加载时，会自动连接到你运行的 apple_watch_mcp.py 脚本。左侧状态会从未连接变为“Restart | Disconnect”，并且 “History” 面板会显示一个 initialize 事件。
2）发现你的功能
Inspector 的核心在顶部的标签页：Resources、Prompts 和 Tools。
- Resources：点击 “Resources” 标签页。点击 “List Resources” 按钮，可以查看你创建的所有数据端点，它们会显示在下方面板中。
- Tools 和 Prompts：你也可以对其它标签页执行同样操作。点击 “Tools” 再点击 “List Tools” 会显示你的 query_step_data 和 get_all_steps 函数。“Prompts” 标签页则会列出你预定义的提示模板。
3）运行测试
让我们测试一个简单的工具，看看它的运行效果。
进入 Tools 标签页
从列表中点击一个工具，比如 get_all_steps
右侧面板会显示该工具的详细信息。因为 get_all_steps 不需要参数，你只需点击 “Run Tool” 按钮即可。
点击“Run Tool”后，你服务器返回的 JSON 响应会显示在右侧面板中。你可以检查数据是否正确、格式是否符合预期，并查看是否有错误。对于需要参数的工具（如 query_step_data），填写参数后点击“Run”即可同样测试。
通过这些步骤，你可以测试 MCP 服务器的各个部分，确保其稳定并准备好与 Claude Desktop 集成。
停止测试服务器
mcp dev 命令启动的是用于调试的本地开发服务器。确认 Resources、Tools 和 Prompts 功能正常后，可以停止测试服务器。
方法是回到运行该进程的终端窗口，按 Ctrl+C。
...
Connected MCP client to backing server transport
Created client/server transports
Set up MCP proxy
^C服务器测试完成后，下一步是在 Claude Desktop 中安装它，由 Claude Desktop 管理服务器的运行。
添加到 LLM 客户端：最后一步
什么是 Claude Desktop？
为了更实用和直观地与 MCP 服务器交互，我们将使用 Claude Desktop 作为 MCP 客户端。与传统命令行工具不同，MCP 服务器需要兼容 Model Context Protocol 的客户端，不能直接运行命令。Claude Desktop 提供了这样的客户端界面，使我们可以发送 MCP 服务器可处理的自然语言查询。
安装：你可以在 Meet Claude \ Anthropic 免费下载 Claude Desktop。免费版本完全足够本文测试使用。
在 Claude 中安装服务器
安装 Claude Desktop 后，我们来注册 MCP 服务器：
# Install the server for Claude Desktop
mcp install apple_watch_mcp.py --name "Apple Health Steps"确认安装是否成功
现在，让我们确认命令是否按预期执行。mcp install 命令会向一个 JSON 配置文件添加条目。如果你使用了 API 密钥，配置中会包含相应的环境变量。运行以下命令查看该文件内容。
cat ~/Library/Application\ Support/Claude/claude_desktop_config.json你应该会在文件中看到类似如下结构，位于 "mcpServers" 键下：
{
"mcpServers": {
"Apple Health Steps": {
"command": "/full/path/to/your/uv",
"args": [
"--directory",
"/path/to/repository/apple-watch-health-mcp",
"run",
"apple_watch_mcp.py"
// Other arguments may appear depending on your setup
],
"env": {
"ES_API_KEY": "YOUR_ES_API_KEY"
}
}
// ... other servers might be listed here
}
}如果你找到了 “Apple Health Steps” 这一块，恭喜！你的服务器已成功配置。
重启提醒
重要：安装服务器后，需要重启 Claude Desktop 以加载新配置。
只有打开 Claude 时服务器才会启动。
如果出现执行错误，可以直接在 Claude Desktop 查看日志。
与 Claude 的交互示例
配置完成后，最有趣的部分来了：使用自然语言与你的 MCP 服务器对话。
开始之前：准备好你的环境
要与新服务器交互，需要进入客户端界面。打开 Claude Desktop 应用，确保你处于主聊天界面。
这里是你提问的地方。屏幕应如下图所示，文本框内已经有我们的第一个问题准备好。
一切准备就绪，开始吧。
示例 1：基础对话（RAG 实践）
现在，只需按 “Enter” 发送你输入的问题。这是 RAG 的一个很好示例，模型会检索特定数据进行回答。
你输入：
How many steps did I take last Thursday?期望结果：
Last Thursday (June 5th), you took 5,360 steps total. Your step count ranged from 450 to 2,340 steps across 4 different recordings throughout the day.
这很有用，但 MCP 的真正力量在于它不仅能返回文本。
例子 2：可视化你的活动（MCP 的力量）
现在，让我们请求服务器处理数据并以结构化格式返回，比如柱状图。
你输入：
Show me a summary of my step count from the last 7 days as a bar chart, including some insights on it.因为你的服务器理解意图并有工具来聚合数据，它不仅返回一句话。它生成结构化数据，以便客户端（ Claude ）可以渲染可视化。
恭喜！
你已成功创建一个功能性的 MCP 服务器，将 Claude 连接到你的健康数据，使用 Elasticsearch 作为后端，并支持自然语言交互。
总结
在这篇博客中，我们探讨了如何创建一个将 Apple HealthKit 数据与 LLM 代理集成的 MCP 服务器，使用 FastMCP 和 Elasticsearch。我们还展示了如何使用 MCP Inspector 和 Claude Desktop 本地测试服务器。代码在这里可用。
这种方法允许 AI 助手以上下文相关、安全且直观的方式访问和分析健康数据。
附加资源：
- MCP Python SDK 文档
- MCP 规范
- Elasticsearch 文档
- Apple HealthKit 文档
- 官方 Elasticsearch MCP 服务器
- FastMCP 文档
- 使用 asyncio
原文：Building an MCP server with Elasticsearch for real health data - Elasticsearch Labs