963 lines
39 KiB
Python
963 lines
39 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""MCP tool definitions and handlers."""
|
|
|
|
import json
|
|
from typing import Any, Optional
|
|
|
|
from .. import audit
|
|
from ..context import store as context_store
|
|
from ..jwt_utils import has_scope
|
|
from ..knowledge import apps_indexer, gitea, gitea_indexer, indexer, paperless
|
|
from ..integrations import casa, homeassistant, irrigazione, turni
|
|
|
|
|
|
def tool_definitions() -> list[dict]:
|
|
return [
|
|
{
|
|
"name": "ping",
|
|
"description": "Verifica che il server MCP Loogle risponda",
|
|
"inputSchema": {"type": "object", "properties": {}},
|
|
},
|
|
{
|
|
"name": "whoami",
|
|
"description": "Restituisce l'utente autenticato e gli scope attivi",
|
|
"inputSchema": {"type": "object", "properties": {}},
|
|
},
|
|
{
|
|
"name": "list_projects",
|
|
"description": "Elenca i progetti dell'utente corrente",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {"include_archived": {"type": "boolean", "default": False}},
|
|
},
|
|
},
|
|
{
|
|
"name": "create_project",
|
|
"description": "Crea un nuovo progetto con archivio contesto; opzionalmente collega un repo Gitea",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["title"],
|
|
"properties": {
|
|
"title": {"type": "string"},
|
|
"tags": {"type": "array", "items": {"type": "string"}},
|
|
"gitea_repo": {
|
|
"type": "string",
|
|
"description": "Repository Gitea owner/name (es. daniele/rete)",
|
|
},
|
|
"seed_from_gitea": {
|
|
"type": "boolean",
|
|
"default": False,
|
|
"description": "Importa README.md nel context.md se collegato a Gitea",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "link_project_repo",
|
|
"description": "Collega o scollega un repository Gitea da un progetto esistente",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["project_id"],
|
|
"properties": {
|
|
"project_id": {"type": "string"},
|
|
"gitea_repo": {
|
|
"type": "string",
|
|
"description": "owner/name da collegare; omit o stringa vuota per scollegare",
|
|
},
|
|
"seed_from_gitea": {
|
|
"type": "boolean",
|
|
"default": False,
|
|
"description": "Importa README.md nel context.md (solo se non già importato)",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "get_project_context",
|
|
"description": "Recupera meta, context.md, sessioni recenti e arricchimento Gitea del progetto",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["project_id"],
|
|
"properties": {
|
|
"project_id": {"type": "string"},
|
|
"session_limit": {"type": "integer", "default": 5},
|
|
"include_gitea": {
|
|
"type": "boolean",
|
|
"default": True,
|
|
"description": "Include README/docs dal repo Gitea collegato",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "save_context",
|
|
"description": "Salva o appende memoria persistente in un progetto",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["project_id", "content"],
|
|
"properties": {
|
|
"project_id": {"type": "string"},
|
|
"content": {"type": "string"},
|
|
"mode": {"type": "string", "enum": ["append", "replace"], "default": "append"},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "archive_project",
|
|
"description": "Archivia o ripristina un progetto",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["project_id"],
|
|
"properties": {
|
|
"project_id": {"type": "string"},
|
|
"archived": {"type": "boolean", "default": True},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "search_context",
|
|
"description": "Ricerca semantica nei contesti salvati dell'utente",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["query"],
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
"limit": {"type": "integer", "default": 8},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "search_knowledge",
|
|
"description": "Ricerca semantica su Paperless, contesti salvati e repository Gitea indicizzati",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["query"],
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
"limit": {"type": "integer", "default": 8},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "search_gitea_knowledge",
|
|
"description": "Ricerca semantica solo sui file Gitea indicizzati (runbook, markdown, codice)",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["query"],
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
"limit": {"type": "integer", "default": 8},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "list_gitea_indexed_files",
|
|
"description": "Elenca gli ultimi file Gitea indicizzati nel vector store",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "Filtra per owner/name"},
|
|
"limit": {"type": "integer", "default": 30},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "reindex_gitea_repo",
|
|
"description": "Re-indicizza i file testo di un repository Gitea (admin o owner repo)",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["repo"],
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "owner/name, es. daniele/rete"},
|
|
"max_files": {
|
|
"type": "integer",
|
|
"description": "Limite file per questa esecuzione (default: config globale)",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "get_document",
|
|
"description": "Recupera il contenuto testuale di un documento Paperless per ID",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["doc_id"],
|
|
"properties": {"doc_id": {"type": "integer"}},
|
|
},
|
|
},
|
|
{
|
|
"name": "list_recent_documents",
|
|
"description": "Elenca gli ultimi documenti indicizzati",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {"limit": {"type": "integer", "default": 20}},
|
|
},
|
|
},
|
|
{
|
|
"name": "reindex_document",
|
|
"description": "Re-indicizza un documento Paperless (admin)",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["doc_id"],
|
|
"properties": {"doc_id": {"type": "integer"}},
|
|
},
|
|
},
|
|
{
|
|
"name": "list_repos",
|
|
"description": "Elenca i repository Gitea accessibili all'utente corrente",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"page": {"type": "integer", "default": 1},
|
|
"limit": {"type": "integer", "default": 50},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "get_file",
|
|
"description": "Legge un file (o elenca una directory) da un repository Gitea",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["repo", "path"],
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "owner/name, es. daniele/rete"},
|
|
"path": {"type": "string", "description": "Percorso nel repo, es. ha/RUNBOOK-failover.md"},
|
|
"ref": {"type": "string", "description": "Branch o tag (default: branch predefinito del repo)"},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "search_code",
|
|
"description": "Cerca testo nel codice o nei file di un repository Gitea",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["query"],
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
"repo": {"type": "string", "description": "Limita la ricerca a owner/name"},
|
|
"limit": {"type": "integer", "default": 20},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "list_issues",
|
|
"description": "Elenca le issue di un repository Gitea",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["repo"],
|
|
"properties": {
|
|
"repo": {"type": "string"},
|
|
"state": {"type": "string", "enum": ["open", "closed", "all"], "default": "open"},
|
|
"page": {"type": "integer", "default": 1},
|
|
"limit": {"type": "integer", "default": 20},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "get_issue",
|
|
"description": "Recupera una issue Gitea per numero",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["repo", "number"],
|
|
"properties": {
|
|
"repo": {"type": "string"},
|
|
"number": {"type": "integer"},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "create_issue",
|
|
"description": "Crea una nuova issue su un repository Gitea",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["repo", "title"],
|
|
"properties": {
|
|
"repo": {"type": "string"},
|
|
"title": {"type": "string"},
|
|
"body": {"type": "string", "default": ""},
|
|
"labels": {"type": "array", "items": {"type": "string"}},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "create_gitea_repo",
|
|
"description": "Crea un nuovo repository Gitea sotto l'utente corrente (workspace personale)",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["name"],
|
|
"properties": {
|
|
"name": {"type": "string", "description": "Nome repo (es. progetti)"},
|
|
"private": {"type": "boolean", "default": True},
|
|
"description": {"type": "string", "default": ""},
|
|
"auto_init": {
|
|
"type": "boolean",
|
|
"default": True,
|
|
"description": "Crea README.md iniziale",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "create_or_update_file",
|
|
"description": "Crea o aggiorna un file su Gitea (commit singolo via API, equivalente a push di un file)",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["repo", "path", "content", "message"],
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "owner/name"},
|
|
"path": {"type": "string", "description": "Percorso file nel repo"},
|
|
"content": {"type": "string", "description": "Contenuto testo del file"},
|
|
"message": {"type": "string", "description": "Messaggio di commit"},
|
|
"branch": {"type": "string", "description": "Branch (default: branch principale del repo)"},
|
|
"reindex": {
|
|
"type": "boolean",
|
|
"default": True,
|
|
"description": "Aggiorna subito il vector store RAG per questo file",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "get_home_dashboard",
|
|
"description": "Dashboard Loogle Casa: meteo, alert, rete, notifiche",
|
|
"inputSchema": {"type": "object", "properties": {}},
|
|
},
|
|
{
|
|
"name": "get_home_weather",
|
|
"description": "Meteo attuale e previsioni per casa (Loogle Casa)",
|
|
"inputSchema": {"type": "object", "properties": {}},
|
|
},
|
|
{
|
|
"name": "get_network_overview",
|
|
"description": "Panoramica rete domestica: dispositivi online/offline, IP pubblico",
|
|
"inputSchema": {"type": "object", "properties": {}},
|
|
},
|
|
{
|
|
"name": "get_network_failover_status",
|
|
"description": "Stato cluster failover LOOGLE (Pi, NAS, servizi)",
|
|
"inputSchema": {"type": "object", "properties": {}},
|
|
},
|
|
{
|
|
"name": "get_ha_entity",
|
|
"description": "Legge lo stato di un'entità Home Assistant (read-only)",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["entity_id"],
|
|
"properties": {
|
|
"entity_id": {
|
|
"type": "string",
|
|
"description": "Es. switch.pompa_pozzo, sensor.temperatura_sala",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "list_ha_entities",
|
|
"description": "Elenca entità Home Assistant, opzionalmente filtrate per dominio",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"domain": {
|
|
"type": "string",
|
|
"description": "Filtra per dominio: switch, sensor, light, climate, …",
|
|
},
|
|
"limit": {"type": "integer", "default": 50},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "search_ha_entities",
|
|
"description": "Cerca entità Home Assistant per nome o entity_id",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["query"],
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
"limit": {"type": "integer", "default": 20},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "get_irrigation_status",
|
|
"description": "Stato irrigazione: zone, programma, pozzo, sensori, connessione HA",
|
|
"inputSchema": {"type": "object", "properties": {}},
|
|
},
|
|
{
|
|
"name": "get_irrigation_zones",
|
|
"description": "Elenco zone irrigazione con stato valvole e portata",
|
|
"inputSchema": {"type": "object", "properties": {}},
|
|
},
|
|
{
|
|
"name": "get_irrigation_history",
|
|
"description": "Storico irrigazioni recenti",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {"limit": {"type": "integer", "default": 30}},
|
|
},
|
|
},
|
|
{
|
|
"name": "get_turni_status",
|
|
"description": "Stato servizio Turni-Live (versione, ambiente)",
|
|
"inputSchema": {"type": "object", "properties": {}},
|
|
},
|
|
{
|
|
"name": "get_my_shifts",
|
|
"description": "Turni di lavoro dell'utente corrente (Turni-Live)",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"from_date": {"type": "string", "description": "ISO date YYYY-MM-DD"},
|
|
"to_date": {"type": "string", "description": "ISO date YYYY-MM-DD"},
|
|
"limit": {"type": "integer", "default": 50},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "list_turni_doctors",
|
|
"description": "Elenco medici/operatori in Turni-Live",
|
|
"inputSchema": {"type": "object", "properties": {}},
|
|
},
|
|
{
|
|
"name": "search_apps_knowledge",
|
|
"description": "Ricerca semantica su storico Irrigazione e Turni indicizzati",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"required": ["query"],
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
"source": {
|
|
"type": "string",
|
|
"enum": ["irrigazione", "turni"],
|
|
"description": "Filtra per sorgente app",
|
|
},
|
|
"limit": {"type": "integer", "default": 8},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "list_apps_indexed",
|
|
"description": "Elenca record Irrigazione/Turni indicizzati nel vector store",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"source": {"type": "string", "enum": ["irrigazione", "turni"]},
|
|
"limit": {"type": "integer", "default": 30},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "reindex_apps",
|
|
"description": "Forza re-indicizzazione RAG da Irrigazione e/o Turni (admin)",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"source": {
|
|
"type": "string",
|
|
"enum": ["irrigazione", "turni", "all"],
|
|
"default": "all",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
]
|
|
|
|
|
|
def _text_result(payload: Any) -> dict:
|
|
return {"content": [{"type": "text", "text": json.dumps(payload, ensure_ascii=False, indent=2)}]}
|
|
|
|
|
|
def call_tool(name: str, arguments: dict, claims: dict) -> dict:
|
|
username = claims["sub"]
|
|
is_admin = "admin" in (claims.get("scope") or "").split()
|
|
|
|
if name == "ping":
|
|
audit.log_tool(username, name)
|
|
return _text_result({"pong": True, "service": "loogle-mcp"})
|
|
|
|
if name == "whoami":
|
|
audit.log_tool(username, name)
|
|
return _text_result({"username": username, "scope": claims.get("scope"), "is_admin": is_admin})
|
|
|
|
if name == "list_projects":
|
|
if not has_scope(claims, "context:read"):
|
|
raise PermissionError("Scope context:read richiesto")
|
|
projects = context_store.list_projects(username, arguments.get("include_archived", False))
|
|
audit.log_tool(username, name)
|
|
return _text_result({"projects": projects})
|
|
|
|
if name == "create_project":
|
|
if not has_scope(claims, "context:write"):
|
|
raise PermissionError("Scope context:write richiesto")
|
|
gitea_repo = arguments.get("gitea_repo")
|
|
if gitea_repo and not has_scope(claims, "gitea:read"):
|
|
raise PermissionError("Scope gitea:read richiesto per collegare un repository")
|
|
try:
|
|
meta = context_store.create_project(
|
|
username,
|
|
arguments["title"],
|
|
arguments.get("tags"),
|
|
gitea_repo=gitea_repo,
|
|
seed_from_gitea=bool(arguments.get("seed_from_gitea")),
|
|
)
|
|
except RuntimeError as exc:
|
|
raise PermissionError(str(exc)) from exc
|
|
except ValueError as exc:
|
|
raise ValueError(str(exc)) from exc
|
|
audit.log_tool(username, name, meta["id"])
|
|
return _text_result(meta)
|
|
|
|
if name == "link_project_repo":
|
|
if not has_scope(claims, "context:write"):
|
|
raise PermissionError("Scope context:write richiesto")
|
|
gitea_repo = arguments.get("gitea_repo")
|
|
if gitea_repo and not has_scope(claims, "gitea:read"):
|
|
raise PermissionError("Scope gitea:read richiesto per collegare un repository")
|
|
try:
|
|
meta = context_store.link_project_repo(
|
|
username,
|
|
arguments["project_id"],
|
|
gitea_repo=gitea_repo,
|
|
seed_from_gitea=bool(arguments.get("seed_from_gitea")),
|
|
)
|
|
except RuntimeError as exc:
|
|
raise PermissionError(str(exc)) from exc
|
|
except ValueError as exc:
|
|
raise ValueError(str(exc)) from exc
|
|
audit.log_tool(username, name, arguments["project_id"])
|
|
return _text_result(meta)
|
|
|
|
if name == "get_project_context":
|
|
if not has_scope(claims, "context:read"):
|
|
raise PermissionError("Scope context:read richiesto")
|
|
include_gitea = arguments.get("include_gitea", True)
|
|
if include_gitea and not has_scope(claims, "gitea:read"):
|
|
include_gitea = False
|
|
data = context_store.get_project_context(
|
|
username,
|
|
arguments["project_id"],
|
|
arguments.get("session_limit", 5),
|
|
include_gitea=include_gitea,
|
|
)
|
|
if include_gitea is False and arguments.get("include_gitea", True):
|
|
meta = data.get("meta") or {}
|
|
if meta.get("gitea_repo"):
|
|
data["gitea"] = {
|
|
"linked": True,
|
|
"repo": meta["gitea_repo"],
|
|
"available": False,
|
|
"error": "Scope gitea:read richiesto per arricchimento repository",
|
|
}
|
|
audit.log_tool(username, name, arguments["project_id"])
|
|
return _text_result(data)
|
|
|
|
if name == "save_context":
|
|
if not has_scope(claims, "context:write"):
|
|
raise PermissionError("Scope context:write richiesto")
|
|
meta = context_store.save_context(
|
|
username,
|
|
arguments["project_id"],
|
|
arguments["content"],
|
|
arguments.get("mode", "append"),
|
|
)
|
|
try:
|
|
indexer.index_context_snippet(
|
|
username, arguments["project_id"], arguments["content"], meta["title"]
|
|
)
|
|
except Exception:
|
|
pass
|
|
audit.log_tool(username, name, arguments["project_id"])
|
|
return _text_result(meta)
|
|
|
|
if name == "archive_project":
|
|
if not has_scope(claims, "context:write"):
|
|
raise PermissionError("Scope context:write richiesto")
|
|
meta = context_store.archive_project(
|
|
username, arguments["project_id"], arguments.get("archived", True)
|
|
)
|
|
audit.log_tool(username, name, arguments["project_id"])
|
|
return _text_result(meta)
|
|
|
|
if name == "search_context":
|
|
if not has_scope(claims, "context:read"):
|
|
raise PermissionError("Scope context:read richiesto")
|
|
hits = indexer.search_context(username, arguments["query"], arguments.get("limit", 8))
|
|
audit.log_tool(username, name, detail={"query": arguments["query"]})
|
|
return _text_result({"results": hits})
|
|
|
|
if name == "search_knowledge":
|
|
if not has_scope(claims, "knowledge:read"):
|
|
raise PermissionError("Scope knowledge:read richiesto")
|
|
hits = indexer.search_knowledge(
|
|
username, arguments["query"], arguments.get("limit", 8), is_admin=is_admin
|
|
)
|
|
audit.log_tool(username, name, detail={"query": arguments["query"]})
|
|
return _text_result({"results": hits})
|
|
|
|
if name == "search_gitea_knowledge":
|
|
if not has_scope(claims, "knowledge:read"):
|
|
raise PermissionError("Scope knowledge:read richiesto")
|
|
if not gitea.is_configured(username):
|
|
raise PermissionError("Gitea non configurato per questo utente")
|
|
hits = indexer.search_gitea_knowledge(
|
|
username, arguments["query"], arguments.get("limit", 8), is_admin=is_admin
|
|
)
|
|
audit.log_tool(username, name, detail={"query": arguments["query"]})
|
|
return _text_result({"results": hits})
|
|
|
|
if name == "list_gitea_indexed_files":
|
|
if not has_scope(claims, "knowledge:read"):
|
|
raise PermissionError("Scope knowledge:read richiesto")
|
|
files = gitea_indexer.list_indexed_files(
|
|
arguments.get("limit", 30),
|
|
repo=arguments.get("repo"),
|
|
)
|
|
audit.log_tool(username, name, detail={"repo": arguments.get("repo")})
|
|
return _text_result({"files": files, "count": len(files)})
|
|
|
|
if name == "reindex_gitea_repo":
|
|
if not has_scope(claims, "gitea:read"):
|
|
raise PermissionError("Scope gitea:read richiesto")
|
|
if not gitea.is_configured(username):
|
|
raise PermissionError("Gitea non configurato per questo utente")
|
|
repo = arguments["repo"]
|
|
owner = repo.split("/", 1)[0].lower()
|
|
if not is_admin and owner != username:
|
|
raise PermissionError("Puoi re-indicizzare solo repository di cui sei owner")
|
|
try:
|
|
result = gitea_indexer.index_repo(
|
|
repo,
|
|
username=username,
|
|
force=True,
|
|
max_files=arguments.get("max_files"),
|
|
)
|
|
except Exception as exc:
|
|
raise PermissionError(f"Re-indicizzazione Gitea fallita: {exc}") from exc
|
|
audit.log_tool(username, name, detail={"repo": repo})
|
|
return _text_result(result)
|
|
|
|
if name == "get_document":
|
|
if not has_scope(claims, "knowledge:read"):
|
|
raise PermissionError("Scope knowledge:read richiesto")
|
|
try:
|
|
text = paperless.download_document_text(int(arguments["doc_id"]), username=username)
|
|
except Exception as exc:
|
|
raise PermissionError(f"Documento non accessibile con il tuo account Paperless: {exc}")
|
|
audit.log_tool(username, name, str(arguments["doc_id"]))
|
|
return _text_result({"doc_id": arguments["doc_id"], "content": text})
|
|
|
|
if name == "list_recent_documents":
|
|
if not has_scope(claims, "knowledge:read"):
|
|
raise PermissionError("Scope knowledge:read richiesto")
|
|
docs = indexer.list_recent_documents(arguments.get("limit", 20))
|
|
audit.log_tool(username, name)
|
|
return _text_result({"documents": docs})
|
|
|
|
if name == "reindex_document":
|
|
if not is_admin:
|
|
raise PermissionError("Scope admin richiesto")
|
|
result = indexer.index_document(int(arguments["doc_id"]), force=True)
|
|
audit.log_tool(username, name, str(arguments["doc_id"]))
|
|
return _text_result(result)
|
|
|
|
if name in (
|
|
"list_repos", "get_file", "search_code", "list_issues", "get_issue", "create_issue",
|
|
"create_gitea_repo", "create_or_update_file",
|
|
):
|
|
if not gitea.is_configured(username):
|
|
raise PermissionError(
|
|
"Gitea non configurato per questo utente. "
|
|
"Aggiungi GITEA_API_TOKEN_{USER} in .env — vedi docs/GITEA-TOKEN.md"
|
|
)
|
|
|
|
if name == "list_repos":
|
|
if not has_scope(claims, "gitea:read"):
|
|
raise PermissionError("Scope gitea:read richiesto")
|
|
try:
|
|
result = gitea.list_repos(
|
|
username=username,
|
|
page=int(arguments.get("page", 1)),
|
|
limit=int(arguments.get("limit", 50)),
|
|
)
|
|
except Exception as exc:
|
|
raise PermissionError(f"Gitea non accessibile: {exc}") from exc
|
|
audit.log_tool(username, name)
|
|
return _text_result(result)
|
|
|
|
if name == "get_file":
|
|
if not has_scope(claims, "gitea:read"):
|
|
raise PermissionError("Scope gitea:read richiesto")
|
|
try:
|
|
result = gitea.get_file(
|
|
arguments["repo"],
|
|
arguments["path"],
|
|
ref=arguments.get("ref"),
|
|
username=username,
|
|
)
|
|
except FileNotFoundError as exc:
|
|
raise ValueError(str(exc)) from exc
|
|
except Exception as exc:
|
|
raise PermissionError(f"File Gitea non accessibile: {exc}") from exc
|
|
audit.log_tool(username, name, detail={"repo": arguments["repo"], "path": arguments["path"]})
|
|
return _text_result(result)
|
|
|
|
if name == "search_code":
|
|
if not has_scope(claims, "gitea:read"):
|
|
raise PermissionError("Scope gitea:read richiesto")
|
|
try:
|
|
result = gitea.search_code(
|
|
arguments["query"],
|
|
repo=arguments.get("repo"),
|
|
limit=int(arguments.get("limit", 20)),
|
|
username=username,
|
|
)
|
|
except Exception as exc:
|
|
raise PermissionError(f"Ricerca Gitea fallita: {exc}") from exc
|
|
audit.log_tool(username, name, detail={"query": arguments["query"], "repo": arguments.get("repo")})
|
|
return _text_result(result)
|
|
|
|
if name == "list_issues":
|
|
if not has_scope(claims, "gitea:read"):
|
|
raise PermissionError("Scope gitea:read richiesto")
|
|
try:
|
|
result = gitea.list_issues(
|
|
arguments["repo"],
|
|
state=arguments.get("state", "open"),
|
|
page=int(arguments.get("page", 1)),
|
|
limit=int(arguments.get("limit", 20)),
|
|
username=username,
|
|
)
|
|
except Exception as exc:
|
|
raise PermissionError(f"Issue Gitea non accessibili: {exc}") from exc
|
|
audit.log_tool(username, name, detail={"repo": arguments["repo"]})
|
|
return _text_result(result)
|
|
|
|
if name == "get_issue":
|
|
if not has_scope(claims, "gitea:read"):
|
|
raise PermissionError("Scope gitea:read richiesto")
|
|
try:
|
|
result = gitea.get_issue(
|
|
arguments["repo"],
|
|
int(arguments["number"]),
|
|
username=username,
|
|
)
|
|
except Exception as exc:
|
|
raise PermissionError(f"Issue Gitea non accessibile: {exc}") from exc
|
|
audit.log_tool(username, name, detail={"repo": arguments["repo"], "number": arguments["number"]})
|
|
return _text_result(result)
|
|
|
|
if name == "create_issue":
|
|
if not has_scope(claims, "gitea:write"):
|
|
raise PermissionError("Scope gitea:write richiesto")
|
|
try:
|
|
result = gitea.create_issue(
|
|
arguments["repo"],
|
|
arguments["title"],
|
|
body=arguments.get("body", ""),
|
|
labels=arguments.get("labels"),
|
|
username=username,
|
|
)
|
|
except Exception as exc:
|
|
raise PermissionError(f"Creazione issue Gitea fallita: {exc}") from exc
|
|
audit.log_tool(username, name, detail={"repo": arguments["repo"], "title": arguments["title"]})
|
|
return _text_result(result)
|
|
|
|
if name == "create_gitea_repo":
|
|
if not has_scope(claims, "gitea:write"):
|
|
raise PermissionError("Scope gitea:write richiesto")
|
|
try:
|
|
result = gitea.create_repo(
|
|
arguments["name"],
|
|
username=username,
|
|
private=bool(arguments.get("private", True)),
|
|
description=arguments.get("description", ""),
|
|
auto_init=bool(arguments.get("auto_init", True)),
|
|
)
|
|
except Exception as exc:
|
|
raise PermissionError(f"Creazione repo Gitea fallita: {exc}") from exc
|
|
audit.log_tool(username, name, detail={"name": arguments["name"]})
|
|
return _text_result(result)
|
|
|
|
if name == "create_or_update_file":
|
|
if not has_scope(claims, "gitea:write"):
|
|
raise PermissionError("Scope gitea:write richiesto")
|
|
repo = arguments["repo"]
|
|
try:
|
|
gitea.assert_repo_owner(username, repo, is_admin=is_admin)
|
|
result = gitea.create_or_update_file(
|
|
repo,
|
|
arguments["path"],
|
|
arguments["content"],
|
|
arguments["message"],
|
|
branch=arguments.get("branch"),
|
|
username=username,
|
|
)
|
|
if arguments.get("reindex", True):
|
|
try:
|
|
owner, repo_name = gitea.parse_repo(repo)
|
|
meta = gitea._request(
|
|
"GET", f"/repos/{owner}/{repo_name}", username=username
|
|
)
|
|
private = bool(meta.get("private"))
|
|
idx = gitea_indexer.index_file(
|
|
repo,
|
|
arguments["path"],
|
|
username=username,
|
|
private=private,
|
|
force=True,
|
|
)
|
|
result["reindex"] = idx
|
|
except Exception:
|
|
result["reindex"] = {"skipped": True}
|
|
except PermissionError:
|
|
raise
|
|
except Exception as exc:
|
|
raise PermissionError(f"Scrittura file Gitea fallita: {exc}") from exc
|
|
audit.log_tool(
|
|
username,
|
|
name,
|
|
detail={"repo": repo, "path": arguments["path"], "action": result.get("action")},
|
|
)
|
|
return _text_result(result)
|
|
|
|
if name in (
|
|
"get_home_dashboard", "get_home_weather", "get_network_overview",
|
|
"get_network_failover_status",
|
|
):
|
|
if not has_scope(claims, "home:read"):
|
|
raise PermissionError("Scope home:read richiesto")
|
|
if not casa.is_configured(username):
|
|
raise PermissionError(
|
|
"Loogle Casa non configurato. Imposta LOOGLE_CASA_PASSWORD_{USER} in .env"
|
|
)
|
|
try:
|
|
if name == "get_home_dashboard":
|
|
payload = casa.get_dashboard(username)
|
|
elif name == "get_home_weather":
|
|
payload = casa.get_weather_home(username)
|
|
elif name == "get_network_overview":
|
|
payload = casa.get_network_overview(username)
|
|
else:
|
|
payload = casa.get_network_failover_status(username)
|
|
except Exception as exc:
|
|
raise PermissionError(f"Loogle Casa non accessibile: {exc}") from exc
|
|
audit.log_tool(username, name)
|
|
return _text_result(payload)
|
|
|
|
if name in ("get_ha_entity", "list_ha_entities", "search_ha_entities"):
|
|
if not has_scope(claims, "home:read"):
|
|
raise PermissionError("Scope home:read richiesto")
|
|
if not homeassistant.is_configured():
|
|
raise PermissionError("Home Assistant non configurato — imposta HA_TOKEN in .env")
|
|
try:
|
|
if name == "get_ha_entity":
|
|
payload = homeassistant.get_entity(arguments["entity_id"])
|
|
elif name == "list_ha_entities":
|
|
payload = {
|
|
"entities": homeassistant.list_entities(
|
|
domain=arguments.get("domain"),
|
|
limit=int(arguments.get("limit", 50)),
|
|
),
|
|
}
|
|
else:
|
|
payload = {
|
|
"results": homeassistant.search_entities(
|
|
arguments["query"],
|
|
limit=int(arguments.get("limit", 20)),
|
|
),
|
|
}
|
|
except Exception as exc:
|
|
raise PermissionError(f"Home Assistant non accessibile: {exc}") from exc
|
|
audit.log_tool(username, name, detail=arguments.get("entity_id") or arguments.get("query"))
|
|
return _text_result(payload)
|
|
|
|
if name in ("get_irrigation_status", "get_irrigation_zones", "get_irrigation_history"):
|
|
if not has_scope(claims, "irrigation:read"):
|
|
raise PermissionError("Scope irrigation:read richiesto")
|
|
if not irrigazione.is_configured(username):
|
|
raise PermissionError(
|
|
"Irrigazione non configurata. Imposta IRRIGAZIONE_PASSWORD_{USER} in .env"
|
|
)
|
|
try:
|
|
if name == "get_irrigation_status":
|
|
payload = irrigazione.get_status(username)
|
|
elif name == "get_irrigation_zones":
|
|
payload = irrigazione.get_zones(username)
|
|
else:
|
|
payload = irrigazione.get_history(username, int(arguments.get("limit", 30)))
|
|
except Exception as exc:
|
|
raise PermissionError(f"Irrigazione non accessibile: {exc}") from exc
|
|
audit.log_tool(username, name)
|
|
return _text_result(payload)
|
|
|
|
if name in ("get_turni_status", "get_my_shifts", "list_turni_doctors"):
|
|
if name != "get_turni_status" and not has_scope(claims, "turni:read"):
|
|
raise PermissionError("Scope turni:read richiesto")
|
|
try:
|
|
if name == "get_turni_status":
|
|
payload = turni.get_status()
|
|
elif name == "list_turni_doctors":
|
|
if not turni.is_configured(username):
|
|
raise PermissionError("Turni non configurato per questo utente")
|
|
payload = {"doctors": turni.list_doctors(username)}
|
|
else:
|
|
if not turni.is_configured(username):
|
|
raise PermissionError("Turni non configurato per questo utente")
|
|
payload = turni.get_my_shifts(
|
|
username,
|
|
from_date=arguments.get("from_date"),
|
|
to_date=arguments.get("to_date"),
|
|
limit=int(arguments.get("limit", 50)),
|
|
)
|
|
except PermissionError:
|
|
raise
|
|
except Exception as exc:
|
|
raise PermissionError(f"Turni non accessibile: {exc}") from exc
|
|
audit.log_tool(username, name)
|
|
return _text_result(payload)
|
|
|
|
if name == "search_apps_knowledge":
|
|
if not has_scope(claims, "knowledge:read"):
|
|
raise PermissionError("Scope knowledge:read richiesto")
|
|
hits = apps_indexer.search_apps_knowledge(
|
|
arguments["query"],
|
|
limit=int(arguments.get("limit", 8)),
|
|
source=arguments.get("source"),
|
|
)
|
|
audit.log_tool(username, name, detail={"query": arguments["query"]})
|
|
return _text_result({"results": hits})
|
|
|
|
if name == "list_apps_indexed":
|
|
if not has_scope(claims, "knowledge:read"):
|
|
raise PermissionError("Scope knowledge:read richiesto")
|
|
records = apps_indexer.list_indexed_records(
|
|
source=arguments.get("source"),
|
|
limit=int(arguments.get("limit", 30)),
|
|
)
|
|
audit.log_tool(username, name)
|
|
return _text_result({"records": records, "count": len(records)})
|
|
|
|
if name == "reindex_apps":
|
|
if not is_admin:
|
|
raise PermissionError("Scope admin richiesto")
|
|
src = arguments.get("source", "all")
|
|
if src == "irrigazione":
|
|
result = apps_indexer.index_irrigazione()
|
|
elif src == "turni":
|
|
result = apps_indexer.index_turni()
|
|
else:
|
|
result = apps_indexer.index_all()
|
|
audit.log_tool(username, name, detail={"source": src})
|
|
return _text_result(result)
|
|
|
|
raise ValueError(f"Tool sconosciuto: {name}")
|
|
|
|
|
|
def list_resources(claims: dict) -> list:
|
|
username = claims["sub"]
|
|
return context_store.list_resources(username)
|
|
|
|
|
|
def read_resource(uri: str, claims: dict) -> dict:
|
|
username = claims["sub"]
|
|
return context_store.read_resource(username, uri)
|