""" Tool definitions для Claude Haiku 4.5. Каждый tool описан схемой (Anthropic format) + обёрткой-executor. Большинство tools — тонкие прокси к /api/voice/tools/* на планшете. Таймер — тоже прокси, но POST (создаёт таймер, тот появляется на дашборде). Аутентификация: Bearer VOICE_API_KEY (тот же что для /api/voice/event). """ import os from typing import Any import requests from .config import log TABLET_URL = os.getenv("TABLET_URL", "").rstrip("/") VOICE_API_KEY = os.getenv("VOICE_API_KEY", "") _session = requests.Session() def _headers() -> dict: return {"Authorization": f"Bearer {VOICE_API_KEY}"} def _tablet_get(path: str, params: dict | None = None) -> dict: url = f"{TABLET_URL}{path}" r = _session.get(url, headers=_headers(), params=params, timeout=8) r.raise_for_status() return r.json() def _tablet_post(path: str, payload: dict) -> dict: url = f"{TABLET_URL}{path}" r = _session.post(url, headers=_headers(), json=payload, timeout=8) r.raise_for_status() return r.json() # ───────────────────────────────────────────────────────────── # Tool schemas (Anthropic format). Порядок = приоритет в Claude подсказках. # ───────────────────────────────────────────────────────────── TOOL_SCHEMAS: list[dict] = [ { "name": "get_weather", "description": ( "Получить текущую погоду и короткий прогноз для города. " "Для вопросов вроде 'какая сегодня погода', 'холодно ли на улице', " "'нужен ли зонт'. По умолчанию — Санкт-Петербург." ), "input_schema": { "type": "object", "properties": { "city": { "type": "string", "description": "Город на русском или шорткод (spb, msk, sochi, ekb, kzn, nsk, krd). По умолчанию Санкт-Петербург.", }, }, }, }, { "name": "get_transport", "description": ( "Расписание ближайших трамваев на остановке Ул. Антонова-Овсеенко. " "Для вопросов 'когда следующий 23-й', 'что ближайшее в центр', 'пора идти на остановку'." ), "input_schema": { "type": "object", "properties": { "direction": { "type": "string", "enum": ["to_center", "from_center", "all"], "description": "to_center = в центр (к Новочеркасской), from_center = от центра (к Большевиков), all = оба направления", }, "routes": { "type": "string", "description": "Фильтр маршрутов через запятую, например '23' или '23,27'. Пусто = все маршруты.", }, }, }, }, { "name": "get_today_events", "description": ( "События из календаря на сегодня или на неделю. " "Для вопросов 'что сегодня', 'какие планы', 'во сколько встреча'." ), "input_schema": { "type": "object", "properties": { "range": { "type": "string", "enum": ["today", "week"], "description": "today (по умолчанию) или week (7 дней)", }, }, }, }, { "name": "get_notes", "description": ( "Список заметок и списков покупок с планшета. " "Для 'что мне купить', 'что в списке', 'какие записи'." ), "input_schema": {"type": "object", "properties": {}}, }, { "name": "set_timer", "description": ( "Запустить таймер на планшете. Показывает обратный отсчёт с названием " "и звенит по окончании. " "Используй для 'поставь таймер на 10 минут', 'напомни через час', " "'засеки 5 минут для чайника'." ), "input_schema": { "type": "object", "properties": { "seconds": { "type": "integer", "description": "Длительность в секундах (1..86400)", "minimum": 1, "maximum": 86400, }, "label": { "type": "string", "description": "Короткое название таймера (например 'Чайник', 'Паста')", }, }, "required": ["seconds", "label"], }, }, { "name": "cancel_timer", "description": ( "Отменить активный таймер по его названию. " "Для 'отмени таймер чайник', 'убери таймер пасты', 'останови отсчёт'." ), "input_schema": { "type": "object", "properties": { "label": { "type": "string", "description": "Название таймера (примерное совпадение — можно частично).", }, }, "required": ["label"], }, }, { "name": "adjust_timer", "description": ( "Изменить оставшееся время таймера. " "Для 'добавь ещё 5 минут', 'убавь на минуту', 'накинь времени чайнику'. " "Положительный delta_seconds = добавить, отрицательный = уменьшить." ), "input_schema": { "type": "object", "properties": { "label": { "type": "string", "description": "Название таймера для которого меняем время.", }, "delta_seconds": { "type": "integer", "description": "Секунды (+ добавить, - уменьшить). Например 300 = +5 минут, -60 = -1 минута.", }, }, "required": ["label", "delta_seconds"], }, }, ] # ───────────────────────────────────────────────────────────── # Executors — возвращают JSON-совместимый dict или строку-ошибку. # ───────────────────────────────────────────────────────────── def _exec_get_weather(params: dict, agent_id: str) -> Any: city = params.get("city", "") return _tablet_get("/api/voice/tools/weather", params={"city": city} if city else None) def _exec_get_transport(params: dict, agent_id: str) -> Any: q = {} if "direction" in params: q["direction"] = params["direction"] if "routes" in params: q["routes"] = params["routes"] return _tablet_get("/api/voice/tools/transport", params=q) def _exec_get_today_events(params: dict, agent_id: str) -> Any: range_ = params.get("range", "today") return _tablet_get("/api/voice/tools/events", params={"range": range_}) def _exec_get_notes(params: dict, agent_id: str) -> Any: return _tablet_get("/api/voice/tools/notes") def _exec_set_timer(params: dict, agent_id: str) -> Any: seconds = int(params.get("seconds", 0)) label = params.get("label", "Таймер") if seconds < 1: return {"error": "seconds must be positive"} return _tablet_post( "/api/voice/timer", {"action": "start", "seconds": seconds, "label": label, "agent": agent_id}, ) def _exec_cancel_timer(params: dict, agent_id: str) -> Any: label = params.get("label", "").strip() if not label: return {"error": "label required"} return _tablet_post("/api/voice/timer", {"action": "cancel", "label": label}) def _exec_adjust_timer(params: dict, agent_id: str) -> Any: label = params.get("label", "").strip() delta = int(params.get("delta_seconds", 0)) if not label: return {"error": "label required"} if delta == 0: return {"error": "delta_seconds must be non-zero"} return _tablet_post( "/api/voice/timer", {"action": "adjust", "label": label, "delta_seconds": delta}, ) EXECUTORS = { "get_weather": _exec_get_weather, "get_transport": _exec_get_transport, "get_today_events": _exec_get_today_events, "get_notes": _exec_get_notes, "set_timer": _exec_set_timer, "cancel_timer": _exec_cancel_timer, "adjust_timer": _exec_adjust_timer, } def execute_tool(name: str, params: dict, agent_id: str = "cosmo") -> Any: """Выполнить tool по имени. Возвращает результат (dict/list/str). При ошибке возвращает {'error': '...'} — это отправляется в Claude как результат.""" fn = EXECUTORS.get(name) if fn is None: return {"error": f"unknown tool: {name}"} try: result = fn(params, agent_id) return result except requests.HTTPError as e: log.warning(f"Tool {name} HTTP {e.response.status_code}: {e.response.text[:200]}") return {"error": f"tool_http_{e.response.status_code}"} except requests.RequestException as e: log.warning(f"Tool {name} network error: {e}") return {"error": "tool_network_error"} except Exception as e: log.exception(f"Tool {name} failed") return {"error": f"tool_exception: {e}"}