""" Tool-Registry für den AI Agent. Wrappt bestehende Django-ORM-Abfragen (analog zu mcp_server/tools/lesen.py) mit direktem DB-Zugriff und PII-Filterung basierend auf Django-User-Berechtigungen. Schreib-Tools sind standardmäßig deaktiviert. """ from __future__ import annotations import json import logging from decimal import Decimal from typing import Any logger = logging.getLogger(__name__) # ────────────────────────────────────────────────────────────────────────────── # Hilfsfunktionen # ────────────────────────────────────────────────────────────────────────────── def _get_role(user) -> str: """Leitet MCP-Rolle aus Django-User ab.""" if user.is_superuser or user.has_perm("stiftung.access_administration"): return "admin" return "readonly" def _serialize(obj: Any) -> Any: """Serialisiert Django-Modell-Werte zu JSON-fähigen Typen.""" if obj is None: return None if isinstance(obj, Decimal): return float(obj) if hasattr(obj, "isoformat"): return obj.isoformat() if hasattr(obj, "__str__"): return str(obj) return obj def _apply_pii(data: dict, model_type: str, role: str) -> dict: """Wendet PII-Filterung via mcp_server.privacy an.""" from mcp_server.privacy import apply_privacy_filter return apply_privacy_filter(data, model_type, role) # ────────────────────────────────────────────────────────────────────────────── # Tool-Implementierungen (Read-Only) # ────────────────────────────────────────────────────────────────────────────── def tool_destinataer_suchen(user, suchbegriff: str = "", aktiv: bool | None = None, limit: int = 20) -> str: from django.db.models import Q from stiftung.models import Destinataer role = _get_role(user) limit = min(limit, 50) qs = Destinataer.objects.all() if suchbegriff: qs = qs.filter( Q(vorname__icontains=suchbegriff) | Q(nachname__icontains=suchbegriff) | Q(institution__icontains=suchbegriff) ) if aktiv is not None: qs = qs.filter(aktiv=aktiv) qs = qs.order_by("nachname", "vorname")[:limit] results = [] for obj in qs: item = { "id": str(obj.id), "vorname": obj.vorname, "nachname": obj.nachname, "familienzweig": obj.familienzweig, "aktiv": obj.aktiv, "ort": obj.ort, "email": obj.email, } results.append(_apply_pii(item, "destinataer", role)) return json.dumps({"count": len(results), "destinataere": results}, default=_serialize, ensure_ascii=False) def tool_land_suchen(user, suchbegriff: str = "", limit: int = 20) -> str: from django.db.models import Q from stiftung.models import Land limit = min(limit, 50) qs = Land.objects.all() if suchbegriff: qs = qs.filter( Q(bezeichnung__icontains=suchbegriff) | Q(gemarkung__icontains=suchbegriff) | Q(ort__icontains=suchbegriff) ) qs = qs.order_by("bezeichnung")[:limit] results = [] for obj in qs: results.append({ "id": str(obj.id), "bezeichnung": obj.bezeichnung, "gemarkung": getattr(obj, "gemarkung", ""), "ort": getattr(obj, "ort", ""), "flaeche_ha": _serialize(getattr(obj, "flaeche_ha", None)), "aktiv": getattr(obj, "aktiv", True), }) return json.dumps({"count": len(results), "laendereien": results}, default=_serialize, ensure_ascii=False) def tool_konten_uebersicht(user) -> str: from stiftung.models import StiftungsKonto role = _get_role(user) konten = StiftungsKonto.objects.all().order_by("bezeichnung") results = [] for k in konten: item = { "id": str(k.id), "bezeichnung": k.bezeichnung, "bank": getattr(k, "bank", ""), "kontonummer": getattr(k, "kontonummer", ""), "iban": getattr(k, "iban", ""), "aktiv": getattr(k, "aktiv", True), } results.append(_apply_pii(item, "konto", role)) return json.dumps({"count": len(results), "konten": results}, default=_serialize, ensure_ascii=False) def tool_foerderungen_suchen(user, suchbegriff: str = "", status: str = "", limit: int = 20) -> str: from django.db.models import Q from stiftung.models import Foerderung limit = min(limit, 50) qs = Foerderung.objects.select_related("destinataer").all() if suchbegriff: qs = qs.filter( Q(bezeichnung__icontains=suchbegriff) | Q(destinataer__nachname__icontains=suchbegriff) | Q(destinataer__vorname__icontains=suchbegriff) ) if status: qs = qs.filter(status=status) qs = qs.order_by("-erstellt_am")[:limit] results = [] for obj in qs: results.append({ "id": str(obj.id), "bezeichnung": getattr(obj, "bezeichnung", ""), "destinataer": str(obj.destinataer) if obj.destinataer else None, "betrag": _serialize(getattr(obj, "betrag", None)), "status": getattr(obj, "status", ""), "erstellt_am": _serialize(getattr(obj, "erstellt_am", None)), }) return json.dumps({"count": len(results), "foerderungen": results}, default=_serialize, ensure_ascii=False) def tool_verwaltungskosten(user, jahr: int | None = None, limit: int = 20) -> str: from stiftung.models import Verwaltungskosten limit = min(limit, 50) qs = Verwaltungskosten.objects.all() if jahr: qs = qs.filter(datum__year=jahr) qs = qs.order_by("-datum")[:limit] results = [] for obj in qs: results.append({ "id": str(obj.id), "datum": _serialize(getattr(obj, "datum", None)), "bezeichnung": getattr(obj, "bezeichnung", ""), "betrag": _serialize(getattr(obj, "betrag", None)), "kategorie": getattr(obj, "kategorie", ""), }) return json.dumps({"count": len(results), "verwaltungskosten": results}, default=_serialize, ensure_ascii=False) def tool_termine_anzeigen(user, limit: int = 10) -> str: from django.utils import timezone from stiftung.models import StiftungsKalenderEintrag now = timezone.now().date() qs = StiftungsKalenderEintrag.objects.filter(datum__gte=now).order_by("datum")[:limit] results = [] for obj in qs: results.append({ "id": str(obj.id), "titel": getattr(obj, "titel", ""), "datum": _serialize(getattr(obj, "datum", None)), "beschreibung": getattr(obj, "beschreibung", ""), "typ": getattr(obj, "typ", ""), }) return json.dumps({"count": len(results), "termine": results}, default=_serialize, ensure_ascii=False) def tool_transaktionen_suchen(user, suchbegriff: str = "", limit: int = 20) -> str: from django.db.models import Q from stiftung.models import BankTransaction limit = min(limit, 50) qs = BankTransaction.objects.all() if suchbegriff: qs = qs.filter( Q(verwendungszweck__icontains=suchbegriff) | Q(auftraggeber__icontains=suchbegriff) ) qs = qs.order_by("-buchungsdatum")[:limit] results = [] for obj in qs: results.append({ "id": str(obj.id), "datum": _serialize(getattr(obj, "buchungsdatum", None)), "betrag": _serialize(getattr(obj, "betrag", None)), "verwendungszweck": getattr(obj, "verwendungszweck", ""), "auftraggeber": getattr(obj, "auftraggeber", ""), }) return json.dumps({"count": len(results), "transaktionen": results}, default=_serialize, ensure_ascii=False) def tool_dashboard(user) -> str: """Gibt eine Übersicht über Schlüsselkennzahlen zurück.""" from stiftung.models import Destinataer, Foerderung, Land, StiftungsKonto try: destinataere_aktiv = Destinataer.objects.filter(aktiv=True).count() destinataere_gesamt = Destinataer.objects.count() laendereien = Land.objects.count() konten = StiftungsKonto.objects.count() foerderungen_offen = Foerderung.objects.filter(status="offen").count() if hasattr(Foerderung, 'objects') else 0 return json.dumps({ "destinataere_aktiv": destinataere_aktiv, "destinataere_gesamt": destinataere_gesamt, "laendereien": laendereien, "konten": konten, "foerderungen_offen": foerderungen_offen, }, ensure_ascii=False) except Exception as e: return json.dumps({"fehler": str(e)}, ensure_ascii=False) # ────────────────────────────────────────────────────────────────────────────── # Tool-Dispatch und Schema # ────────────────────────────────────────────────────────────────────────────── TOOL_FUNCTIONS = { "destinataer_suchen": tool_destinataer_suchen, "land_suchen": tool_land_suchen, "konten_uebersicht": tool_konten_uebersicht, "foerderungen_suchen": tool_foerderungen_suchen, "verwaltungskosten": tool_verwaltungskosten, "termine_anzeigen": tool_termine_anzeigen, "transaktionen_suchen": tool_transaktionen_suchen, "dashboard": tool_dashboard, } TOOL_SCHEMAS = [ { "type": "function", "function": { "name": "destinataer_suchen", "description": "Sucht Destinatäre (Förderungsempfänger) nach Name oder Status.", "parameters": { "type": "object", "properties": { "suchbegriff": {"type": "string", "description": "Vor-/Nachname oder Institution"}, "aktiv": {"type": "boolean", "description": "true=nur Aktive, false=nur Inaktive"}, "limit": {"type": "integer", "description": "Max. Anzahl Ergebnisse (Standard: 20)"}, }, }, }, }, { "type": "function", "function": { "name": "land_suchen", "description": "Sucht Ländereien (Grundstücke) der Stiftung nach Bezeichnung oder Ort.", "parameters": { "type": "object", "properties": { "suchbegriff": {"type": "string", "description": "Bezeichnung, Gemarkung oder Ort"}, "limit": {"type": "integer", "description": "Max. Anzahl Ergebnisse"}, }, }, }, }, { "type": "function", "function": { "name": "konten_uebersicht", "description": "Zeigt alle Stiftungskonten mit Bankverbindungen.", "parameters": {"type": "object", "properties": {}}, }, }, { "type": "function", "function": { "name": "foerderungen_suchen", "description": "Sucht Förderungen nach Bezeichnung oder Destinatär.", "parameters": { "type": "object", "properties": { "suchbegriff": {"type": "string", "description": "Bezeichnung oder Destinatär-Name"}, "status": {"type": "string", "description": "Status-Filter (z.B. 'offen', 'genehmigt')"}, "limit": {"type": "integer"}, }, }, }, }, { "type": "function", "function": { "name": "verwaltungskosten", "description": "Listet Verwaltungskosten, optional nach Jahr gefiltert.", "parameters": { "type": "object", "properties": { "jahr": {"type": "integer", "description": "Filterjahr (z.B. 2025)"}, "limit": {"type": "integer"}, }, }, }, }, { "type": "function", "function": { "name": "termine_anzeigen", "description": "Zeigt bevorstehende Termine und Fristen der Stiftung.", "parameters": { "type": "object", "properties": { "limit": {"type": "integer", "description": "Max. Anzahl Termine"}, }, }, }, }, { "type": "function", "function": { "name": "transaktionen_suchen", "description": "Sucht Banktransaktionen nach Verwendungszweck oder Auftraggeber.", "parameters": { "type": "object", "properties": { "suchbegriff": {"type": "string"}, "limit": {"type": "integer"}, }, }, }, }, { "type": "function", "function": { "name": "dashboard", "description": "Zeigt Schlüsselkennzahlen der Stiftung (Anzahl Destinatäre, Ländereien, Konten etc.).", "parameters": {"type": "object", "properties": {}}, }, }, ] def execute_tool(name: str, arguments: dict, user) -> str: """Führt ein Tool aus und gibt das Ergebnis als String zurück.""" fn = TOOL_FUNCTIONS.get(name) if fn is None: return json.dumps({"fehler": f"Unbekanntes Tool: {name}"}, ensure_ascii=False) try: return fn(user, **arguments) except TypeError as e: logger.warning("Tool %s Parameterfehler: %s", name, e) return json.dumps({"fehler": f"Ungültige Parameter: {e}"}, ensure_ascii=False) except Exception as e: logger.error("Tool %s Fehler: %s", name, e, exc_info=True) return json.dumps({"fehler": f"Tool-Ausführung fehlgeschlagen: {e}"}, ensure_ascii=False)