"""Hevy Workout Analytics MCP Server."""
import os
import sqlite3
import sys
from pathlib import Path
from typing import Any
import yaml
from mcp.server import Server
from mcp.types import Tool, TextContent
from .schema import get_schema_description
# Configuration
DEFAULT_DB_PATH = Path.home() / "hevy.db"
CONFIG_DIR = Path(__file__).parent.parent / "config"
app = Server("hevy-history")
def get_db_path() -> Path:
"""Get database path from environment or default."""
db_path = os.environ.get("HEVY_DB_PATH", str(DEFAULT_DB_PATH))
return Path(db_path)
def get_config_file(filename: str) -> Path:
"""Get path to config file, checking environment override first."""
env_var = f"HEVY_{filename.upper().replace('.', '_')}"
custom_path = os.environ.get(env_var)
if custom_path:
return Path(custom_path)
return CONFIG_DIR / filename
def execute_readonly_sql(query: str) -> dict[str, Any]:
"""
Execute a read-only SQL query against the Hevy database.
Args:
query: SQL query to execute (SELECT statements only)
Returns:
Dictionary with 'columns' and 'rows' keys
Raises:
ValueError: If query contains non-SELECT statements
sqlite3.Error: If query execution fails
"""
# Safety check: only allow SELECT statements
query_upper = query.strip().upper()
dangerous_keywords = [
"INSERT",
"UPDATE",
"DELETE",
"DROP",
"CREATE",
"ALTER",
"TRUNCATE",
"REPLACE",
]
for keyword in dangerous_keywords:
if keyword in query_upper:
raise ValueError(
f"Only SELECT queries are allowed. Found forbidden keyword: {keyword}"
)
db_path = get_db_path()
if not db_path.exists():
raise FileNotFoundError(
f"Database not found at {db_path}. "
f"Run 'python scripts/import_csv.py <csv_file>' to create it."
)
# Connect with read-only mode
conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
conn.row_factory = sqlite3.Row
try:
cursor = conn.execute(query)
rows = cursor.fetchall()
# Convert to list of dicts
columns = [desc[0] for desc in cursor.description] if cursor.description else []
results = [dict(row) for row in rows]
return {"columns": columns, "rows": results, "row_count": len(results)}
finally:
conn.close()
def load_yaml_file(file_path: Path) -> dict:
"""Load and parse a YAML file."""
if not file_path.exists():
raise FileNotFoundError(f"Config file not found: {file_path}")
with open(file_path, "r", encoding="utf-8") as f:
return yaml.safe_load(f)
@app.list_tools()
async def list_tools() -> list[Tool]:
"""List available MCP tools."""
return [
Tool(
name="execute_sql",
description=(
"Execute a read-only SQL query against the Hevy workout database. "
"Use this to analyze workout history, find PRs, calculate volumes, "
"identify trends, etc. Only SELECT queries are allowed."
),
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": (
"SQL SELECT query to execute. "
"See get_schema() for table structure and column descriptions."
),
}
},
"required": ["query"],
},
),
Tool(
name="get_schema",
description=(
"Get the database schema with detailed column descriptions and query tips. "
"Use this first to understand the data structure before writing SQL queries."
),
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="get_exercise_taxonomy",
description=(
"Get the exercise taxonomy mapping exercises to muscle groups. "
"This includes primary/secondary muscles and compound vs isolation classification. "
"Use this to understand which exercises target which muscles for analysis."
),
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="get_tracking_conventions",
description=(
"Get personal tracking conventions including form resets, RPE usage, "
"deload periods, injuries, and other context. "
"Use this to provide accurate analysis that accounts for the user's specific training history."
),
inputSchema={"type": "object", "properties": {}},
),
]
@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
"""Handle tool calls."""
try:
if name == "execute_sql":
query = arguments.get("query")
if not query:
return [TextContent(type="text", text="Error: 'query' parameter is required")]
result = execute_readonly_sql(query)
# Format results as markdown table
if result["row_count"] == 0:
response = "Query executed successfully. No rows returned."
else:
lines = [f"Query returned {result['row_count']} rows.\n"]
# Create markdown table
if result["columns"]:
lines.append("| " + " | ".join(result["columns"]) + " |")
lines.append("| " + " | ".join(["---"] * len(result["columns"])) + " |")
for row in result["rows"][:100]: # Limit to first 100 rows in display
values = [str(row.get(col, "")) for col in result["columns"]]
lines.append("| " + " | ".join(values) + " |")
if result["row_count"] > 100:
lines.append(f"\n(Showing first 100 of {result['row_count']} rows)")
response = "\n".join(lines)
return [TextContent(type="text", text=response)]
elif name == "get_schema":
schema = get_schema_description()
return [TextContent(type="text", text=schema)]
elif name == "get_exercise_taxonomy":
taxonomy_path = get_config_file("taxonomy.yaml")
taxonomy = load_yaml_file(taxonomy_path)
# Format as readable text
lines = ["# Exercise Taxonomy\n"]
lines.append("## Muscle Groups:\n")
for group, subgroups in taxonomy.get("muscle_groups", {}).items():
lines.append(f"- **{group}**: {', '.join(subgroups)}")
lines.append("\n## Exercises:\n")
for exercise, info in taxonomy.get("exercises", {}).items():
lines.append(f"\n### {exercise}")
lines.append(f"- Category: {info.get('category', 'unknown')}")
lines.append(f"- Primary: {', '.join(info.get('primary', []))}")
secondary = info.get("secondary", [])
if secondary:
lines.append(f"- Secondary: {', '.join(secondary)}")
lines.append(
f"\n\n*Taxonomy loaded from: {taxonomy_path}*\n"
"*This file is editable - add your exercises and customize muscle group mappings.*"
)
return [TextContent(type="text", text="\n".join(lines))]
elif name == "get_tracking_conventions":
conventions_path = get_config_file("conventions.yaml")
conventions = load_yaml_file(conventions_path)
# Format as readable text
lines = ["# Personal Tracking Conventions\n"]
# Form resets
if "form_resets" in conventions and conventions["form_resets"]:
lines.append("## Form Resets:\n")
for reset in conventions["form_resets"]:
lines.append(
f"- **{reset['exercise']}** ({reset['date']}): {reset['reason']}"
)
# RPE usage
if "rpe_usage" in conventions:
lines.append("\n## RPE Usage:\n")
for note in conventions["rpe_usage"]:
lines.append(f"- {note}")
# Set types
if "set_type_usage" in conventions:
lines.append("\n## Set Type Usage:\n")
for note in conventions["set_type_usage"]:
lines.append(f"- {note}")
# Supersets
if "superset_usage" in conventions:
lines.append("\n## Superset Usage:\n")
for note in conventions["superset_usage"]:
lines.append(f"- {note}")
# Exercise notes patterns
if "exercise_notes_patterns" in conventions:
lines.append("\n## Exercise Notes Patterns:\n")
for note in conventions["exercise_notes_patterns"]:
lines.append(f"- {note}")
# Progression strategy
if "progression_strategy" in conventions:
lines.append("\n## Progression Strategy:\n")
for note in conventions["progression_strategy"]:
lines.append(f"- {note}")
# Deload periods
if "deload_periods" in conventions and conventions["deload_periods"]:
lines.append("\n## Deload Periods:\n")
for deload in conventions["deload_periods"]:
lines.append(
f"- {deload['date']} ({deload['duration_days']} days): {deload['reason']}"
)
# Injuries
if "injury_periods" in conventions and conventions["injury_periods"]:
lines.append("\n## Injury Periods:\n")
for injury in conventions["injury_periods"]:
affected = ", ".join(injury.get("affected_exercises", []))
lines.append(
f"- {injury['start_date']} to {injury['end_date']}: {injury['reason']}"
)
if affected:
lines.append(f" - Affected: {affected}")
# Equipment changes
if "equipment_changes" in conventions and conventions["equipment_changes"]:
lines.append("\n## Equipment Changes:\n")
for change in conventions["equipment_changes"]:
lines.append(f"- {change['date']}: {change['change']}")
if "affected_exercises" in change:
affected = ", ".join(change["affected_exercises"])
lines.append(f" - Affected: {affected}")
# Analysis notes
if "analysis_notes" in conventions:
lines.append("\n## Analysis Notes:\n")
for note in conventions["analysis_notes"]:
lines.append(f"- {note}")
lines.append(
f"\n\n*Conventions loaded from: {conventions_path}*\n"
"*This file is editable - add your personal tracking rules and context.*"
)
return [TextContent(type="text", text="\n".join(lines))]
else:
return [TextContent(type="text", text=f"Error: Unknown tool '{name}'")]
except Exception as e:
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def main():
"""Run the MCP server."""
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await app.run(read_stream, write_stream, app.create_initialization_options())
if __name__ == "__main__":
import asyncio
asyncio.run(main())
