睡眠教练MCP服务器💤
一个专门的MCP（模型上下文协议）服务器，用于睡眠分析和推荐，内置于Python中。该服务器可以与任何支持MCP协议的聊天机器人或应用程序集成。
🌟 特性
- 睡眠常规分析：分析睡眠模式并检测问题
- 个性化推荐：根据特定用户数据提供建议
- 标准MCP协议：与任何MCP客户端兼容
- 跨平台：适用于Windows、macOS和Linux
- 易于集成：可以从任何编程语言连接
- 用户配置文件：通过时间类型分析创建详细的睡眠档案
- 周进度表：生成优化的每周睡眠时间表
- 快速建议：获取特定查询的即时睡眠提示
🔧 可用工具
create_user_profile
创建具有当前睡眠习惯的个性化用户配置文件。
参数：
user_id（string）：唯一用户标识符name（string）：用户名age（整数）：用户年龄chronotype（string）：“morning_lark”、“night_owl”或“intermediate”current_bedtime（string）：当前就寝时间，HH:MM格式current_wake_time（字符串）：HH:MM格式的当前唤醒时间sleep_duration_hours（数字）：平均睡眠时间goals（数组）：睡眠目标（例如，“更好的质量”、“更多的能量”）work_schedule（string）：工作进度说明screen_time_before_bed（整数）：睡前屏幕时间分钟数stress_level（整数）：压力水平从1到10sleep_quality_rating（整数）：自评睡眠质量1-10
analyze_sleep_pattern
分析用户当前的睡眠模式并检测问题。
参数：
user_id（string）：用户标识符
get_personalized_recommendations
根据用户配置文件生成个性化推荐。
参数：
user_id（string）：用户标识符
create_weekly_schedule
通过个性化例程创建优化的每周时间表。
参数：
user_id（string）：用户标识符
quick_sleep_advice
根据特定查询提供快速建议。
参数：
query（string）：特定睡眠相关问题user_context（字符串，可选）：其他用户上下文
📋 需求
- Python 3.8+
- 依赖关系：
mcp库（安装pip install mcp)
🚀 安装
选项1：直接下载
# Download the server
curl -O https://raw.githubusercontent.com/your-repo/sleep-coach-mcp/main/sleep_coach.py选项2：克隆存储库
git clone https://github.com/your-repo/sleep-coach-mcp
cd sleep-coach-mcp再进行
pip install mcp🔌 用法
运行服务器
python sleep_coach.py服务器使用stdio（标准输入/输出）进行MCP通信。
MCP协议通信
服务器实现MCP协议2024-11-05。以下是基本顺序：
1.初始化
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}
},
"clientInfo": {
"name": "your-client",
"version": "1.0.0"
}
}
}2.初始化通知
{
"jsonrpc": "2.0",
"method": "notifications/initialized",
"params": {}
}3.列出工具
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}4.调用工具
{
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "quick_sleep_advice",
"arguments": {
"query": "I can't fall asleep at night"
}
}
}💻 集成示例
Python客户端示例
import asyncio
import json
import subprocess
from mcp import ClientSession
from mcp.client.stdio import stdio_client
class SleepCoachClient:
def __init__(self):
self.server_process = None
self.session = None
async def start(self):
# Start the server process
self.server_process = subprocess.Popen(
['python', 'sleep_coach.py'],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# Create MCP session
read_stream, write_stream = stdio_client(self.server_process)
self.session = ClientSession(read_stream, write_stream)
# Initialize
await self.session.initialize()
print("Sleep Coach server started and initialized")
async def create_profile(self, profile_data):
"""Create a user profile"""
result = await self.session.call_tool(
"create_user_profile",
profile_data
)
return result.content[0].text
async def get_quick_advice(self, query):
"""Get quick sleep advice"""
result = await self.session.call_tool(
"quick_sleep_advice",
{"query": query}
)
return result.content[0].text
async def analyze_pattern(self, user_id):
"""Analyze user's sleep pattern"""
result = await self.session.call_tool(
"analyze_sleep_pattern",
{"user_id": user_id}
)
return result.content[0].text
async def get_recommendations(self, user_id):
"""Get personalized recommendations"""
result = await self.session.call_tool(
"get_personalized_recommendations",
{"user_id": user_id}
)
return result.content[0].text
async def stop(self):
"""Stop the server"""
if self.session:
await self.session.close()
if self.server_process:
self.server_process.terminate()
# Usage example
async def main():
client = SleepCoachClient()
await client.start()
try:
# Quick advice
advice = await client.get_quick_advice("I wake up tired every morning")
print("Advice:", advice)
# Create profile
profile = {
"user_id": "john123",
"name": "John Doe",
"age": 30,
"chronotype": "night_owl",
"current_bedtime": "00:30",
"current_wake_time": "08:00",
"sleep_duration_hours": 7.5,
"goals": ["better_quality", "more_energy"],
"work_schedule": "9-17",
"screen_time_before_bed": 90,
"stress_level": 6,
"sleep_quality_rating": 5
}
result = await client.create_profile(profile)
print("Profile created:", result)
# Get analysis and recommendations
analysis = await client.analyze_pattern("john123")
print("Analysis:", analysis)
recommendations = await client.get_recommendations("john123")
print("Recommendations:", recommendations)
finally:
await client.stop()
if __name__ == "__main__":
asyncio.run(main())Node.js客户端示例
const { spawn } = require('child_process');
const { EventEmitter } = require('events');
class SleepCoachClient extends EventEmitter {
constructor() {
super();
this.server = null;
this.requestId = 1;
this.pendingRequests = new Map();
}
async start() {
return new Promise((resolve, reject) => {
this.server = spawn('python', ['sleep_coach.py'], {
stdio: ['pipe', 'pipe', 'pipe']
});
this.server.stdout.on('data', (data) => {
const lines = data.toString().split('\n');
lines.forEach(line => {
if (line.trim()) {
try {
const response = JSON.parse(line);
this.handleResponse(response);
} catch (e) {
console.error('Parse error:', e, line);
}
}
});
});
this.server.stderr.on('data', (data) => {
console.error('Server error:', data.toString());
});
// Initialize
this.sendMessage({
jsonrpc: "2.0",
id: this.requestId++,
method: "initialize",
params: {
protocolVersion: "2024-11-05",
capabilities: { tools: {} },
clientInfo: { name: "node-client", version: "1.0.0" }
}
}).then(() => {
// Send initialized notification
this.sendNotification("notifications/initialized", {});
resolve();
}).catch(reject);
});
}
sendMessage(message) {
return new Promise((resolve, reject) => {
if (message.id) {
this.pendingRequests.set(message.id, { resolve, reject });
}
this.server.stdin.write(JSON.stringify(message) + '\n');
if (!message.id) resolve(); // For notifications
});
}
sendNotification(method, params) {
const message = {
jsonrpc: "2.0",
method: method,
params: params
};
this.server.stdin.write(JSON.stringify(message) + '\n');
}
handleResponse(response) {
if (response.id && this.pendingRequests.has(response.id)) {
const { resolve, reject } = this.pendingRequests.get(response.id);
this.pendingRequests.delete(response.id);
if (response.error) {
reject(new Error(response.error.message));
} else {
resolve(response.result);
}
}
}
async callTool(name, arguments) {
const result = await this.sendMessage({
jsonrpc: "2.0",
id: this.requestId++,
method: "tools/call",
params: { name, arguments }
});
return result.content[0].text;
}
async getQuickAdvice(query) {
return await this.callTool("quick_sleep_advice", { query });
}
async createProfile(profileData) {
return await this.callTool("create_user_profile", profileData);
}
stop() {
if (this.server) {
this.server.kill();
}
}
}
// Usage
(async () => {
const client = new SleepCoachClient();
await client.start();
try {
const advice = await client.getQuickAdvice("I can't sleep because of stress");
console.log('Sleep advice:', advice);
const profileResult = await client.createProfile({
user_id: "jane456",
name: "Jane Smith",
age: 28,
chronotype: "morning_lark",
current_bedtime: "22:00",
current_wake_time: "06:00",
sleep_duration_hours: 8,
goals: ["stress_reduction", "better_quality"],
work_schedule: "flexible",
screen_time_before_bed: 30,
stress_level: 8,
sleep_quality_rating: 4
});
console.log('Profile created:', profileResult);
} finally {
client.stop();
}
})();Claude桌面集成
添加到您的Claude Desktop配置中：
{
"mcpServers": {
"sleep-coach": {
"command": "python",
"args": ["/path/to/sleep_coach.py"]
}
}
}🛠️ 定制
修改建议
编辑 SleepCoachEngine 类在 sleep_coach.py:
def _load_sleep_knowledge(self) -> Dict:
"""Customize the sleep knowledge base"""
return {
"optimal_sleep_duration": {
"18-25": (7, 9),
"26-64": (7, 9),
"65+": (7, 8)
},
# Add your custom sleep rules here
"custom_recommendations": [
"Your custom sleep advice here",
# ...
]
}添加新工具
@app.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
# Add new tool handler
if name == "your_new_tool":
# Your custom logic here
result = "Your custom result"
return [types.TextContent(type="text", text=result)]
# ... existing handlers ...
@app.list_tools()
async def handle_list_tools() -> list[types.Tool]:
tools = [
# ... existing tools ...
types.Tool(
name="your_new_tool",
description="Description of your new tool",
inputSchema={
"type": "object",
"properties": {
"parameter": {"type": "string", "description": "Parameter description"}
},
"required": ["parameter"]
}
)
]
return tools🧪 测试
基本测试脚本
import asyncio
from your_client_implementation import SleepCoachClient
async def test_sleep_coach():
client = SleepCoachClient()
await client.start()
try:
# Test quick advice
print("Testing quick advice...")
advice = await client.get_quick_advice("I have trouble falling asleep")
assert "sleep" in advice.lower()
print("✅ Quick advice test passed")
# Test profile creation
print("Testing profile creation...")
profile = {
"user_id": "test_user",
"name": "Test User",
"age": 25,
"chronotype": "intermediate",
"current_bedtime": "23:00",
"current_wake_time": "07:00",
"sleep_duration_hours": 8,
"goals": ["better_quality"],
"work_schedule": "9-17"
}
result = await client.create_profile(profile)
assert "created" in result.lower()
print("✅ Profile creation test passed")
print("All tests passed! 🎉")
except Exception as e:
print(f"❌ Test failed: {e}")
finally:
await client.stop()
if __name__ == "__main__":
asyncio.run(test_sleep_coach())🐛 故障排除
常见问题
问题：“连接被拒绝”
- 解决方案： 确保已安装Python 3.8+，并且服务器脚本可执行
问题：“找不到模块'mcp'”
- 解决方案： 安装MCP库：
pip install mcp
问题：“服务器没有响应”
- 解决方案： 检查服务器日志是否有错误。服务器可能正在处理复杂的分析
问题：“找不到工具”
- 解决方案： 确保您发送
initialize使用前的消息tools/list或tools/call
调试模式
通过修改启用详细日志记录 sleep_coach.py:
import logging
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
# Add debug prints in tool handlers
print(f"Received tool call: {name}, args={arguments}")验证通信
import json
def log_mcp_message(direction, message):
print(f"MCP {direction}: {json.dumps(message, indent=2)}")
# Use in your client before sending/after receiving
log_mcp_message("SEND", message)
log_mcp_message("RECV", response)📚 睡眠科学背景
此服务器实施基于证据的睡眠建议：
- 计时码表：基于Michael Breus博士的研究
- 睡眠卫生：遵循睡眠基金会的指导方针
- 昼夜节律：纳入曝光和定时原则
- 睡眠结构：考虑REM和深度睡眠优化
🤝 贡献
- 分叉存储库
- 创建要素分支(
git checkout -b feature/new-feature) - 提交您的更改(
git commit -am 'Add new feature') - 推到分支(
git push origin feature/new-feature) - 创建拉取请求