"""Werkzeuge des Agenten (neutrale Definition + Ausführung auf dem Workspace).""" from __future__ import annotations import io import json from app.agent.workspace import Workspace, is_personal from app.services import llm, preview from app.settings import PLATFORM_DIR TOOLS = [ {"name": "list_files", "description": "Listet alle Dateien des Tenants (inkl. tenant.json) mit Größe und Änderungsstatus.", "schema": {"type": "object", "properties": {}}}, {"name": "read_file", "description": "Liest eine Textdatei des Tenants (prompt.txt, protocol.html, protocol.css, overrides.py, tenant.json …). " "Fehlt eine Datei, gilt der Core-Standard – nutze dann read_core_default.", "schema": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}}, {"name": "read_core_default", "description": "Liest eine Standarddatei des Cores als Vorlage: prompt.txt, protocol.html, protocol.css oder hooks.py (Hook-Dokumentation).", "schema": {"type": "object", "properties": {"name": {"type": "string", "enum": ["prompt.txt", "protocol.html", "protocol.css", "hooks.py"]}}, "required": ["name"]}}, {"name": "write_file", "description": "Schreibt/ersetzt eine Textdatei des Tenants vollständig (z. B. um eine eigene prompt.txt oder protocol.css anzulegen). Nur im Entwurf – ein Mensch gibt frei.", "schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}}, {"name": "edit_file", "description": "Ersetzt in einer Datei genau eine Stelle: old_str muss exakt 1× vorkommen. Bevorzuge das gegenüber write_file bei kleinen Änderungen.", "schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_str": {"type": "string"}, "new_str": {"type": "string"}}, "required": ["path", "old_str", "new_str"]}}, {"name": "delete_file", "description": "Löscht eine Tenant-Datei (z. B. um auf den Core-Standard zurückzufallen).", "schema": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}}, {"name": "set_config", "description": "Setzt einen Wert in tenant.json per Punkt-Pfad, z. B. key='colors.accent', value='#0B6E4F' oder key='flags.ui_kontakt_field', value=false.", "schema": {"type": "object", "properties": {"key": {"type": "string"}, "value": {}}, "required": ["key", "value"]}}, {"name": "validate", "description": "Prüft den aktuellen Workspace mit dem ECHTEN Core (Web-UI + PDF rendern, Prompt laden). Gibt Fehler/Warnungen zurück. Nach Änderungen immer aufrufen.", "schema": {"type": "object", "properties": {}}}, {"name": "render_pdf", "description": "Rendert das Beispielprotokoll als PDF und gibt Seitenzahl und extrahierten Text zurück, damit du Aufbau und Inhalt prüfen kannst.", "schema": {"type": "object", "properties": {}}}, {"name": "show_final_prompt", "description": "Zeigt den finalen System-Prompt, wie der Core ihn an das LLM schickt (nach Platzhalter-Ersetzung und Hooks). Namenslisten werden gekürzt.", "schema": {"type": "object", "properties": {}}}, {"name": "set_sample", "description": "Setzt das Beispielprotokoll, das die PDF-Vorschau rendert (Format wie vom LLM: ===TERMINART=== … ===/TERMINART===, dann Überschriften und Text). " "Sinnvoll, um Layout-Änderungen mit passendem Inhalt zu prüfen.", "schema": {"type": "object", "properties": {"text": {"type": "string"}}, "required": ["text"]}}, {"name": "run_prompt_test", "description": "Führt das Protokoll-LLM mit dem Workspace-Prompt auf einem Beispieltranskript aus (kostet Geld – höchstens 1× pro Antwort). Liefert den Protokolltext.", "schema": {"type": "object", "properties": {"transcript": {"type": "string", "description": "Optional; sonst das Standard-Beispieltranskript."}}}}, ] class ToolRunner: def __init__(self, ws: Workspace): self.ws, self.db, self.tenant = ws, ws.db, ws.tenant self.llm_calls = 0 def _preview(self, kind, **kw): kw.setdefault("sample", self.ws.sample) # Beispielprotokoll des Workspace (None → Standard des Tenants) return preview.run_content(self.db, self.tenant, self.ws.cfg, self.ws.files, kind, **kw) def run(self, name: str, a: dict) -> tuple[str, str]: """→ (Kurzbeschreibung des Arguments für die Anzeige, Ergebnistext).""" try: return getattr(self, f"t_{name}")(a) except (KeyError, ValueError, PermissionError, SyntaxError, json.JSONDecodeError) as exc: return a.get("path") or a.get("key") or "", f"FEHLER: {exc}" except preview.PreviewError as exc: return "", f"FEHLER (Core-Rendering): {str(exc)[-900:]}" def t_list_files(self, a): rows = [{"path": "tenant.json", "text": True}] + self.ws.list_files() return "", json.dumps(rows, ensure_ascii=False) def t_read_file(self, a): return a["path"], self.ws.read(a["path"]) def t_read_core_default(self, a): n = a["name"] p = (PLATFORM_DIR / "core" / ("hooks.py" if n == "hooks.py" else ("defaults" if n == "prompt.txt" else "templates")) / n) \ if n != "hooks.py" else PLATFORM_DIR / "core" / "hooks.py" return n, p.read_text(encoding="utf-8")[:60_000] def t_write_file(self, a): self.ws.write(a["path"], a["content"]) return a["path"], f"geschrieben ({len(a['content'])} Zeichen)" def t_edit_file(self, a): self.ws.edit(a["path"], a["old_str"], a["new_str"]) return a["path"], "ersetzt" def t_delete_file(self, a): self.ws.delete(a["path"]) return a["path"], "gelöscht" def t_set_config(self, a): self.ws.set_config_value(a["key"], a["value"]) return a["key"], f"gesetzt = {json.dumps(a['value'], ensure_ascii=False)}" def t_validate(self, a): r = self._preview("validate") return "Core-Rendering", json.dumps(r, ensure_ascii=False) def t_render_pdf(self, a): pdf = self._preview("pdf") from pypdf import PdfReader rd = PdfReader(io.BytesIO(pdf)) text = "\n".join((pg.extract_text() or "") for pg in rd.pages) return "Beispielprotokoll", f"{len(rd.pages)} Seite(n)\n---\n{text[:3500]}" def t_show_final_prompt(self, a): r = self._preview("prompt", args={"title": "Beispiel", "terminart": "", "projekt": "", "kunde": ""}) p = r["prompt"] # Namenslisten (personenbezogen) nicht an das Agenten-Modell geben marker = "Namensregister" if marker in p: p = p[:p.index(marker)] + "[Namensregister mit personenbezogenen Daten ausgeblendet]" return f"{r['chars']} Zeichen", p[:40_000] def t_run_prompt_test(self, a): if self.llm_calls >= 1: return "", "FEHLER: Pro Antwort ist nur ein Prompt-Test erlaubt." self.llm_calls += 1 from app.agent.testrun import prompt_test res = prompt_test(self.db, self.tenant, self.ws.cfg, self.ws.files, a.get("transcript")) if res["text"].strip(): self.ws.sample = res["text"] # das erzeugte Protokoll wird zum Beispiel der PDF-Vorschau → Prompt-Änderungen sind im PDF sichtbar return "Beispieltranskript", (f"[{res['model']}, {res['input_tokens']}→{res['output_tokens']} Tokens] Das Ergebnis ist jetzt das Beispielprotokoll " f"der PDF-Vorschau.\n{res['text'][:6000]}") def t_set_sample(self, a): self.ws.sample = a["text"] return "Beispielprotokoll", f"gesetzt ({len(a['text'])} Zeichen) – die PDF-Vorschau rendert jetzt dieses Protokoll"