MCP-Frappe/frappe_mcp/tools/analytics.py

"""
Level 6 — Reporting and analytics tools.
Aggregate queries, dashboard data, PDF rendering.
"""

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


def tools() -> list[Tool]:
    return [
        Tool(
            name="frappe_aggregate_documents",
            description=(
                "Aggregate document data: sum, avg, min, max, or count grouped by a field. "
                "e.g. total sales per customer, count of issues per status."
            ),
            inputSchema={
                "type": "object",
                "required": ["doctype", "function", "field"],
                "properties": {
                    "doctype": {"type": "string"},
                    "function": {
                        "type": "string",
                        "enum": ["sum", "avg", "min", "max", "count"],
                    },
                    "field": {"type": "string", "description": "Field to aggregate e.g. 'grand_total'"},
                    "group_by": {"type": "string", "description": "Field to group by e.g. 'customer'"},
                    "filters": {"type": "array", "items": {}},
                    "limit": {"type": "integer", "default": 50},
                },
            },
        ),
        Tool(
            name="frappe_get_dashboard_data",
            description=(
                "Pull data for a Dashboard Chart — returns the chart's data points "
                "for display as a summary or KPI."
            ),
            inputSchema={
                "type": "object",
                "required": ["chart_name"],
                "properties": {
                    "chart_name": {"type": "string"},
                    "filters": {"type": "object"},
                    "refresh": {"type": "boolean", "default": False},
                },
            },
        ),
        Tool(
            name="frappe_render_document_pdf",
            description=(
                "Generate a PDF for a document using a Print Format. "
                "Returns the PDF download URL on the Frappe server."
            ),
            inputSchema={
                "type": "object",
                "required": ["doctype", "name"],
                "properties": {
                    "doctype": {"type": "string"},
                    "name": {"type": "string"},
                    "print_format": {"type": "string", "description": "Print Format name (optional, uses default if empty)"},
                    "letterhead": {"type": "string"},
                    "language": {"type": "string", "default": "en"},
                },
            },
        ),
        Tool(
            name="frappe_get_number_card_value",
            description="Get the current value of a Number Card (single KPI metric).",
            inputSchema={
                "type": "object",
                "required": ["card_name"],
                "properties": {
                    "card_name": {"type": "string"},
                    "filters": {"type": "object"},
                },
            },
        ),
    ]


def handlers() -> dict:
    return {
        "frappe_aggregate_documents": _aggregate_documents,
        "frappe_get_dashboard_data": _get_dashboard_data,
        "frappe_render_document_pdf": _render_document_pdf,
        "frappe_get_number_card_value": _get_number_card_value,
    }


async def _aggregate_documents(args: dict) -> str:
    client = FrappeClient()
    func = args["function"].upper()
    field = args["field"]
    group_by = args.get("group_by")
    filters = args.get("filters")

    if group_by:
        sql = f"SELECT `{group_by}`, {func}(`{field}`) as value FROM `tab{args['doctype']}`"
        conditions = []
        if filters and isinstance(filters, list):
            for f in filters:
                if len(f) >= 3:
                    conditions.append(f"`{f[0]}` {f[1]} '{f[2]}'")
        if conditions:
            sql += " WHERE " + " AND ".join(conditions)
        sql += f" GROUP BY `{group_by}` ORDER BY value DESC LIMIT {args.get('limit', 50)}"
        result = await client.call_method("frappe.client.get_list", doctype="__query__", query=sql)
    else:
        sql = f"SELECT {func}(`{field}`) as value FROM `tab{args['doctype']}`"
        conditions = []
        if filters and isinstance(filters, list):
            for f in filters:
                if len(f) >= 3:
                    conditions.append(f"`{f[0]}` {f[1]} '{f[2]}'")
        if conditions:
            sql += " WHERE " + " AND ".join(conditions)
        result = await client.call_method(
            "frappe.desk.query_report.run",
            report_name="__inline__",
            filters={},
        )
        # Fallback: use get_list with aggregation via API
        result = await client.call_method(
            "frappe.client.get_value",
            doctype=args["doctype"],
            filters=filters,
            fieldname=f"{func.lower()}({field})",
        )
    return json.dumps({"function": func, "field": field, "group_by": group_by, "result": result}, indent=2)


async def _get_dashboard_data(args: dict) -> str:
    client = FrappeClient()
    result = await client.call_method(
        "frappe.desk.doctype.dashboard_chart.dashboard_chart.get",
        chart_name=args["chart_name"],
        filters=json.dumps(args.get("filters", {})),
        refresh=args.get("refresh", False),
    )
    return json.dumps(result, indent=2)


async def _render_document_pdf(args: dict) -> str:
    client = FrappeClient()
    params = {
        "doctype": args["doctype"],
        "name": args["name"],
        "print_format": args.get("print_format", ""),
        "letterhead": args.get("letterhead", ""),
        "lang": args.get("language", "en"),
    }
    pdf_url = (
        f"{client.base_url}/api/method/frappe.utils.pdf.download_pdf"
        f"?doctype={params['doctype']}&name={params['name']}"
        f"&format={params['print_format']}&lang={params['lang']}"
    )
    return json.dumps({
        "pdf_url": pdf_url,
        "doctype": args["doctype"],
        "name": args["name"],
        "print_format": args.get("print_format", "default"),
        "note": "Open pdf_url in browser or download directly from Frappe server",
    }, indent=2)


async def _get_number_card_value(args: dict) -> str:
    client = FrappeClient()
    result = await client.call_method(
        "frappe.desk.doctype.number_card.number_card.get_result",
        doc={"name": args["card_name"]},
        filters=json.dumps(args.get("filters", {})),
    )
    return json.dumps({"card": args["card_name"], "value": result}, indent=2)