""" Level 2 — Document inspection tools. Search, linked records, timeline, children, counts. """ import json from mcp.types import Tool from frappe_mcp.client.frappe_api import FrappeClient def tools() -> list[Tool]: return [ Tool( name="frappe_search_documents", description=( "Generic full-text search across any DocType. " "Searches title/name fields. Use filters for precise queries." ), inputSchema={ "type": "object", "required": ["doctype"], "properties": { "doctype": {"type": "string"}, "query": {"type": "string", "description": "Search text"}, "filters": {"type": "array", "description": "Additional filter conditions"}, "fields": {"type": "array", "items": {"type": "string"}}, "limit": {"type": "integer", "default": 20}, "order_by": {"type": "string", "default": "modified desc"}, }, }, ), Tool( name="frappe_get_document_children", description=( "Read child table rows for a document. " "Use get_doctype_meta to find the child table fieldname first." ), inputSchema={ "type": "object", "required": ["doctype", "name", "child_doctype"], "properties": { "doctype": {"type": "string", "description": "Parent DocType"}, "name": {"type": "string", "description": "Parent document name"}, "child_doctype": {"type": "string", "description": "Child table DocType e.g. 'Sales Order Item'"}, "fields": {"type": "array", "items": {"type": "string"}}, }, }, ), Tool( name="frappe_get_linked_documents", description=( "Returns all documents that link TO this document — " "e.g. all Sales Orders linked to a Customer, or all Invoices for a Sales Order." ), inputSchema={ "type": "object", "required": ["doctype", "name"], "properties": { "doctype": {"type": "string"}, "name": {"type": "string"}, "linked_doctype": { "type": "string", "description": "Filter to only this DocType (optional)", }, "limit": {"type": "integer", "default": 20}, }, }, ), Tool( name="frappe_get_document_timeline", description=( "Get the full activity timeline for a document: " "comments, emails, assignments, workflow changes, version history." ), inputSchema={ "type": "object", "required": ["doctype", "name"], "properties": { "doctype": {"type": "string"}, "name": {"type": "string"}, "limit": {"type": "integer", "default": 30}, }, }, ), Tool( name="frappe_get_recent_documents", description="Get recently modified documents of a DocType. Useful for 'show latest X' requests.", inputSchema={ "type": "object", "required": ["doctype"], "properties": { "doctype": {"type": "string"}, "fields": {"type": "array", "items": {"type": "string"}}, "limit": {"type": "integer", "default": 10}, "modified_after": {"type": "string", "description": "ISO datetime filter e.g. '2024-01-01'"}, }, }, ), Tool( name="frappe_count_documents", description="Count records matching filters. Fast — does not return document data.", inputSchema={ "type": "object", "required": ["doctype"], "properties": { "doctype": {"type": "string"}, "filters": {"type": "array"}, }, }, ), Tool( name="frappe_get_document_attachments", description="List all files attached to a specific document.", inputSchema={ "type": "object", "required": ["doctype", "name"], "properties": { "doctype": {"type": "string"}, "name": {"type": "string"}, }, }, ), Tool( name="frappe_get_document_versions", description="Get the change history (versions) of a document — who changed what fields and when.", inputSchema={ "type": "object", "required": ["doctype", "name"], "properties": { "doctype": {"type": "string"}, "name": {"type": "string"}, "limit": {"type": "integer", "default": 10}, }, }, ), ] def handlers() -> dict: return { "frappe_search_documents": _search_documents, "frappe_get_document_children": _get_document_children, "frappe_get_linked_documents": _get_linked_documents, "frappe_get_document_timeline": _get_document_timeline, "frappe_get_recent_documents": _get_recent_documents, "frappe_count_documents": _count_documents, "frappe_get_document_attachments": _get_document_attachments, "frappe_get_document_versions": _get_document_versions, } async def _search_documents(args: dict) -> str: client = FrappeClient() filters = list(args.get("filters") or []) if query := args.get("query"): filters.append(["name", "like", f"%{query}%"]) result = await client.get_list( args["doctype"], fields=args.get("fields", ["name", "modified", "owner"]), filters=filters if filters else None, limit=args.get("limit", 20), order_by=args.get("order_by", "modified desc"), ) return json.dumps(result, indent=2) async def _get_document_children(args: dict) -> str: client = FrappeClient() result = await client.get_list( args["child_doctype"], fields=args.get("fields", ["*"]) or ["*"], filters=[["parent", "=", args["name"]], ["parenttype", "=", args["doctype"]]], limit=500, order_by="idx asc", ) return json.dumps(result, indent=2) async def _get_linked_documents(args: dict) -> str: client = FrappeClient() result = await client.call_method( "frappe.client.get_linked_docs", doctype=args["doctype"], name=args["name"], linkinfo=args.get("linked_doctype", ""), ) return json.dumps(result, indent=2) async def _get_document_timeline(args: dict) -> str: client = FrappeClient() ref_dt = args["doctype"] ref_name = args["name"] limit = args.get("limit", 30) comments = await client.get_list( "Comment", fields=["name", "comment_type", "comment_by", "content", "creation"], filters=[["reference_doctype", "=", ref_dt], ["reference_name", "=", ref_name]], limit=limit, order_by="creation desc", ) versions = await client.get_list( "Version", fields=["name", "owner", "creation", "data"], filters=[["ref_doctype", "=", ref_dt], ["docname", "=", ref_name]], limit=limit, order_by="creation desc", ) timeline = sorted( [{"type": "comment", **c} for c in (comments if isinstance(comments, list) else [])] + [{"type": "version", **v} for v in (versions if isinstance(versions, list) else [])], key=lambda x: x.get("creation", ""), reverse=True, )[:limit] return json.dumps(timeline, indent=2) async def _get_recent_documents(args: dict) -> str: client = FrappeClient() filters = [] if after := args.get("modified_after"): filters.append(["modified", ">", after]) result = await client.get_list( args["doctype"], fields=args.get("fields", ["name", "modified", "owner", "docstatus"]), filters=filters if filters else None, limit=args.get("limit", 10), order_by="modified desc", ) return json.dumps(result, indent=2) async def _count_documents(args: dict) -> str: client = FrappeClient() result = await client.call_method( "frappe.client.get_count", doctype=args["doctype"], filters=args.get("filters"), ) return json.dumps({"doctype": args["doctype"], "count": result}, indent=2) async def _get_document_attachments(args: dict) -> str: client = FrappeClient() result = await client.get_list( "File", fields=["name", "file_name", "file_url", "file_size", "is_private", "creation"], filters=[ ["attached_to_doctype", "=", args["doctype"]], ["attached_to_name", "=", args["name"]], ], limit=50, ) return json.dumps(result, indent=2) async def _get_document_versions(args: dict) -> str: client = FrappeClient() result = await client.get_list( "Version", fields=["name", "owner", "creation", "data"], filters=[["ref_doctype", "=", args["doctype"]], ["docname", "=", args["name"]]], limit=args.get("limit", 10), order_by="creation desc", ) return json.dumps(result, indent=2)