Senechal MCP Server

#!/usr/bin/env python3 from mcp.server.fastmcp import FastMCP, Context import httpx import os import logging from datetime import datetime from dotenv import load_dotenv from typing import Optional, Dict, Any, List import json from dataclasses import dataclass # Set up logging log_dir = os.path.dirname(os.path.abspath(__file__)) log_file = os.path.join(log_dir, "senechal_mcp_server.log") logging.basicConfig( filename=log_file, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger("senechal_mcp") # Create stats tracking class ServerStats: def __init__(self): self.api_calls = 0 self.resource_requests = 0 self.tool_calls = 0 self.errors = 0 self.start_time = datetime.now() def log_status(self): uptime = datetime.now() - self.start_time hours, remainder = divmod(uptime.total_seconds(), 3600) minutes, seconds = divmod(remainder, 60) uptime_str = f"{int(hours)}h {int(minutes)}m {int(seconds)}s" logger.info(f"SERVER STATUS: Uptime: {uptime_str}, API Calls: {self.api_calls}, " f"Resource Requests: {self.resource_requests}, Tool Calls: {self.tool_calls}, " f"Errors: {self.errors}") # Initialize stats stats = ServerStats() # Load environment variables load_dotenv() logger.info("Starting Senechal Health MCP Server") # Create an MCP server mcp = FastMCP("Senechal Health MCP") # Constants API_BASE_URL = os.getenv("SENECHAL_API_BASE_URL") API_KEY = os.getenv("SENECHAL_API_KEY") if not API_BASE_URL: logger.error("No API URL found. Please set SENECHAL_API_BASE_URL environment variable in .env file") raise ValueError("SENECHAL_API_BASE_URL environment variable is required") if not API_KEY: logger.error("No API key found. Please set SENECHAL_API_KEY environment variable in .env file") raise ValueError("SENECHAL_API_KEY environment variable is required") logger.info(f"Using API URL: {API_BASE_URL}") logger.info(f"API Key: {'*' * len(API_KEY)}") # Helper class for response types @dataclass class HealthSummary: period_type: str summaries: List[Dict[str, Any]] generated_at: str @dataclass class HealthProfile: markdown: str @dataclass class Metrics: markdown: str # Helper functions async def make_api_request(endpoint: str, params: Optional[Dict[str, Any]] = None, expect_json: bool = True) -> Any: """Make a request to the Senechal API with the API key.""" headers = {"X-API-Key": API_KEY} # Log the request stats.api_calls += 1 logger.info(f"API Request: {endpoint} - Params: {params}") stats.log_status() try: async with httpx.AsyncClient() as client: response = await client.get(f"{API_BASE_URL}/{endpoint}", headers=headers, params=params) response.raise_for_status() logger.info(f"API Request successful: {endpoint}") if expect_json: return response.json() else: return response.text except Exception as e: stats.errors += 1 logger.error(f"API Request failed: {endpoint} - Error: {str(e)}") raise # Resources @mcp.resource(uri="senechal://health/availablemetrics") def get_available_metrics(): """Get a list of available health metrics in markdown format.""" async def impl(resource_uri, query_params): stats.resource_requests += 1 logger.info(f"Resource Request: health/availablemetrics - Params: {query_params}") try: # Call the actual API, expecting markdown text instead of JSON data = await make_api_request("health/availablemetrics", expect_json=False) # Return as a Metrics object with markdown field return Metrics(markdown=data) except Exception as e: logger.error(f"API error in health/availablemetrics: {str(e)}") # Provide fallback data in case API fails return Metrics(markdown="# Error\n\nUnable to retrieve available health metrics. Please check your API key and connection.") return impl @mcp.resource(uri="senechal://health/summary/{period}") def get_health_summary(period: str): """ Get a health summary for a specific period (day, week, month, year). Example: senechal://health/summary/day?metrics=all&span=7 """ async def impl(resource_uri, query_params): # Record stats stats.resource_requests += 1 logger.info(f"Resource Request: health/summary/{period} - Params: {query_params}") # Convert query parameters for the API request_params = {} if 'metrics' in query_params: request_params['metrics'] = query_params['metrics'] if 'span' in query_params: request_params['span'] = query_params['span'] if 'offset' in query_params: request_params['offset'] = query_params['offset'] # Call the actual API try: data = await make_api_request(f"health/summary/{period}", request_params) return data except Exception as e: logger.error(f"API error in health/summary/{period}: {str(e)}") # Provide fallback data in case API fails return { "error": f"API request failed: {str(e)}", "message": f"Unable to retrieve health summary data for period '{period}'. Please check your API key and connection." } return impl @mcp.resource(uri="senechal://health/profile") def get_health_profile(): """Get the user's health profile in markdown format.""" async def impl(resource_uri, query_params): stats.resource_requests += 1 logger.info(f"Resource Request: health/profile - Params: {query_params}") try: # Call the actual API, expecting markdown text instead of JSON data = await make_api_request("health/profile", expect_json=False) # Return as a HealthProfile object with markdown field return HealthProfile(markdown=data) except Exception as e: logger.error(f"API error in health/profile: {str(e)}") # Provide fallback data in case API fails return HealthProfile(markdown="# Error\n\nUnable to retrieve health profile data. Please check your API key and connection.") return impl # Tools @mcp.tool() async def fetch_available_metrics() -> str: """ Fetch a list of available health metrics in markdown format. Returns: A markdown string containing the available metrics data """ stats.tool_calls += 1 logger.info("Tool Call: fetch_available_metrics") try: # Call the actual API, expecting markdown text data = await make_api_request("health/availablemetrics", expect_json=False) return data except Exception as e: logger.error(f"API error in fetch_available_metrics: {str(e)}") # Provide fallback data in case API fails return "# Error\n\nUnable to retrieve available health metrics. Please check your API key and connection." @mcp.tool() async def fetch_health_summary(period: str, metrics: str = "all", span: int = 1, offset: int = 0) -> Dict[str, Any]: """ Fetch health summary for a specific period. Args: period: The period type (day, week, month, year) metrics: Comma-separated metrics/groups or 'all' span: Number of periods to return (1-52) offset: Number of periods to offset from now Returns: A dictionary containing the health summary data """ stats.tool_calls += 1 logger.info(f"Tool Call: fetch_health_summary - Args: period={period}, metrics={metrics}, span={span}, offset={offset}") # Parameters for the API request params = { 'metrics': metrics, 'span': span, 'offset': offset } try: # Call the actual API data = await make_api_request(f"health/summary/{period}", params) return data except Exception as e: logger.error(f"API error in fetch_health_summary: {str(e)}") # Provide fallback data in case API fails return { "error": f"API request failed: {str(e)}", "message": f"Unable to retrieve health summary data for period '{period}'. Please check your API key and connection." } @mcp.tool() async def fetch_health_profile() -> str: """ Fetch the user's health profile in markdown format. Returns: A markdown string containing the health profile data """ stats.tool_calls += 1 logger.info("Tool Call: fetch_health_profile") try: # Call the actual API, expecting markdown text data = await make_api_request("health/profile", expect_json=False) return data except Exception as e: logger.error(f"API error in fetch_health_profile: {str(e)}") # Provide fallback data in case API fails return "# Error\n\nUnable to retrieve health profile data. Please check your API key and connection." # Prompts @mcp.prompt() def analyze_health_summary() -> str: """ Create a prompt to analyze health summaries. """ logger.info("Prompt Requested: analyze_health_summary") return """ Please analyze the following health summary data and provide insights: 1. Identify any metrics that are outside of normal ranges 2. Highlight trends or patterns in the data 3. Suggest potential actions based on the data You can load the health data from: - senechal://health/summary/day?span=7 for daily data - senechal://health/summary/week?span=4 for weekly data - senechal://health/summary/month?span=3 for monthly data For available metrics information, check senechal://health/availablemetrics """ # Run the server if __name__ == "__main__": logger.info("Starting MCP server") mcp.run()