268 lines
9.6 KiB
Python
268 lines
9.6 KiB
Python
"""
|
|
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", "items": {}, "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", "items": {}},
|
|
},
|
|
},
|
|
),
|
|
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)
|