MCP-Frappe/frappe_mcp/tools/workflow_tools.py

"""
Level 5 — Workflow and approval tools.
List workflows, read state, available transitions, execute transitions, approval summary.
"""

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


def tools() -> list[Tool]:
    return [
        Tool(
            name="frappe_list_workflows",
            description="List all Workflows, optionally filtered by DocType.",
            inputSchema={
                "type": "object",
                "properties": {
                    "document_type": {"type": "string"},
                    "is_active": {"type": "integer", "description": "1 = active only"},
                },
            },
        ),
        Tool(
            name="frappe_get_workflow",
            description="Get full workflow definition: all states and all transitions.",
            inputSchema={
                "type": "object",
                "required": ["workflow_name"],
                "properties": {"workflow_name": {"type": "string"}},
            },
        ),
        Tool(
            name="frappe_get_workflow_state",
            description="Get the current workflow state of a specific document.",
            inputSchema={
                "type": "object",
                "required": ["doctype", "name"],
                "properties": {
                    "doctype": {"type": "string"},
                    "name": {"type": "string"},
                },
            },
        ),
        Tool(
            name="frappe_list_workflow_transitions",
            description=(
                "List the workflow transitions available for a document in its current state. "
                "Shows what actions the current user can take: Approve, Reject, Send Back, etc."
            ),
            inputSchema={
                "type": "object",
                "required": ["doctype", "name"],
                "properties": {
                    "doctype": {"type": "string"},
                    "name": {"type": "string"},
                },
            },
        ),
        Tool(
            name="frappe_execute_workflow_transition",
            description=(
                "Perform a workflow transition on a document — e.g. Approve, Reject, Send Back. "
                "Use list_workflow_transitions first to see allowed actions."
            ),
            inputSchema={
                "type": "object",
                "required": ["doctype", "name", "action"],
                "properties": {
                    "doctype": {"type": "string"},
                    "name": {"type": "string"},
                    "action": {"type": "string", "description": "Transition action name e.g. 'Approve'"},
                },
            },
        ),
        Tool(
            name="frappe_get_approval_summary",
            description=(
                "Get a summary of pending approvals for a DocType: "
                "who needs to approve, current state, and how many documents are waiting."
            ),
            inputSchema={
                "type": "object",
                "required": ["doctype"],
                "properties": {
                    "doctype": {"type": "string"},
                    "workflow_state": {"type": "string", "description": "Filter by specific state e.g. 'Pending Approval'"},
                    "limit": {"type": "integer", "default": 20},
                },
            },
        ),
        Tool(
            name="frappe_update_workflow",
            description="Enable/disable a workflow or update its properties.",
            inputSchema={
                "type": "object",
                "required": ["workflow_name"],
                "properties": {
                    "workflow_name": {"type": "string"},
                    "is_active": {"type": "integer"},
                    "updates": {"type": "object"},
                },
            },
        ),
    ]


def handlers() -> dict:
    return {
        "frappe_list_workflows": _list_workflows,
        "frappe_get_workflow": _get_workflow,
        "frappe_get_workflow_state": _get_workflow_state,
        "frappe_list_workflow_transitions": _list_workflow_transitions,
        "frappe_execute_workflow_transition": _execute_workflow_transition,
        "frappe_get_approval_summary": _get_approval_summary,
        "frappe_update_workflow": _update_workflow,
    }


async def _list_workflows(args: dict) -> str:
    client = FrappeClient()
    filters = []
    if dt := args.get("document_type"):
        filters.append(["document_type", "=", dt])
    if "is_active" in args:
        filters.append(["is_active", "=", args["is_active"]])
    result = await client.get_list(
        "Workflow",
        fields=["name", "document_type", "is_active", "workflow_state_field"],
        filters=filters if filters else None,
        limit=50,
    )
    return json.dumps(result, indent=2)


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


async def _get_workflow_state(args: dict) -> str:
    client = FrappeClient()
    doc = await client.get_doc(args["doctype"], args["name"])
    workflow_state = doc.get("workflow_state")
    workflow_field = "workflow_state"
    workflows = await client.get_list(
        "Workflow",
        fields=["name", "workflow_state_field", "is_active"],
        filters=[["document_type", "=", args["doctype"]], ["is_active", "=", 1]],
        limit=1,
    )
    if isinstance(workflows, list) and workflows:
        workflow_field = workflows[0].get("workflow_state_field", "workflow_state")
        workflow_state = doc.get(workflow_field)
    return json.dumps({
        "doctype": args["doctype"],
        "name": args["name"],
        "workflow_state_field": workflow_field,
        "workflow_state": workflow_state,
        "docstatus": doc.get("docstatus", 0),
    }, indent=2)


async def _list_workflow_transitions(args: dict) -> str:
    client = FrappeClient()
    result = await client.call_method(
        "frappe.model.workflow.get_transitions",
        doc={"doctype": args["doctype"], "name": args["name"]},
    )
    return json.dumps(result, indent=2)


async def _execute_workflow_transition(args: dict) -> str:
    client = FrappeClient()
    result = await client.call_method(
        "frappe.model.workflow.apply_workflow",
        doc={"doctype": args["doctype"], "name": args["name"]},
        action=args["action"],
    )
    return json.dumps(result, indent=2)


async def _get_approval_summary(args: dict) -> str:
    client = FrappeClient()
    filters = []
    if ws := args.get("workflow_state"):
        filters.append(["workflow_state", "=", ws])
    docs = await client.get_list(
        args["doctype"],
        fields=["name", "workflow_state", "owner", "modified", "docstatus"],
        filters=filters if filters else None,
        limit=args.get("limit", 20),
        order_by="modified desc",
    )
    docs_list = docs if isinstance(docs, list) else []
    count = await client.call_method(
        "frappe.client.get_count",
        doctype=args["doctype"],
        filters=filters if filters else None,
    )
    return json.dumps({
        "doctype": args["doctype"],
        "total_pending": count,
        "documents": docs_list,
    }, indent=2)


async def _update_workflow(args: dict) -> str:
    client = FrappeClient()
    updates = args.get("updates", {})
    if "is_active" in args:
        updates["is_active"] = args["is_active"]
    result = await client.update_doc("Workflow", args["workflow_name"], updates)
    return json.dumps(result, indent=2)