import json from typing import Any from sqlalchemy.orm import Session from app.pomodoro.service import PomodoroService from app.projects.service import ProjectService 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": "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]) -> str: pomodoro = PomodoroService(db) projects = ProjectService(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": result = projects.sync_taiga_projects() 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"), ) 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)