MCP-Frappe/frappe_mcp/tools/admin.py

"""Admin / bench tools — cache, migrate, app management, system settings."""

import json
from mcp.types import Tool
from frappe_mcp.client.frappe_api import FrappeClient
from frappe_mcp.config import get_settings


def tools() -> list[Tool]:
    return [
        Tool(
            name="frappe_clear_cache",
            description="Clear Frappe's server-side cache.",
            inputSchema={"type": "object", "properties": {}},
        ),
        Tool(
            name="frappe_get_installed_apps",
            description="List all Frappe apps installed on the site.",
            inputSchema={"type": "object", "properties": {}},
        ),
        Tool(
            name="frappe_get_system_settings",
            description="Get the current System Settings document.",
            inputSchema={"type": "object", "properties": {}},
        ),
        Tool(
            name="frappe_update_system_settings",
            description="Update System Settings fields.",
            inputSchema={
                "type": "object",
                "required": ["updates"],
                "properties": {
                    "updates": {
                        "type": "object",
                        "description": "Fields to update e.g. {\"language\": \"en\", \"time_zone\": \"Asia/Kolkata\"}",
                    },
                },
            },
        ),
        Tool(
            name="frappe_run_report",
            description="Run a Script Report or Query Report and return results.",
            inputSchema={
                "type": "object",
                "required": ["report_name"],
                "properties": {
                    "report_name": {"type": "string"},
                    "filters": {
                        "type": "object",
                        "description": "Report filters as key-value pairs",
                    },
                },
            },
        ),
        Tool(
            name="frappe_execute_query",
            description=(
                "Execute a read-only SQL query on the Frappe database via frappe.db.sql. "
                "Only SELECT statements are allowed."
            ),
            inputSchema={
                "type": "object",
                "required": ["query"],
                "properties": {
                    "query": {"type": "string", "description": "SQL SELECT query"},
                    "values": {
                        "type": "array",
                        "description": "Parameterized values for %s placeholders",
                    },
                },
            },
        ),
        Tool(
            name="frappe_get_site_info",
            description="Get basic site information: Frappe version, installed apps, site config.",
            inputSchema={"type": "object", "properties": {}},
        ),
        Tool(
            name="frappe_create_workflow",
            description="Create a Workflow for a DocType with states and transitions.",
            inputSchema={
                "type": "object",
                "required": ["workflow_name", "document_type", "states", "transitions"],
                "properties": {
                    "workflow_name": {"type": "string"},
                    "document_type": {"type": "string"},
                    "is_active": {"type": "integer", "default": 1},
                    "states": {
                        "type": "array",
                        "description": "List of workflow state objects with: state, doc_status, allow_edit",
                        "items": {"type": "object"},
                    },
                    "transitions": {
                        "type": "array",
                        "description": "List of transition objects with: state, action, next_state, allowed",
                        "items": {"type": "object"},
                    },
                    "send_email_alert": {"type": "integer", "default": 0},
                },
            },
        ),
    ]


def handlers() -> dict:
    return {
        "frappe_clear_cache": _clear_cache,
        "frappe_get_installed_apps": _get_installed_apps,
        "frappe_get_system_settings": _get_system_settings,
        "frappe_update_system_settings": _update_system_settings,
        "frappe_run_report": _run_report,
        "frappe_execute_query": _execute_query,
        "frappe_get_site_info": _get_site_info,
        "frappe_create_workflow": _create_workflow,
    }


async def _clear_cache(args: dict) -> str:
    client = FrappeClient()
    result = await client.call_method("frappe.sessions.clear")
    return json.dumps(result, indent=2)


async def _get_installed_apps(args: dict) -> str:
    client = FrappeClient()
    result = await client.call_method("frappe.utils.change_log.get_versions")
    return json.dumps(result, indent=2)


async def _get_system_settings(args: dict) -> str:
    client = FrappeClient()
    result = await client.get_doc("System Settings", "System Settings")
    return json.dumps(result, indent=2)


async def _update_system_settings(args: dict) -> str:
    if get_settings().read_only_mode:
        return "Error: read_only_mode is enabled."
    client = FrappeClient()
    result = await client.update_doc("System Settings", "System Settings", args["updates"])
    return json.dumps(result, indent=2)


async def _run_report(args: dict) -> str:
    client = FrappeClient()
    result = await client.call_method(
        "frappe.desk.query_report.run",
        report_name=args["report_name"],
        filters=args.get("filters", {}),
    )
    return json.dumps(result, indent=2)


async def _execute_query(args: dict) -> str:
    query = args["query"].strip().lower()
    if not query.startswith("select"):
        return "Error: only SELECT queries are allowed."
    client = FrappeClient()
    result = await client.call_method(
        "frappe.client.get_list",
        doctype="__query__",
        query=args["query"],
        values=args.get("values", []),
    )
    return json.dumps(result, indent=2)


async def _get_site_info(args: dict) -> str:
    client = FrappeClient()
    result = await client.call_method("frappe.utils.change_log.get_versions")
    return json.dumps(result, indent=2)


async def _create_workflow(args: dict) -> str:
    client = FrappeClient()
    payload = {
        "workflow_name": args["workflow_name"],
        "document_type": args["document_type"],
        "is_active": args.get("is_active", 1),
        "send_email_alert": args.get("send_email_alert", 0),
        "states": args["states"],
        "transitions": args["transitions"],
    }
    result = await client.create_doc("Workflow", payload)
    return json.dumps(result, indent=2)