MCP-Frappe/frappe_mcp/tools/reports.py

"""Report CRUD tools — create Query Reports and Script Reports."""

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


def tools() -> list[Tool]:
    return [
        Tool(
            name="frappe_list_reports",
            description="List Reports, optionally filtered by type or DocType.",
            inputSchema={
                "type": "object",
                "properties": {
                    "report_type": {
                        "type": "string",
                        "enum": ["Query Report", "Script Report", "Report Builder", "Custom Report"],
                    },
                    "ref_doctype": {"type": "string"},
                    "limit": {"type": "integer", "default": 30},
                },
            },
        ),
        Tool(
            name="frappe_get_report",
            description="Get a Report definition including its query or script.",
            inputSchema={
                "type": "object",
                "required": ["name"],
                "properties": {"name": {"type": "string"}},
            },
        ),
        Tool(
            name="frappe_create_query_report",
            description=(
                "Create a Query Report using a raw SQL SELECT statement. "
                "The query can use %(filter_field)s for parameterized filters."
            ),
            inputSchema={
                "type": "object",
                "required": ["name", "ref_doctype", "query"],
                "properties": {
                    "name": {"type": "string"},
                    "ref_doctype": {"type": "string"},
                    "query": {"type": "string", "description": "SQL SELECT query"},
                    "filters": {
                        "type": "array",
                        "items": {"type": "object"},
                        "description": "Filter field definitions",
                    },
                    "is_standard": {"type": "string", "enum": ["Yes", "No"], "default": "No"},
                    "disabled": {"type": "integer", "default": 0},
                },
            },
        ),
        Tool(
            name="frappe_create_script_report",
            description=(
                "Create a Script Report using Python. "
                "The script must define `execute(filters)` that returns (columns, data)."
            ),
            inputSchema={
                "type": "object",
                "required": ["name", "ref_doctype", "script"],
                "properties": {
                    "name": {"type": "string"},
                    "ref_doctype": {"type": "string"},
                    "script": {"type": "string", "description": "Python script with execute(filters) function"},
                    "javascript": {"type": "string", "description": "Optional JS for filters/formatting"},
                    "filters": {
                        "type": "array",
                        "items": {"type": "object"},
                        "description": "Filter field definitions",
                    },
                    "is_standard": {"type": "string", "enum": ["Yes", "No"], "default": "No"},
                    "disabled": {"type": "integer", "default": 0},
                },
            },
        ),
        Tool(
            name="frappe_update_report",
            description="Update a Report's query, script, or filters.",
            inputSchema={
                "type": "object",
                "required": ["name"],
                "properties": {
                    "name": {"type": "string"},
                    "query": {"type": "string"},
                    "script": {"type": "string"},
                    "javascript": {"type": "string"},
                    "disabled": {"type": "integer"},
                    "updates": {"type": "object"},
                },
            },
        ),
        Tool(
            name="frappe_run_query_report",
            description="Execute a report and return its data.",
            inputSchema={
                "type": "object",
                "required": ["report_name"],
                "properties": {
                    "report_name": {"type": "string"},
                    "filters": {"type": "object", "description": "Filter key-value pairs"},
                },
            },
        ),
    ]


def handlers() -> dict:
    return {
        "frappe_list_reports": _list_reports,
        "frappe_get_report": _get_report,
        "frappe_create_query_report": _create_query_report,
        "frappe_create_script_report": _create_script_report,
        "frappe_update_report": _update_report,
        "frappe_run_query_report": _run_query_report,
    }


async def _list_reports(args: dict) -> str:
    client = FrappeClient()
    filters = []
    if rt := args.get("report_type"):
        filters.append(["report_type", "=", rt])
    if dt := args.get("ref_doctype"):
        filters.append(["ref_doctype", "=", dt])
    result = await client.get_list(
        "Report",
        fields=["name", "report_type", "ref_doctype", "disabled", "is_standard"],
        filters=filters if filters else None,
        limit=args.get("limit", 30),
    )
    return json.dumps(result, indent=2)

async def _get_report(args: dict) -> str:
    client = FrappeClient()
    result = await client.get_doc("Report", args["name"])
    return json.dumps(result, indent=2)

async def _create_query_report(args: dict) -> str:
    client = FrappeClient()
    payload = {
        "report_name": args["name"],
        "report_type": "Query Report",
        "ref_doctype": args["ref_doctype"],
        "query": args["query"],
        "filters": args.get("filters", []),
        "is_standard": args.get("is_standard", "No"),
        "disabled": args.get("disabled", 0),
    }
    result = await client.create_doc("Report", payload)
    return json.dumps(result, indent=2)

async def _create_script_report(args: dict) -> str:
    client = FrappeClient()
    payload = {
        "report_name": args["name"],
        "report_type": "Script Report",
        "ref_doctype": args["ref_doctype"],
        "script": args["script"],
        "javascript": args.get("javascript", ""),
        "filters": args.get("filters", []),
        "is_standard": args.get("is_standard", "No"),
        "disabled": args.get("disabled", 0),
    }
    result = await client.create_doc("Report", payload)
    return json.dumps(result, indent=2)

async def _update_report(args: dict) -> str:
    client = FrappeClient()
    updates = args.get("updates", {})
    for field in ("query", "script", "javascript", "disabled"):
        if field in args:
            updates[field] = args[field]
    result = await client.update_doc("Report", args["name"], updates)
    return json.dumps(result, indent=2)

async def _run_query_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)