""" System Handler - Settings, Server, Schema, Activity, Presets, Notifications Phase J.4: 10 tools - Settings: get, update (2) - Server: info, health, graphql_sdl, openapi_spec (4) - Schema: snapshot, diff, apply (3) - Activity: list (1) """ import json from typing import Any from plugins.directus.client import DirectusClient def get_tool_specifications() -> list[dict[str, Any]]: """Return tool specifications for ToolGenerator (10 tools)""" return [ # ===================== # SETTINGS (2) # ===================== { "name": "get_settings", "method_name": "get_settings", "description": "Get system settings.", "schema": {"type": "object", "properties": {}, "required": []}, "scope": "read", }, { "name": "update_settings", "method_name": "update_settings", "description": "Update system settings (project name, logo, colors, etc.).", "schema": { "type": "object", "properties": {"data": {"type": "object", "description": "Settings to update"}}, "required": ["data"], }, "scope": "admin", }, # ===================== # SERVER (4) # ===================== { "name": "get_server_info", "method_name": "get_server_info", "description": "Get Directus server information (version, extensions, etc.).", "schema": {"type": "object", "properties": {}, "required": []}, "scope": "read", }, { "name": "health_check", "method_name": "health_check", "description": "Check Directus server health status.", "schema": {"type": "object", "properties": {}, "required": []}, "scope": "read", }, { "name": "get_graphql_sdl", "method_name": "get_graphql_sdl", "description": "Get the GraphQL SDL schema.", "schema": {"type": "object", "properties": {}, "required": []}, "scope": "read", }, { "name": "get_openapi_spec", "method_name": "get_openapi_spec", "description": "Get the OpenAPI specification.", "schema": {"type": "object", "properties": {}, "required": []}, "scope": "read", }, # ===================== # SCHEMA (3) # ===================== { "name": "get_schema_snapshot", "method_name": "get_schema_snapshot", "description": "Get complete schema snapshot for migration/backup.", "schema": {"type": "object", "properties": {}, "required": []}, "scope": "admin", }, { "name": "schema_diff", "method_name": "schema_diff", "description": "Get diff between current schema and provided snapshot.", "schema": { "type": "object", "properties": { "snapshot": { "type": "object", "description": "Schema snapshot to compare against", } }, "required": ["snapshot"], }, "scope": "admin", }, { "name": "schema_apply", "method_name": "schema_apply", "description": "Apply schema diff to database.", "schema": { "type": "object", "properties": {"diff": {"type": "object", "description": "Schema diff to apply"}}, "required": ["diff"], }, "scope": "admin", }, # ===================== # ACTIVITY (1) # ===================== { "name": "list_activity", "method_name": "list_activity", "description": "List activity log (all actions performed in Directus).", "schema": { "type": "object", "properties": { "filter": { "anyOf": [{"type": "object"}, {"type": "null"}], "description": 'Filter object (e.g., {"action": {"_eq": "create"}})', }, "sort": { "anyOf": [{"type": "array", "items": {"type": "string"}}, {"type": "null"}], "description": "Sort fields (default: ['-timestamp'])", }, "limit": { "type": "integer", "description": "Maximum activities to return", "default": 100, }, }, "required": [], }, "scope": "read", }, ] # ===================== # HANDLER FUNCTIONS # ===================== async def get_settings(client: DirectusClient) -> str: """Get system settings.""" try: result = await client.get_settings() return json.dumps( {"success": True, "settings": result.get("data")}, indent=2, ensure_ascii=False ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def update_settings(client: DirectusClient, data: dict[str, Any]) -> str: """Update system settings.""" try: result = await client.update_settings(data) return json.dumps( {"success": True, "message": "Settings updated", "settings": result.get("data")}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def get_server_info(client: DirectusClient) -> str: """Get server info.""" try: result = await client.get_server_info() return json.dumps( {"success": True, "server": result.get("data")}, indent=2, ensure_ascii=False ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def health_check(client: DirectusClient) -> str: """Check server health.""" try: result = await client.health_check() return json.dumps( {"success": True, "status": result.get("status", "unknown"), "health": result}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def get_graphql_sdl(client: DirectusClient) -> str: """Get GraphQL SDL.""" try: result = await client.get_graphql_sdl() # GraphQL SDL is usually returned as text if isinstance(result, dict): sdl = result.get("data", result) else: sdl = result return json.dumps({"success": True, "sdl": sdl}, indent=2, ensure_ascii=False) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def get_openapi_spec(client: DirectusClient) -> str: """Get OpenAPI specification.""" try: result = await client.get_openapi_spec() return json.dumps({"success": True, "spec": result}, indent=2, ensure_ascii=False) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def get_schema_snapshot(client: DirectusClient) -> str: """Get schema snapshot.""" try: result = await client.get_schema_snapshot() return json.dumps( {"success": True, "snapshot": result.get("data")}, indent=2, ensure_ascii=False ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def schema_diff(client: DirectusClient, snapshot: dict[str, Any]) -> str: """Get schema diff.""" try: result = await client.schema_diff(snapshot) return json.dumps( {"success": True, "diff": result.get("data")}, indent=2, ensure_ascii=False ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def schema_apply(client: DirectusClient, diff: dict[str, Any]) -> str: """Apply schema diff.""" try: result = await client.schema_apply(diff) return json.dumps( { "success": True, "message": "Schema applied", "result": result.get("data") if result else None, }, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def list_activity( client: DirectusClient, filter: dict | None = None, sort: list[str] | None = None, limit: int = 100, ) -> str: """List activity log.""" try: result = await client.list_activity(filter=filter, sort=sort, limit=limit) activities = result.get("data", []) return json.dumps( {"success": True, "total": len(activities), "activities": activities}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2)