import asyncio import requests SCRIPT_API_URL = "http://localhost:8082/run" SCRIPT_API_KEY = "DEIN_API_KEY" async def run_script( command: str, args: dict | None = None ): """ Führt einen Befehl über die Raspberry Pi Script API aus. """ try: def request(): headers = { "X-API-Key": SCRIPT_API_KEY, "Content-Type": "application/json" } payload = { "command": command, "args": args or {} } response = requests.post( SCRIPT_API_URL, json=payload, headers=headers, timeout=30 ) response.raise_for_status() return response.json() result = await asyncio.to_thread( request ) if result.get("ok"): return ( f"Befehl '{command}' erfolgreich ausgeführt.\n" f"Ausgabe:\n" f"{result.get('stdout','')}" ) else: return ( f"Befehl fehlgeschlagen.\n" f"Fehler:\n" f"{result.get('stderr','')}" ) except Exception as e: return f"Fehler beim Ausführen des Scripts: {e}" # ========================= # BEDROCK TOOL # ========================= script_api_tool = { "toolSpec": { "name": "run_script", "description": ( "Führt einen vordefinierten Raspberry Pi Befehl " "über die Script API aus. " "Nutze dieses Tool für Systemaktionen, " "Server starten/stoppen, Programme steuern " "und eigene Automationen." ), "inputSchema": { "json": { "type": "object", "properties": { "command": { "type": "string", "description": "ID des auszuführenden Befehls aus commands.json" }, "args": { "type": "object", "description": "Optionale Argumente für den Befehl" } }, "required": [ "command" ] } } } }