d1f95669e0
Two new Claude tools for voice control over existing timers.
Both accept a {label} (fuzzy match on tablet side, case-insensitive
substring) so the LLM doesn'\''t need to know internal timer ids.
- cancel_timer(label) → POST /api/voice/timer {action:cancel}
- adjust_timer(label, delta_seconds) → POST {action:adjust}
Use cases:
'\''Отмени таймер чайник'\'' → cancel_timer(label="чайник")
'\''Добавь ещё 5 минут к пасте'\'' → adjust_timer(label="паста", delta_seconds=300)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
263 lines
11 KiB
Python
263 lines
11 KiB
Python
"""
|
||
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}"}
|