home-voice-assistant/satellite/tools.py

"""
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"],
        },
    },
]


# ─────────────────────────────────────────────────────────────
# 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},
    )


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,
}


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}"}