from __future__ import annotations import json from typing import Any from mcp.server.fastmcp import FastMCP from ..config import DEFAULT_PROFILE from ..errors import QingflowApiError, backend_code_value_int, message_looks_like_invalid_token, raise_tool_error from ..list_type_labels import get_record_list_type_label, get_task_type_label from .base import ToolBase TASK_BOX_TO_TYPE = { "todo": 1, "initiated": 2, "cc": 3, "done": 5, } FLOW_STATUS_TO_PROCESS_STATUS = { "all": 1, "in_progress": 2, "approved": 3, "rejected": 4, "pending_fix": 5, "urged": 6, "overdue": 7, "due_soon": 8, "unread": 9, "ended": 10, } class TaskTools(ToolBase): """任务中心工具（中文名：任务列表与操作） 类型：流程任务运营工具。 提供对工作流任务的管理功能，包括： - 查询待办/已办列表 - 查看任务统计 - 标记已读/催办 - 查询节点和表单分组信息 消息类型 (type): - 1: 待办 (TODO) - 需要当前用户处理的任务 - 2: 我发起的 (INITIATED) - 当前用户发起的流程 - 3: 抄送 (CC) - 抄送给当前用户的任务 - 5: 已办 (DONE) - 当前用户已处理的任务 流程状态 (process_status): - 1: 全部 - 2: 流程中 - 3: 已通过 - 4: 已拒绝 - 5: 待完善 - 6: 催办 - 7: 超时 - 8: 即将超时 - 9: 未读 - 10: 流程结束 """ def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool() def task_summary( profile: str = DEFAULT_PROFILE, app_key: str | None = None, ) -> dict[str, Any]: return self.task_summary( profile=profile, app_key=app_key, ) @mcp.tool() def task_list( profile: str = DEFAULT_PROFILE, task_box: str = "todo", flow_status: str = "all", app_key: str | None = None, workflow_node_id: int | None = None, query: str | None = None, page: int = 1, page_size: int = 20, sort_by: str | None = None, sort_direction: str = "desc", ) -> dict[str, Any]: return self.task_list_public( profile=profile, task_box=task_box, flow_status=flow_status, app_key=app_key, workflow_node_id=workflow_node_id, query=query, page=page, page_size=page_size, sort_by=sort_by, sort_direction=sort_direction, ) @mcp.tool() def task_facets( profile: str = DEFAULT_PROFILE, task_box: str = "todo", flow_status: str = "all", dimension: str = "worksheet", app_key: str | None = None, query: str | None = None, limit: int = 50, ) -> dict[str, Any]: return self.task_facets( profile=profile, task_box=task_box, flow_status=flow_status, dimension=dimension, app_key=app_key, query=query, limit=limit, ) @mcp.tool() def task_mark_read( profile: str = DEFAULT_PROFILE, app_key: str = "", task_id: int = 0, task_box: str = "todo", ) -> dict[str, Any]: return self.task_mark_read_public(profile=profile, app_key=app_key, task_id=task_id, task_box=task_box) @mcp.tool() def task_mark_all_cc_read( profile: str = DEFAULT_PROFILE, flow_status: str = "all", ) -> dict[str, Any]: return self.task_mark_all_cc_read_public(profile=profile, flow_status=flow_status) @mcp.tool() def task_urge( profile: str = DEFAULT_PROFILE, app_key: str = "", record_id: int = 0, ) -> dict[str, Any]: return self.task_urge_public(profile=profile, app_key=app_key, record_id=record_id) def task_summary( self, *, profile: str, app_key: str | None, ) -> dict[str, Any]: """执行任务相关逻辑。""" try: raw = self.task_statistics(profile=profile, app_key=app_key) except RuntimeError as exc: if not _is_optional_task_runtime_error(exc): raise return self._runtime_error_as_auxiliary_result( exc, error_code="TASK_SUMMARY_UNAVAILABLE", selection={"app_key": app_key}, fallback_hint="Use task list/get directly; task summary is an auxiliary statistics entrypoint.", ) statistics = raw.get("statistics", {}) summary = self._normalize_task_summary_payload(statistics) return { "profile": profile, "ws_id": raw.get("ws_id"), "ok": True, "request_route": raw.get("request_route"), "warnings": [], "output_profile": "normal", "data": { "summary": summary, }, } def task_list_public( self, *, profile: str, task_box: str, flow_status: str, app_key: str | None, workflow_node_id: int | None, query: str | None, page: int, page_size: int, sort_by: str | None, sort_direction: str, ) -> dict[str, Any]: """执行任务相关逻辑。""" normalized_type = self._task_box_to_type(task_box) normalized_status = self._flow_status_to_process_status(flow_status) create_time_asc = self._task_sort_to_create_time_asc(sort_by, sort_direction) raw = self.task_list( profile=profile, type=normalized_type, process_status=normalized_status, app_key=app_key, node_id=workflow_node_id, search_key=query, page_num=page, page_size=page_size, create_time_asc=create_time_asc, ) task_page = raw.get("page", {}) return { "profile": profile, "ws_id": raw.get("ws_id"), "ok": True, "request_route": raw.get("request_route"), "warnings": [], "output_profile": "normal", "data": { "items": _task_page_items(task_page), "pagination": { "page": page, "page_size": page_size, "returned_items": len(_task_page_items(task_page)), "page_amount": _task_page_amount(task_page), "reported_total": _task_page_total(task_page), }, "selection": { "task_box": task_box, "flow_status": flow_status, "app_key": app_key, "workflow_node_id": workflow_node_id, "query": query, "applied_sort": self._task_applied_sort(sort_by, sort_direction), }, }, } def task_facets( self, *, profile: str, task_box: str, flow_status: str, dimension: str, app_key: str | None, query: str | None, limit: int, ) -> dict[str, Any]: """执行任务相关逻辑。""" normalized_type = self._task_box_to_type(task_box) normalized_status = self._flow_status_to_process_status(flow_status) if dimension not in {"worksheet", "workflow_node"}: raise_tool_error(QingflowApiError.config_error("dimension must be worksheet or workflow_node")) if limit <= 0: raise_tool_error(QingflowApiError.config_error("limit must be positive")) try: if dimension == "worksheet": raw = self.task_worksheet_statistics( profile=profile, type=normalized_type, worksheet_name=query, page_num=1, page_size=max(limit, 20), ) source = raw.get("page", {}) elif app_key: raw = self.task_node_statistics( profile=profile, app_key=app_key, type=normalized_type, search_key=query, ) source = raw.get("nodes", {}) else: raw = self.task_workflow_nodes( profile=profile, type=normalized_type, status=str(normalized_status), app_key_list=None, search_key=query, page_num=1, page_size=max(limit, 20), ) source = raw.get("page", {}) except RuntimeError as exc: if not _is_optional_task_runtime_error(exc): raise return self._runtime_error_as_auxiliary_result( exc, error_code="TASK_FACETS_UNAVAILABLE", selection={ "task_box": task_box, "flow_status": flow_status, "dimension": dimension, "app_key": app_key, "query": query, "limit": limit, }, fallback_hint="Use task list/get directly; task facets are an auxiliary grouping entrypoint.", ) groups = self._normalize_task_facets(source) rows_truncated = len(groups) > limit returned_groups = groups[:limit] return { "profile": profile, "ws_id": raw.get("ws_id"), "ok": True, "request_route": raw.get("request_route"), "warnings": [], "output_profile": "normal", "data": { "groups": returned_groups, "rows_truncated": rows_truncated, "statement_scope": "returned_groups_only" if rows_truncated else "full_population", "selection": { "task_box": task_box, "flow_status": flow_status, "dimension": dimension, "app_key": app_key, "query": query, "limit": limit, }, }, } def task_mark_read_public( self, *, profile: str, app_key: str, task_id: int, task_box: str, ) -> dict[str, Any]: """执行任务相关逻辑。""" raw = self.task_mark_read(profile=profile, app_key=app_key, id=task_id, type=self._task_box_to_type(task_box)) return self._public_task_action_response( raw, action="mark_read", resource={"app_key": app_key, "task_id": task_id}, selection={"task_box": task_box}, ) def task_mark_all_cc_read_public( self, *, profile: str, flow_status: str, ) -> dict[str, Any]: """执行任务相关逻辑。""" raw = self.task_mark_all_cc_read(profile=profile, type=TASK_BOX_TO_TYPE["cc"], process_status=self._flow_status_to_process_status(flow_status)) return self._public_task_action_response( raw, action="mark_all_cc_read", resource={}, selection={"task_box": "cc", "flow_status": flow_status}, ) def task_urge_public( self, *, profile: str, app_key: str, record_id: int, ) -> dict[str, Any]: """执行任务相关逻辑。""" raw = self.task_urge(profile=profile, app_key=app_key, row_record_id=record_id) return self._public_task_action_response( raw, action="urge", resource={"app_key": app_key, "record_id": record_id}, selection={}, ) def task_list( self, *, profile: str, type: int, process_status: int, app_key: str | None, node_id: int | None, search_key: str | None, page_num: int, page_size: int, create_time_asc: bool | None, ) -> dict[str, Any]: """执行任务相关逻辑。""" self._validate_type(type) self._validate_process_status(process_status) def runner(session_profile, context): payload: dict[str, Any] = { "type": type, "processStatus": process_status, "pageNum": page_num, "pageSize": page_size, } if app_key is not None: payload["appKey"] = app_key if node_id is not None: payload["nodeId"] = node_id if search_key: payload["searchKey"] = search_key if create_time_asc is not None: payload["createTimeAsc"] = create_time_asc result = self.backend.request("POST", context, "/task/dynamic/page", json_body=payload) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "type": type, "type_label": get_task_type_label(type), "list_type_label": get_task_type_label(type), "process_status": process_status, "page": result, } return self._run(profile, runner, tool_name='任务列表（兼容）') def task_list_grouped( self, *, profile: str, type: int, process_status: int, app_key: str | None, node_id: int | None, search_key: str | None, page_num: int, page_size: int, ) -> dict[str, Any]: """执行任务相关逻辑。""" self._validate_type(type) self._validate_process_status(process_status) def runner(session_profile, context): payload: dict[str, Any] = { "type": type, "processStatus": process_status, "pageNum": page_num, "pageSize": page_size, } if app_key is not None: payload["appKey"] = app_key if node_id is not None: payload["nodeId"] = node_id if search_key: payload["searchKey"] = search_key result = self.backend.request("POST", context, "/task/dynamic/page/group", json_body=payload) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "type": type, "type_label": get_task_type_label(type), "list_type_label": get_task_type_label(type), "process_status": process_status, "page": result, } return self._run(profile, runner, tool_name='任务分组列表') def task_statistics( self, *, profile: str, app_key: str | None, ) -> dict[str, Any]: """执行任务相关逻辑。""" def runner(session_profile, context): params: dict[str, Any] = {} if app_key: params["appKey"] = app_key result = self.backend.request("GET", context, "/task/dynamic/statics", params=params) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "statistics": result, } return self._run(profile, runner, tool_name='任务统计') def task_workflow_nodes( self, *, profile: str, type: int, status: str | None, app_key_list: list[str] | None, search_key: str | None, page_num: int, page_size: int, ) -> dict[str, Any]: """执行任务相关逻辑。""" self._validate_type(type) def runner(session_profile, context): params: dict[str, Any] = { "type": type, "pageNum": page_num, "pageSize": page_size, } if status is not None: params["status"] = status if app_key_list: params["appKeyList"] = app_key_list if search_key: params["searchKey"] = search_key result = self.backend.request("GET", context, "/task/dynamic/workflow/nodes", params=params) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "type": type, "type_label": get_task_type_label(type), "list_type_label": get_task_type_label(type), "page": result, } return self._run(profile, runner, tool_name='任务流程节点') def task_node_statistics( self, *, profile: str, app_key: str, type: int, search_key: str | None, ) -> dict[str, Any]: """执行任务相关逻辑。""" self._validate_type(type) if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) def runner(session_profile, context): params: dict[str, Any] = { "appKey": app_key, "type": type, } if search_key: params["searchKey"] = search_key result = self.backend.request("GET", context, "/task/dynamic/statics/node", params=params) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "app_key": app_key, "type": type, "type_label": get_task_type_label(type), "list_type_label": get_task_type_label(type), "nodes": result, } return self._run(profile, runner, tool_name='任务节点统计') def task_worksheet_statistics( self, *, profile: str, type: int, worksheet_name: str | None, page_num: int, page_size: int, ) -> dict[str, Any]: """执行任务相关逻辑。""" self._validate_type(type) def runner(session_profile, context): params: dict[str, Any] = { "type": type, "pageNum": page_num, "pageSize": page_size, } if worksheet_name: params["worksheetName"] = worksheet_name result = self.backend.request("GET", context, "/task/dynamic/statics/worksheet", params=params) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "type": type, "type_label": get_task_type_label(type), "list_type_label": get_task_type_label(type), "page": result, } return self._run(profile, runner, tool_name='任务应用统计') def task_mark_read( self, *, profile: str, app_key: str, id: int, type: int, ) -> dict[str, Any]: """执行任务相关逻辑。""" if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) if id <= 0: raise_tool_error(QingflowApiError.config_error("id must be positive")) self._validate_type(type) def runner(session_profile, context): result = self.backend.request( "POST", context, f"/task/dynamic/{app_key}/{id}/read/{type}", ) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "app_key": app_key, "id": id, "type": type, "type_label": get_task_type_label(type), "list_type_label": get_task_type_label(type), "result": result, } return self._run(profile, runner, tool_name='任务已读（兼容）') def task_mark_all_cc_read( self, *, profile: str, type: int, process_status: int, ) -> dict[str, Any]: """执行任务相关逻辑。""" self._validate_type(type) self._validate_process_status(process_status) def runner(session_profile, context): payload: dict[str, Any] = { "type": type, "processStatus": process_status, } result = self.backend.request("POST", context, "/task/dynamic/cc/readAll", json_body=payload) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "type": type, "type_label": get_task_type_label(type), "list_type_label": get_task_type_label(type), "process_status": process_status, "result": result, } return self._run(profile, runner, tool_name='抄送已读（兼容）') def task_urge( self, *, profile: str, app_key: str, row_record_id: int, ) -> dict[str, Any]: """执行任务相关逻辑。""" if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) if row_record_id <= 0: raise_tool_error(QingflowApiError.config_error("row_record_id must be positive")) def runner(session_profile, context): result = self.backend.request( "POST", context, f"/task/dynamic/{app_key}/{row_record_id}/urge", ) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "app_key": app_key, "row_record_id": row_record_id, "result": result, } return self._run(profile, runner, tool_name='任务催办（兼容）') def task_group_detail( self, *, profile: str, app_key: str, group_id: int, ) -> dict[str, Any]: """执行任务相关逻辑。""" if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) if group_id <= 0: raise_tool_error(QingflowApiError.config_error("group_id must be positive")) def runner(session_profile, context): result = self.backend.request( "GET", context, f"/task/dynamic/app/{app_key}/group/{group_id}/detail", ) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "app_key": app_key, "group_id": group_id, "detail": result, } return self._run(profile, runner, tool_name='任务分组详情') def task_batch_processing_amount( self, *, profile: str, app_key: str, list_type: int, task_center_filter: dict[str, Any] | None, ) -> dict[str, Any]: """执行任务相关逻辑。""" if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) def runner(session_profile, context): payload: dict[str, Any] = { "listType": list_type, } if task_center_filter is not None: payload["taskCenterFilter"] = task_center_filter result = self.backend.request( "POST", context, f"/task/app/{app_key}/batchProcessingAmount", json_body=payload, ) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "app_key": app_key, "list_type": list_type, "list_type_label": get_record_list_type_label(list_type), "amount": result, } return self._run(profile, runner, tool_name='批处理任务数量') def _validate_type(self, type: int) -> None: """执行内部辅助逻辑。""" valid_types = [1, 2, 3, 5] # TODO, INITIATED, CC, DONE if type not in valid_types: raise_tool_error( QingflowApiError.config_error( f"Invalid type: {type}. Must be one of {valid_types} (1=待办, 2=我发起的, 3=抄送, 5=已办)" ) ) def _validate_process_status(self, process_status: int) -> None: """执行内部辅助逻辑。""" valid_statuses = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] if process_status not in valid_statuses: raise_tool_error( QingflowApiError.config_error( f"Invalid process_status: {process_status}. Must be one of {valid_statuses}" ) ) def _task_box_to_type(self, task_box: str) -> int: """执行内部辅助逻辑。""" normalized = (task_box or "").strip().lower() if normalized not in TASK_BOX_TO_TYPE: raise_tool_error(QingflowApiError.config_error("task_box must be todo, initiated, cc, or done")) return TASK_BOX_TO_TYPE[normalized] def _flow_status_to_process_status(self, flow_status: str) -> int: """执行内部辅助逻辑。""" normalized = (flow_status or "").strip().lower() if normalized not in FLOW_STATUS_TO_PROCESS_STATUS: raise_tool_error( QingflowApiError.config_error( "flow_status must be all, in_progress, approved, rejected, pending_fix, urged, overdue, due_soon, unread, or ended" ) ) return FLOW_STATUS_TO_PROCESS_STATUS[normalized] def _task_sort_to_create_time_asc(self, sort_by: str | None, sort_direction: str) -> bool | None: """执行内部辅助逻辑。""" normalized_sort_by = (sort_by or "").strip().lower() if sort_by is not None else "" normalized_direction = (sort_direction or "desc").strip().lower() if normalized_direction not in {"asc", "desc"}: raise_tool_error(QingflowApiError.config_error("sort_direction must be asc or desc")) if not normalized_sort_by: return None if normalized_sort_by not in {"created_at", "create_time"}: raise_tool_error(QingflowApiError.config_error("task_list only supports sort_by=created_at")) return normalized_direction == "asc" def _task_applied_sort(self, sort_by: str | None, sort_direction: str) -> list[dict[str, Any]]: """执行内部辅助逻辑。""" if not sort_by: return [] return [{"by": "created_at", "order": (sort_direction or "desc").strip().lower()}] def _normalize_task_summary_payload(self, payload: Any) -> dict[str, Any]: """执行内部辅助逻辑。""" if not isinstance(payload, dict): return {} return { "todo_count": payload.get("todoCount", payload.get("todo_count")), "overdue_count": payload.get("timeoutCount", payload.get("timeout_count")), "due_soon_count": payload.get("preTimeoutCount", payload.get("pre_timeout_count")), "urged_count": payload.get("urgedCount", payload.get("urged_count")), "cc_unread_count": payload.get("ccUnreadCount", payload.get("cc_unread_count")), "in_progress_initiated_count": payload.get("processingCount", payload.get("processing_count")), } def _normalize_task_facets(self, payload: Any) -> list[dict[str, Any]]: """执行内部辅助逻辑。""" items = _task_page_items(payload) groups: list[dict[str, Any]] = [] for item in items: if not isinstance(item, dict): continue key = item.get("worksheetId", item.get("groupId", item.get("nodeId", item.get("id", item.get("key"))))) label = item.get("worksheetName", item.get("groupName", item.get("nodeName", item.get("name", item.get("label", item.get("title")))))) count = item.get("count", item.get("taskCount", item.get("amount", item.get("todoCount", item.get("num"))))) groups.append({"key": key if key is not None else label, "label": label, "count": count}) return groups def _runtime_error_as_auxiliary_result( self, error: RuntimeError, *, error_code: str, selection: dict[str, Any], fallback_hint: str, ) -> dict[str, Any]: """Return a structured failure for optional task discovery helpers.""" try: payload = json.loads(str(error)) except json.JSONDecodeError: payload = {"message": str(error)} details = payload.get("details") if isinstance(payload.get("details"), dict) else {} warning: dict[str, Any] = { "code": error_code, "message": fallback_hint, } for key in ("category", "backend_code", "request_id", "http_status"): if payload.get(key) is not None: warning[key] = payload.get(key) response: dict[str, Any] = { "ok": False, "status": "failed", "error_code": details.get("error_code") or error_code, "message": payload.get("message") or str(error), "warnings": [warning], "output_profile": "normal", "data": { "selection": selection, "fallback_hint": fallback_hint, }, } for key in ("category", "backend_code", "request_id", "http_status"): if payload.get(key) is not None: response[key] = payload.get(key) if details: response["details"] = details return response def _public_task_action_response( self, raw: dict[str, Any], *, action: str, resource: dict[str, Any], selection: dict[str, Any], ) -> dict[str, Any]: """执行内部辅助逻辑。""" response = dict(raw) response["ok"] = bool(raw.get("ok", True)) response["warnings"] = [] response["output_profile"] = "normal" response["data"] = { "action": action, "resource": resource, "selection": selection, "result": raw.get("result"), } return response def _request_route_payload(self, context) -> dict[str, Any]: # type: ignore[no-untyped-def] """执行内部辅助逻辑。""" describe_route = getattr(self.backend, "describe_route", None) if callable(describe_route): payload = describe_route(context) if isinstance(payload, dict): return payload return { "base_url": getattr(context, "base_url", None), "qf_version": getattr(context, "qf_version", None), "qf_version_source": getattr(context, "qf_version_source", None) or ("context" if getattr(context, "qf_version", None) else "unknown"), } def _is_optional_task_runtime_error(error: RuntimeError) -> bool: try: payload = json.loads(str(error)) except json.JSONDecodeError: return False if not isinstance(payload, dict): return False if str(payload.get("category") or "").strip().lower() == "auth": return False if message_looks_like_invalid_token(payload.get("message")): return False if backend_code_value_int(payload.get("http_status")) == 401: return False return backend_code_value_int(payload.get("backend_code")) in {40002, 40027, 404} or backend_code_value_int(payload.get("http_status")) == 404 def _task_page_items(payload: Any) -> list[Any]: if isinstance(payload, list): return payload if not isinstance(payload, dict): return [] for key in ("list", "items", "rows", "result"): value = payload.get(key) if isinstance(value, list): return value for container_key in ("page", "data"): nested = payload.get(container_key) if isinstance(nested, dict): nested_items = _task_page_items(nested) if nested_items: return nested_items return [] def _task_page_amount(payload: Any) -> Any: if isinstance(payload, dict): return payload.get("pageAmount", payload.get("page_amount")) return None def _task_page_total(payload: Any) -> Any: if isinstance(payload, dict): return payload.get("total", payload.get("count")) return None