""" 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": ( "События из календаря (Даниил + Света). Вернёт id события, " "title, start, end, owner ('daniil' или 'sveta'). " "ВАЖНО: для update_event / delete_event сначала вызывай этот tool " "чтобы получить event_id." ), "input_schema": { "type": "object", "properties": { "range": { "type": "string", "enum": ["today", "week", "month"], "description": "today (по умолчанию), week (7 дней) или month (текущий месяц)", }, }, }, }, { "name": "create_event", "description": ( "Создать событие в Google Calendar. " "ВАЖНО: параметр owner обязателен. Если пользователь не сказал " "чей это календарь — СПРОСИ у него ('в твой календарь или в Светин?') " "и только потом вызывай tool. Не угадывай." ), "input_schema": { "type": "object", "properties": { "title": {"type": "string", "description": "Название события"}, "date": {"type": "string", "description": "Дата в формате YYYY-MM-DD"}, "start_time": { "type": "string", "description": "Время начала в формате HH:MM (24-часовой). Обязательно если all_day=false.", }, "end_time": { "type": "string", "description": "Время окончания в формате HH:MM. По умолчанию start_time + 1 час.", }, "all_day": { "type": "boolean", "description": "Событие на весь день без времени. По умолчанию false.", }, "owner": { "type": "string", "enum": ["daniil", "sveta"], "description": "Чей это календарь — Даниила или Светы", }, }, "required": ["title", "date", "owner"], }, }, { "name": "update_event", "description": ( "Изменить существующее событие. Сначала обязательно вызови get_today_events " "чтобы получить event_id и owner нужного события. Передавай только те поля " "которые меняешь." ), "input_schema": { "type": "object", "properties": { "event_id": {"type": "string"}, "owner": { "type": "string", "enum": ["daniil", "sveta"], "description": "Чей календарь (из get_today_events)", }, "title": {"type": "string"}, "date": {"type": "string", "description": "YYYY-MM-DD"}, "start_time": {"type": "string", "description": "HH:MM"}, "end_time": {"type": "string", "description": "HH:MM"}, "all_day": {"type": "boolean"}, }, "required": ["event_id", "owner"], }, }, { "name": "delete_event", "description": ( "Удалить событие из календаря. Сначала вызови get_today_events чтобы " "найти event_id и определить owner. Подтверди удаление с пользователем " "если событие важное (встреча, врач, работа)." ), "input_schema": { "type": "object", "properties": { "event_id": {"type": "string"}, "owner": { "type": "string", "enum": ["daniil", "sveta"], }, }, "required": ["event_id", "owner"], }, }, { "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_create_event(params: dict, agent_id: str) -> Any: payload = { "title": params.get("title", "").strip(), "date": params.get("date", "").strip(), "owner": params.get("owner", "daniil"), "all_day": bool(params.get("all_day", False)), } if not payload["title"] or not payload["date"]: return {"error": "title and date required"} if not payload["all_day"]: payload["start_time"] = params.get("start_time", "") if "end_time" in params: payload["end_time"] = params.get("end_time", "") return _tablet_post("/api/voice/tools/events", payload) def _exec_update_event(params: dict, agent_id: str) -> Any: event_id = params.get("event_id", "").strip() owner = params.get("owner", "").strip() if not event_id or not owner: return {"error": "event_id and owner required"} payload = {"event_id": event_id, "owner": owner} for k in ("title", "date", "start_time", "end_time", "all_day"): if k in params: payload[k] = params[k] url = f"{TABLET_URL}/api/voice/tools/events" r = _session.put(url, headers={**_headers(), "Content-Type": "application/json"}, json=payload, timeout=8) r.raise_for_status() return r.json() def _exec_delete_event(params: dict, agent_id: str) -> Any: event_id = params.get("event_id", "").strip() owner = params.get("owner", "daniil").strip() if not event_id: return {"error": "event_id required"} url = f"{TABLET_URL}/api/voice/tools/events" r = _session.delete( url, headers=_headers(), params={"event_id": event_id, "owner": owner}, timeout=8, ) r.raise_for_status() return r.json() 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, "create_event": _exec_create_event, "update_event": _exec_update_event, "delete_event": _exec_delete_event, "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}"}