"""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)