tracker/mcp_server.py

"""MCP stdio server exposing tracker and planner operations as tools."""

import json
from mcp.server.fastmcp import FastMCP
from sqlmodel import Session

from tracker.agent_profile import resolve_agent_profile
from tracker.config import load_config
from tracker.config_updates import patch_active_program_id
from tracker.db import default_engine
from tracker.food_service import (
    append_meal_log,
    delete_planned_meal,
    list_meal_logs as list_meal_logs_svc,
    list_planned_meals as list_planned_meals_svc,
    upsert_planned_meal,
)
from tracker.models import MealLogCreateInput, MealPlanUpsertInput
from tracker.planner_schemas import ScheduleEventInput, ScheduleWorkoutsInput
from tracker.planner_service import (
    build_shopping_gap,
    build_tracker_state,
    create_daily_weighin,
    delete_calendar_event as delete_calendar_event_svc,
    fetch_calendar_view,
    fetch_pantry,
    run_schedule_event,
    run_schedule_workouts,
    search_recipes,
)
from tracker.training_schemas import LogSetInput, LogWorkoutInput, WeightInput
from tracker.training_service import (
    fetch_active_program_detail,
    fetch_today,
    list_exercise_catalog,
    list_program_summaries,
    log_body_weight,
    log_workout_session,
)

mcp = FastMCP("workout-tracker")


def _session() -> Session:
    return Session(default_engine())


@mcp.tool()
def get_agent_profile() -> str:
    return resolve_agent_profile(load_config()).model_dump_json()


@mcp.tool()
def get_today_workout(day_label: str, program_id: int | None = None) -> str:
    cfg = load_config()
    with _session() as session:
        data, err = fetch_today(session, cfg, day_label, program_id)
    if err is not None:
        return json.dumps({"error": err})
    return data.model_dump_json() if data else "{}"


@mcp.tool()
def log_workout(
    program_id: int,
    day_label: str,
    sets_json: str,
) -> str:
    raw = json.loads(sets_json)
    if not isinstance(raw, list):
        return json.dumps({"error": "sets_json must be a JSON array"})
    sets = [LogSetInput.model_validate(s) for s in raw]
    body = LogWorkoutInput(program_id=program_id, day_label=day_label, sets=sets)
    with _session() as session:
        count = log_workout_session(session, body)
    return json.dumps({"logged": count})


@mcp.tool()
def log_weight(weight_kg: float) -> str:
    body = WeightInput(weight_kg=weight_kg)
    with _session() as session:
        kg = log_body_weight(session, body)
    return json.dumps({"weight_kg": kg})


@mcp.tool()
def list_programs() -> str:
    with _session() as session:
        rows = list_program_summaries(session)
    return json.dumps([r.model_dump(mode="json") for r in rows])


@mcp.tool()
def get_active_program() -> str:
    cfg = load_config()
    with _session() as session:
        data, err = fetch_active_program_detail(session, cfg)
    if err is not None:
        return json.dumps({"error": err})
    return data.model_dump_json() if data else "{}"


@mcp.tool()
def set_active_program(program_id: int | None) -> str:
    patch_active_program_id(program_id)
    return json.dumps({"active_program_id": program_id})


@mcp.tool()
def list_exercises() -> str:
    with _session() as session:
        rows = list_exercise_catalog(session)
    return json.dumps([r.model_dump(mode="json") for r in rows])


@mcp.tool()
def get_calendar(days_ahead: int = 7) -> str:
    return fetch_calendar_view(days_ahead).model_dump_json()


@mcp.tool()
def schedule_workouts(
    days_ahead: int,
    duration_minutes: int,
    calendar_id: str | None = None,
    workout_starts_iso_json: str | None = None,
    allow_overlap_with_existing: bool = False,
) -> str:
    cfg = load_config()
    starts = None
    if workout_starts_iso_json is not None and workout_starts_iso_json.strip() != "":
        starts = json.loads(workout_starts_iso_json)
        if not isinstance(starts, list):
            return json.dumps({"error": "workout_starts_iso_json must be a JSON array of ISO datetime strings"})
    body = ScheduleWorkoutsInput(
        days_ahead=days_ahead,
        duration_minutes=duration_minutes,
        calendar_id=calendar_id,
        workout_starts_iso=starts,
        allow_overlap_with_existing=allow_overlap_with_existing,
    )
    with _session() as session:
        data, err = run_schedule_workouts(session, cfg, body)
    if err is not None:
        return json.dumps({"error": err})
    return json.dumps([e.model_dump(mode="json") for e in data]) if data else "[]"


@mcp.tool()
def schedule_event(
    summary: str,
    start: str,
    end: str,
    description: str,
    calendar_id: str | None = None,
) -> str:
    cfg = load_config()
    body = ScheduleEventInput(
        summary=summary,
        start=start,
        end=end,
        description=description,
        calendar_id=calendar_id,
    )
    return run_schedule_event(cfg, body).model_dump_json()


@mcp.tool()
def delete_calendar_event(event_id: str) -> str:
    cfg = load_config()
    return delete_calendar_event_svc(cfg, event_id).model_dump_json()


@mcp.tool()
def schedule_daily_weighin() -> str:
    cfg = load_config()
    return create_daily_weighin(cfg).model_dump_json()


@mcp.tool()
def search_recipes_tool(
    q: str,
    max_active_minutes: int | None = None,
    max_total_minutes: int | None = None,
    min_calories: int | None = None,
    max_calories: int | None = None,
    min_protein: float | None = None,
    limit: int = 20,
) -> str:
    cfg = load_config()
    profile = resolve_agent_profile(cfg)
    defaults = profile.recipe_search_defaults
    eff_max_active = max_active_minutes if max_active_minutes is not None else defaults.max_active_minutes
    eff_max_total = max_total_minutes if max_total_minutes is not None else defaults.max_total_minutes
    eff_min_cal = min_calories if min_calories is not None else defaults.min_calories
    eff_max_cal = max_calories if max_calories is not None else defaults.max_calories
    eff_min_prot = min_protein if min_protein is not None else defaults.min_protein
    eff_limit = defaults.search_limit if defaults.search_limit is not None else limit
    return search_recipes(
        cfg,
        q,
        eff_max_active,
        eff_max_total,
        eff_min_cal,
        eff_max_cal,
        eff_min_prot,
        eff_limit,
    ).model_dump_json()


@mcp.tool()
def get_pantry(user_id: int | None = None) -> str:
    cfg = load_config()
    profile = resolve_agent_profile(cfg)
    uid = user_id if user_id is not None else profile.pantry_user_id
    return fetch_pantry(cfg, uid).model_dump_json()


@mcp.tool()
def build_shopping_list(ingredients: str, user_id: int | None = None) -> str:
    cfg = load_config()
    profile = resolve_agent_profile(cfg)
    uid = user_id if user_id is not None else profile.pantry_user_id
    rows = build_shopping_gap(cfg, ingredients, uid)
    return json.dumps([r.model_dump(mode="json") for r in rows])


@mcp.tool()
def list_meal_plan(start_date: str, end_date: str) -> str:
    with _session() as session:
        rows = list_planned_meals_svc(session, start_date, end_date)
    return json.dumps([r.model_dump(mode="json") for r in rows])


@mcp.tool()
def upsert_meal_plan(meal_plan_json: str) -> str:
    body = MealPlanUpsertInput.model_validate_json(meal_plan_json)
    with _session() as session:
        row = upsert_planned_meal(session, body)
    return row.model_dump_json()


@mcp.tool()
def log_meal(meal_log_json: str) -> str:
    body = MealLogCreateInput.model_validate_json(meal_log_json)
    with _session() as session:
        row = append_meal_log(session, body)
    return row.model_dump_json()


@mcp.tool()
def list_meal_logs(start_date: str, end_date: str) -> str:
    with _session() as session:
        rows = list_meal_logs_svc(session, start_date, end_date)
    return json.dumps([r.model_dump(mode="json") for r in rows])


@mcp.tool()
def delete_planned_meal_tool(planned_meal_id: int) -> str:
    with _session() as session:
        ok = delete_planned_meal(session, planned_meal_id)
    return json.dumps({"deleted": ok})


@mcp.tool()
def get_tracker_state() -> str:
    cfg = load_config()
    with _session() as session:
        state = build_tracker_state(session, cfg)
    return state.model_dump_json()


def main() -> None:
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()