SleepCoachServer by Fabiola-cc - MCP Server

Sleep Coach MCP Server 💤

A specialized MCP (Model Context Protocol) server for sleep analysis and recommendations, built in Python. This server can be integrated with any chatbot or application that supports the MCP protocol.

🌟 Features

Sleep routine analysis: Analyzes sleep patterns and detects issues
Personalized recommendations: Provides advice based on specific user data
Standard MCP protocol: Compatible with any MCP client
Cross-platform: Works on Windows, macOS, and Linux
Easy integration: Can be connected from any programming language
User profiles: Create detailed sleep profiles with chronotype analysis
Weekly scheduling: Generate optimized weekly sleep schedules
Quick advice: Get instant sleep tips for specific queries

🔧 Available Tools

`create_user_profile`

Creates a personalized user profile with current sleep habits.

Parameters:

user_id (string): Unique user identifier
name (string): User's name
age (integer): User's age
chronotype (string): "morning_lark", "night_owl", or "intermediate"
current_bedtime (string): Current bedtime in HH:MM format
current_wake_time (string): Current wake time in HH:MM format
sleep_duration_hours (number): Average sleep duration
goals (array): Sleep goals (e.g., "better_quality", "more_energy")
work_schedule (string): Work schedule description
screen_time_before_bed (integer): Minutes of screen time before bed
stress_level (integer): Stress level from 1-10
sleep_quality_rating (integer): Self-rated sleep quality from 1-10

`analyze_sleep_pattern`

Analyzes the user's current sleep pattern and detects problems.

Parameters:

user_id (string): User identifier

`get_personalized_recommendations`

Generates personalized recommendations based on user profile.

Parameters:

user_id (string): User identifier

`create_weekly_schedule`

Creates an optimized weekly schedule with personalized routines.

Parameters:

user_id (string): User identifier

`quick_sleep_advice`

Provides quick advice based on a specific query.

Parameters:

query (string): Specific sleep-related question
user_context (string, optional): Additional user context

📋 Requirements

Python 3.8+
Dependencies: mcp library (install with pip install mcp)

🚀 Installation

Option 1: Direct Download

# Download the server
curl -O https://raw.githubusercontent.com/your-repo/sleep-coach-mcp/main/sleep_coach.py

Option 2: Clone Repository

git clone https://github.com/your-repo/sleep-coach-mcp
cd sleep-coach-mcp

Install Dependencies

pip install mcp

🔌 Usage

Running the Server

python sleep_coach.py

The server runs using stdio (standard input/output) for MCP communication.

MCP Protocol Communication

The server implements MCP protocol 2024-11-05. Here's the basic sequence:

1. Initialization

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "tools": {}
    },
    "clientInfo": {
      "name": "your-client",
      "version": "1.0.0"
    }
  }
}

2. Initialized Notification

{
  "jsonrpc": "2.0",
  "method": "notifications/initialized",
  "params": {}
}

3. List Tools

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/list",
  "params": {}
}

4. Call Tool

{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "tools/call",
  "params": {
    "name": "quick_sleep_advice",
    "arguments": {
      "query": "I can't fall asleep at night"
    }
  }
}

💻 Integration Examples

Python Client Example

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 Client Example

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 Desktop Integration

Add to your Claude Desktop configuration:

{
  "mcpServers": {
    "sleep-coach": {
      "command": "python",
      "args": ["/path/to/sleep_coach.py"]
    }
  }
}

🛠️ Customization

Modifying Recommendations

Edit the SleepCoachEngine class in 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",
            # ...
        ]
    }

Adding New Tools

@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

🧪 Testing

Basic Test Script

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())

🐛 Troubleshooting

Common Issues

Problem: "Connection refused"

Solution: Ensure Python 3.8+ is installed and the server script is executable

Problem: "Module 'mcp' not found"

Solution: Install MCP library: pip install mcp

Problem: "Server not responding"

Solution: Check server logs for errors. The server may be processing complex analysis

Problem: "Tools not found"

Solution: Ensure you send the initialize message before using tools/list or tools/call

Debug Mode

Enable detailed logging by modifying 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}")

Validating Communication

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)

📚 Sleep Science Background

This server implements evidence-based sleep recommendations:

Chronotypes: Based on research by Dr. Michael Breus
Sleep hygiene: Following guidelines from the Sleep Foundation
Circadian rhythm: Incorporating light exposure and timing principles
Sleep architecture: Considering REM and deep sleep optimization

🤝 Contributing

Fork the repository
Create a feature branch (git checkout -b feature/new-feature)
Commit your changes (git commit -am 'Add new feature')
Push to the branch (git push origin feature/new-feature)
Create a Pull Request