import json from typing import Any from sqlalchemy.orm import Session from app.fitness.service import FitnessService from app.fitness.structuring import structure_meal, structure_workout from app.homelab.digest import build_weather_briefing from app.homelab.image_gen import generate_image as run_generate_image from app.homelab.openmeteo import OpenMeteoClient from app.integrations.openfoodfacts import OpenFoodFactsClient from app.integrations.wger import WgerClient from app.memory.service import MemoryService from app.pomodoro.service import PomodoroService from app.projects.service import ProjectService from app.reminders.service import RemindersService from app.shopping.service import ShoppingService TOOL_DEFINITIONS: list[dict[str, Any]] = [ { "type": "function", "function": { "name": "get_pomodoro_status", "description": "ОБЯЗАТЕЛЬНО вызывай перед любым ответом о таймере. Статус, фаза и прогресс цикла.", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "start_pomodoro", "description": "Запустить фазу работы в цикле помидоро (25 мин по умолчанию).", "parameters": { "type": "object", "properties": { "duration_min": {"type": "integer", "description": "Минуты работы"}, "task_note": {"type": "string", "description": "Над чем работаем"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "start_short_break", "description": "Запустить короткий перерыв между работами.", "parameters": { "type": "object", "properties": { "duration_min": {"type": "integer", "description": "Минуты перерыва"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "start_long_break", "description": "Запустить длинный перерыв после завершения цикла работ.", "parameters": { "type": "object", "properties": { "duration_min": {"type": "integer", "description": "Минуты перерыва"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "stop_pomodoro", "description": "Остановить текущую фазу таймера.", "parameters": { "type": "object", "properties": { "result": {"type": "string", "description": "Отчёт о сделанном"}, "completed": { "type": "boolean", "description": "True если фаза полностью завершена", }, }, "required": [], }, }, }, { "type": "function", "function": { "name": "skip_pomodoro_phase", "description": "Досрочно завершить текущую фазу и перейти к следующей в цикле.", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "reset_pomodoro_cycle", "description": "Сбросить цикл помидоро: обнулить счётчик работ и остановить таймер.", "parameters": { "type": "object", "properties": { "clear_task": { "type": "boolean", "description": "Также очистить текущую задачу", }, }, "required": [], }, }, }, { "type": "function", "function": { "name": "get_pomodoro_history", "description": "История помидоро-сессий (таймер), не Taiga-задачи.", "parameters": { "type": "object", "properties": { "limit": {"type": "integer", "description": "Сколько сессий вернуть"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "sync_taiga_projects", "description": "Синхронизировать список проектов из Taiga API. Вызывай если проекты неизвестны.", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "list_taiga_projects", "description": "Список проектов Taiga с привязкой Gitea.", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "list_taiga_tasks", "description": ( "ОБЯЗАТЕЛЬНО при вопросах «какие задачи», «покажи задачи проекта», «что открыто в Taiga». " "Живые user stories и tasks из Taiga API. НЕ путать с list_work_items." ), "parameters": { "type": "object", "properties": { "project_slug": { "type": "string", "description": "Slug проекта, например home_assistant. Пусто = все проекты.", }, "limit": {"type": "integer", "description": "Макс. на проект, по умолчанию 20"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "create_work_item", "description": ( "Создать фичу/баг из вольного текста: структурировать через LLM, " "создать Taiga story + Gitea issue. Вызывай при «заведи баг», «оформи фичу», «добавь в таигу»." ), "parameters": { "type": "object", "properties": { "text": {"type": "string", "description": "Полное описание от пользователя"}, "project_slug": { "type": "string", "description": "Slug проекта Taiga, если известен", }, }, "required": ["text"], }, }, }, { "type": "function", "function": { "name": "remember_fact", "description": ( "Сохранить факт в долгосрочную память. " "Когда пользователь просит «запомни», или сообщает устойчивое предпочтение/факт." ), "parameters": { "type": "object", "properties": { "content": {"type": "string", "description": "Что запомнить"}, "category": { "type": "string", "description": "preference, person, habit, project, fact", }, "importance": {"type": "integer", "description": "1-5, по умолчанию 3"}, }, "required": ["content"], }, }, }, { "type": "function", "function": { "name": "recall_memories", "description": ( "Поиск в долгосрочной памяти. " "Когда спрашивают «что ты помнишь», «что я говорил про X»." ), "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Подстрока для поиска"}, "category": {"type": "string"}, "limit": {"type": "integer"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "forget_memory", "description": "Удалить (деактивировать) факт по id из recall_memories или снимка памяти.", "parameters": { "type": "object", "properties": { "memory_id": {"type": "integer"}, }, "required": ["memory_id"], }, }, }, { "type": "function", "function": { "name": "update_profile", "description": ( "Обновить профиль пользователя: name, timezone, language, notes. " "Передавай только изменившиеся поля." ), "parameters": { "type": "object", "properties": { "name": {"type": "string"}, "age": {"type": "string", "description": "Возраст пользователя"}, "timezone": {"type": "string"}, "language": {"type": "string"}, "notes": {"type": "string"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "update_session_summary", "description": ( "Сохранить краткую сводку темы текущего чата " "(когда диалог длинный или пользователь просит «сожми контекст»)." ), "parameters": { "type": "object", "properties": { "summary": {"type": "string", "description": "2-5 предложений о теме чата"}, "session_id": {"type": "integer"}, }, "required": ["summary", "session_id"], }, }, }, { "type": "function", "function": { "name": "get_fitness_summary", "description": "Сводка фитнеса за сегодня: ккал, БЖУ, вода, тренировки.", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "set_fitness_profile", "description": "Настроить фитнес-профиль и пересчитать цели ккал/БЖУ/воды.", "parameters": { "type": "object", "properties": { "sex": {"type": "string", "description": "male/female"}, "age": {"type": "integer"}, "height_cm": {"type": "number"}, "weight_kg": {"type": "number"}, "activity_level": { "type": "string", "description": "sedentary/light/moderate/active/very_active", }, "goal": {"type": "string", "description": "lose/maintain/gain"}, "target_weight_kg": {"type": "number"}, "weekly_workouts": {"type": "integer"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "calc_fitness_targets", "description": "Калькулятор BMR/TDEE/макросов без сохранения.", "parameters": { "type": "object", "properties": { "sex": {"type": "string"}, "age": {"type": "integer"}, "height_cm": {"type": "number"}, "weight_kg": {"type": "number"}, "activity_level": {"type": "string"}, "goal": {"type": "string"}, }, "required": ["weight_kg", "height_cm", "age"], }, }, }, { "type": "function", "function": { "name": "log_meal", "description": "Записать приём пищи. LLM оценит ккал и БЖУ из текста.", "parameters": { "type": "object", "properties": { "text": {"type": "string", "description": "Что съел"}, "meal_type": {"type": "string"}, }, "required": ["text"], }, }, }, { "type": "function", "function": { "name": "log_water", "description": "Записать воду в мл.", "parameters": { "type": "object", "properties": { "amount_ml": {"type": "integer"}, }, "required": ["amount_ml"], }, }, }, { "type": "function", "function": { "name": "log_weight", "description": "Записать вес в кг.", "parameters": { "type": "object", "properties": { "weight_kg": {"type": "number"}, "body_fat_pct": {"type": "number"}, "notes": {"type": "string"}, }, "required": ["weight_kg"], }, }, }, { "type": "function", "function": { "name": "log_workout", "description": "Записать тренировку из текста.", "parameters": { "type": "object", "properties": { "text": {"type": "string"}, }, "required": ["text"], }, }, }, { "type": "function", "function": { "name": "lookup_food", "description": "Поиск продукта в Open Food Facts (ккал на 100г).", "parameters": { "type": "object", "properties": { "query": {"type": "string"}, "limit": {"type": "integer"}, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "lookup_exercise", "description": "Поиск упражнения в базе wger.", "parameters": { "type": "object", "properties": { "query": {"type": "string"}, "limit": {"type": "integer"}, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "set_fitness_reminder", "description": "Вкл/выкл или настроить напоминание: water, meal, workout, weigh_in.", "parameters": { "type": "object", "properties": { "kind": {"type": "string"}, "enabled": {"type": "boolean"}, "hour": {"type": "integer"}, "minute": {"type": "integer"}, "interval_hours": {"type": "integer"}, }, "required": ["kind"], }, }, }, { "type": "function", "function": { "name": "get_weather", "description": ( "ОБЯЗАТЕЛЬНО для вопросов о погоде, «что на улице», «будет ли дождь». " "Текущая погода и прогноз по часам." ), "parameters": { "type": "object", "properties": { "hours_ahead": { "type": "integer", "description": "Сколько часов прогноза (по умолчанию 12)", }, }, "required": [], }, }, }, { "type": "function", "function": { "name": "get_morning_briefing", "description": "Утренний брифинг: погода и заголовки новостей.", "parameters": { "type": "object", "properties": { "include_news": { "type": "boolean", "description": "Включить новости (по умолчанию true)", }, }, "required": [], }, }, }, { "type": "function", "function": { "name": "generate_image", "description": ( "Аниме-картинка (Anima через RP-чат). " "«Нарисуй себя» / портрет персонажа → draw_self=true. " "Другая сцена → scene_description на английском (booru-теги). " "Внешность берётся из карточки персонажа. Только по запросу или когда уместно." ), "parameters": { "type": "object", "properties": { "draw_self": { "type": "boolean", "description": "Нарисовать персонажа из карточки в контексте текущего чата", }, "scene_description": { "type": "string", "description": "Описание сцены на английском (booru-теги), если не draw_self", }, }, "required": [], }, }, }, { "type": "function", "function": { "name": "list_shopping_lists", "description": "Все списки покупок с позициями. «Что купить», «покажи списки».", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "create_shopping_list", "description": "Создать новый список покупок.", "parameters": { "type": "object", "properties": { "name": {"type": "string", "description": "Название списка, например «Продукты»"}, }, "required": ["name"], }, }, }, { "type": "function", "function": { "name": "add_shopping_items", "description": "Добавить товары в список. Список создаётся, если не существует.", "parameters": { "type": "object", "properties": { "list_name": {"type": "string", "description": "Название списка"}, "list_id": {"type": "integer"}, "items": { "type": "array", "items": { "type": "object", "properties": { "text": {"type": "string"}, "quantity": {"type": "number"}, "unit": {"type": "string"}, }, "required": ["text"], }, }, }, "required": ["items"], }, }, }, { "type": "function", "function": { "name": "check_shopping_item", "description": "Отметить позицию как купленную (checked=true) или снять отметку (false).", "parameters": { "type": "object", "properties": { "item_id": {"type": "integer"}, "checked": {"type": "boolean"}, }, "required": ["item_id", "checked"], }, }, }, { "type": "function", "function": { "name": "remove_shopping_item", "description": "Удалить позицию из списка по item_id.", "parameters": { "type": "object", "properties": {"item_id": {"type": "integer"}}, "required": ["item_id"], }, }, }, { "type": "function", "function": { "name": "delete_shopping_list", "description": "Удалить весь список покупок.", "parameters": { "type": "object", "properties": {"list_id": {"type": "integer"}}, "required": ["list_id"], }, }, }, { "type": "function", "function": { "name": "list_reminders", "description": "Список активных напоминаний. «Что напомнил», «мои напоминания».", "parameters": { "type": "object", "properties": { "limit": {"type": "integer", "description": "Макс. записей, по умолчанию 20"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "create_reminder", "description": ( "Создать напоминание. due_at — ISO datetime в часовом поясе пользователя " "(см. [Текущее время]). Примеры: через 15 мин, завтра 09:00, 2027-05-12T12:16:00." ), "parameters": { "type": "object", "properties": { "title": {"type": "string", "description": "О чём напомнить"}, "due_at": {"type": "string", "description": "ISO datetime"}, "notes": {"type": "string"}, "all_day": {"type": "boolean"}, "recurrence": { "type": "string", "enum": ["none", "daily", "weekly", "monthly"], "description": "Повтор", }, }, "required": ["title", "due_at"], }, }, }, { "type": "function", "function": { "name": "update_reminder", "description": "Изменить напоминание по id.", "parameters": { "type": "object", "properties": { "reminder_id": {"type": "integer"}, "title": {"type": "string"}, "due_at": {"type": "string"}, "notes": {"type": "string"}, "all_day": {"type": "boolean"}, "recurrence": {"type": "string", "enum": ["none", "daily", "weekly", "monthly"]}, "enabled": {"type": "boolean"}, }, "required": ["reminder_id"], }, }, }, { "type": "function", "function": { "name": "delete_reminder", "description": "Удалить напоминание по id.", "parameters": { "type": "object", "properties": {"reminder_id": {"type": "integer"}}, "required": ["reminder_id"], }, }, }, { "type": "function", "function": { "name": "complete_reminder", "description": "Отметить напоминание выполненным (снять с календаря).", "parameters": { "type": "object", "properties": {"reminder_id": {"type": "integer"}}, "required": ["reminder_id"], }, }, }, { "type": "function", "function": { "name": "list_work_items", "description": ( "Только задачи, созданные ЭТИМ ассистентом через create_work_item (локальная БД). " "НЕ использовать для общего вопроса «какие задачи в Taiga» — для того list_taiga_tasks." ), "parameters": { "type": "object", "properties": { "status": {"type": "string", "description": "open или closed"}, "limit": {"type": "integer"}, }, "required": [], }, }, }, ] async def execute_tool( db: Session, name: str, arguments: dict[str, Any], *, session_id: int | None = None, ) -> str: pomodoro = PomodoroService(db) projects = ProjectService(db) memory = MemoryService(db) fitness = FitnessService(db) shopping = ShoppingService(db) reminders = RemindersService(db) try: if name == "get_pomodoro_status": result = pomodoro.get_status() elif name == "start_pomodoro": result = pomodoro.start_work( duration_min=arguments.get("duration_min"), task_note=arguments.get("task_note", ""), ) elif name == "start_short_break": result = pomodoro.start_short_break(duration_min=arguments.get("duration_min")) elif name == "start_long_break": result = pomodoro.start_long_break(duration_min=arguments.get("duration_min")) elif name == "stop_pomodoro": result = pomodoro.stop( result=arguments.get("result", ""), completed=arguments.get("completed", False), ) elif name == "skip_pomodoro_phase": result = pomodoro.skip_phase() elif name == "reset_pomodoro_cycle": result = pomodoro.reset_cycle(clear_task=arguments.get("clear_task", False)) elif name == "get_pomodoro_history": result = pomodoro.history(limit=arguments.get("limit", 10)) elif name == "sync_taiga_projects": from app.projects.context import invalidate_projects_snapshot_cache result = projects.sync_taiga_projects() invalidate_projects_snapshot_cache() elif name == "list_taiga_projects": result = projects.list_projects() elif name == "list_taiga_tasks": result = projects.list_taiga_open_tasks( project_slug=arguments.get("project_slug"), limit=arguments.get("limit", 20), ) elif name == "create_work_item": result = await projects.create_work_item( arguments.get("text", ""), project_slug=arguments.get("project_slug"), ) elif name == "list_work_items": result = projects.list_work_items( limit=arguments.get("limit", 20), status=arguments.get("status"), ) elif name == "remember_fact": result = memory.remember_fact( arguments.get("content", ""), category=arguments.get("category", "fact"), importance=arguments.get("importance", 3), session_id=session_id, source="tool", ) elif name == "recall_memories": result = memory.recall_memories( query=arguments.get("query"), category=arguments.get("category"), limit=arguments.get("limit", 20), ) elif name == "forget_memory": result = memory.forget_memory(int(arguments["memory_id"])) elif name == "update_profile": updates = { k: arguments[k] for k in ("name", "age", "timezone", "language", "notes") if k in arguments and arguments[k] is not None } result = memory.update_profile(updates) elif name == "update_session_summary": result = memory.update_session_summary( int(arguments["session_id"]), arguments.get("summary", ""), ) elif name == "get_fitness_summary": result = fitness.get_daily_summary() elif name == "set_fitness_profile": updates = { k: arguments[k] for k in ( "sex", "age", "height_cm", "weight_kg", "activity_level", "goal", "target_weight_kg", "weekly_workouts", ) if k in arguments and arguments[k] is not None } result = fitness.set_profile(updates) elif name == "calc_fitness_targets": result = fitness.calc_targets(arguments) elif name == "log_meal": structured = await structure_meal(arguments.get("text", "")) result = fitness.log_meal( description=structured.get("description") or arguments.get("text", ""), meal_type=arguments.get("meal_type") or structured.get("meal_type") or "snack", calories=float(structured.get("calories") or 0), protein_g=float(structured.get("protein_g") or 0), fat_g=float(structured.get("fat_g") or 0), carbs_g=float(structured.get("carbs_g") or 0), source="llm", estimated=True, ) elif name == "log_water": result = fitness.log_water(int(arguments.get("amount_ml", 250))) elif name == "log_weight": result = fitness.log_weight( float(arguments["weight_kg"]), body_fat_pct=arguments.get("body_fat_pct"), notes=arguments.get("notes", ""), ) elif name == "log_workout": structured = await structure_workout(arguments.get("text", "")) result = fitness.log_workout( title=structured.get("title") or "Тренировка", notes=structured.get("notes") or arguments.get("text", ""), duration_min=structured.get("duration_min"), exercises=structured.get("exercises"), ) elif name == "lookup_food": result = OpenFoodFactsClient().search( arguments.get("query", ""), limit=arguments.get("limit", 5), ) elif name == "lookup_exercise": result = WgerClient().search_exercises( arguments.get("query", ""), limit=arguments.get("limit", 8), ) elif name == "set_fitness_reminder": result = fitness.set_reminder( arguments.get("kind", "water"), enabled=arguments.get("enabled"), hour=arguments.get("hour"), minute=arguments.get("minute"), interval_hours=arguments.get("interval_hours"), ) elif name == "get_weather": hours = int(arguments.get("hours_ahead") or 12) client = OpenMeteoClient() weather = client.fetch_current_and_hourly(hours_ahead=hours) result = { "weather": weather, "rain_summary": client.rain_summary(hours_ahead=hours) if weather.get("ok") else "", } elif name == "get_morning_briefing": include_news = arguments.get("include_news", True) result = build_weather_briefing( hours_ahead=12, include_news=bool(include_news), ) elif name == "generate_image": result = await run_generate_image( db, session_id=session_id, draw_self=bool(arguments.get("draw_self")), scene_description=arguments.get("scene_description", ""), ) elif name == "list_shopping_lists": result = shopping.list_lists(include_items=True) elif name == "create_shopping_list": result = shopping.create_list(arguments.get("name", "")) elif name == "add_shopping_items": result = shopping.add_items( arguments.get("items") or [], list_id=arguments.get("list_id"), list_name=arguments.get("list_name"), ) elif name == "check_shopping_item": result = shopping.set_item_checked( int(arguments["item_id"]), bool(arguments.get("checked", True)), ) elif name == "remove_shopping_item": result = shopping.remove_item(int(arguments["item_id"])) elif name == "delete_shopping_list": result = shopping.delete_list(int(arguments["list_id"])) elif name == "list_reminders": result = reminders.list_upcoming(limit=int(arguments.get("limit") or 20)) elif name == "create_reminder": result = reminders.create( title=arguments.get("title", ""), due_at=arguments.get("due_at", ""), notes=arguments.get("notes", ""), all_day=bool(arguments.get("all_day", False)), recurrence=arguments.get("recurrence", "none"), ) elif name == "update_reminder": result = reminders.update( int(arguments["reminder_id"]), title=arguments.get("title"), due_at=arguments.get("due_at"), notes=arguments.get("notes"), all_day=arguments.get("all_day"), recurrence=arguments.get("recurrence"), enabled=arguments.get("enabled"), ) elif name == "delete_reminder": result = reminders.delete(int(arguments["reminder_id"])) elif name == "complete_reminder": result = reminders.complete(int(arguments["reminder_id"])) else: return json.dumps({"error": f"Unknown tool: {name}"}, ensure_ascii=False) return json.dumps(result, ensure_ascii=False) except ValueError as exc: return json.dumps({"error": str(exc)}, ensure_ascii=False) except Exception as exc: return json.dumps({"error": str(exc)}, ensure_ascii=False)