from __future__ import annotations from mcp.server.fastmcp import FastMCP from ..config import DEFAULT_PROFILE from ..errors import QingflowApiError, backend_code_int, is_auth_like_error, raise_tool_error from ..json_types import JSONObject, JSONValue from ..list_type_labels import SYSTEM_VIEW_DEFINITIONS from .base import ToolBase class ViewTools(ToolBase): """视图工具（中文名：视图查询与配置读取）。 类型：视图配置工具。 主要职责： 1. 查询应用视图列表； 2. 获取视图配置与结构信息； 3. 为记录查询和分析提供视图侧元数据。 """ def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool() def view_list(profile: str = DEFAULT_PROFILE, app_key: str = "") -> JSONObject: return self.view_list(profile=profile, app_key=app_key) @mcp.tool() def view_list_flat(profile: str = DEFAULT_PROFILE, app_key: str = "") -> JSONObject: return self.view_list_flat(profile=profile, app_key=app_key) @mcp.tool() def view_get_config(profile: str = DEFAULT_PROFILE, viewgraph_key: str = "") -> JSONObject: return self.view_get_config(profile=profile, viewgraph_key=viewgraph_key) @mcp.tool() def view_get_base_info(profile: str = DEFAULT_PROFILE, viewgraph_key: str = "", passcode: str | None = None) -> JSONObject: return self.view_get_base_info(profile=profile, viewgraph_key=viewgraph_key, passcode=passcode) @mcp.tool() def view_list_questions(profile: str = DEFAULT_PROFILE, viewgraph_key: str = "") -> JSONObject: return self.view_list_questions(profile=profile, viewgraph_key=viewgraph_key) @mcp.tool() def view_list_associations(profile: str = DEFAULT_PROFILE, viewgraph_key: str = "") -> JSONObject: return self.view_list_associations(profile=profile, viewgraph_key=viewgraph_key) @mcp.tool() def view_board_get_lane_base_info(profile: str = DEFAULT_PROFILE, viewgraph_key: str = "", page_num: int = 1, page_size: int = 20) -> JSONObject: return self.view_board_get_lane_base_info(profile=profile, viewgraph_key=viewgraph_key, page_num=page_num, page_size=page_size) @mcp.tool() def view_get_future_nodes(profile: str = DEFAULT_PROFILE, viewgraph_key: str = "", apply_id: int = 0, app_key: str = "") -> JSONObject: return self.view_get_future_nodes(profile=profile, viewgraph_key=viewgraph_key, apply_id=apply_id, app_key=app_key) @mcp.tool() def view_get_workflow_status(profile: str = DEFAULT_PROFILE, viewgraph_key: str = "", row_record_id: int = 0) -> JSONObject: return self.view_get_workflow_status(profile=profile, viewgraph_key=viewgraph_key, row_record_id=row_record_id) def view_list(self, *, profile: str, app_key: str) -> JSONObject: """执行视图相关逻辑。""" self._require_app_key(app_key) def runner(session_profile, context): warnings: list[JSONObject] = [] verification: JSONObject = {"custom_views_loaded": True, "system_view_fallback_used": False} try: result = self.backend.request("GET", context, f"/app/{app_key}/view/viewList") return {"profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "result": result} except QingflowApiError as exc: if not _is_optional_view_list_error(exc): raise items, probe_warnings = self._resolve_accessible_system_views(context, app_key) warnings.extend(probe_warnings) warnings.append(_view_list_fallback_warning("VIEW_LIST_UNAVAILABLE", exc)) verification = {"custom_views_loaded": False, "system_view_fallback_used": True} return { "profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "result": items, "items": items, "warnings": warnings, "verification": verification, } return self._run(profile, runner, tool_name="视图列表") def view_list_flat(self, *, profile: str, app_key: str) -> JSONObject: """执行视图相关逻辑。""" self._require_app_key(app_key) return self._request(profile, "GET", f"/app/{app_key}/view/views", app_key=app_key, tool_name="视图列表（扁平）") def view_reorder(self, *, profile: str, app_key: str, payload: list[JSONObject]) -> JSONObject: """执行视图相关逻辑。""" self._require_app_key(app_key) if not isinstance(payload, list) or not payload: raise_tool_error(QingflowApiError.config_error("payload must be a non-empty array")) return self._request( profile, "POST", f"/app/{app_key}/view/ordinal", app_key=app_key, tool_name="视图排序", json_body=self._normalize_reorder_payload(payload), ) def view_set_column_width(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: """执行视图相关逻辑。""" self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/view/que/width", app_key=app_key, tool_name="设置视图列宽", json_body=body) def view_create(self, *, profile: str, payload: JSONObject) -> JSONObject: """执行视图相关逻辑。""" body = self._require_dict(payload) app_key = str(body.get("appKey") or "") return self._request(profile, "POST", "/view/viewConfig", app_key=app_key, tool_name="创建视图", json_body=body) def view_get_config(self, *, profile: str, viewgraph_key: str) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) return self._request(profile, "GET", f"/view/{viewgraph_key}/viewConfig", viewgraph_key=viewgraph_key, tool_name="视图配置") def view_get_base_info(self, *, profile: str, viewgraph_key: str, passcode: str | None) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) params = {"pass": passcode} if passcode else None def runner(session_profile, context): warnings: list[JSONObject] = [] verification: JSONObject = {"base_info_verified": True, "fallback_config_used": False} try: result = self.backend.request("GET", context, f"/view/{viewgraph_key}/viewConfig/baseInfo", params=params) except QingflowApiError as error: if not _is_optional_view_list_error(error): raise result = self.backend.request("GET", context, f"/view/{viewgraph_key}/viewConfig") verification["base_info_verified"] = False verification["fallback_config_used"] = True warning: JSONObject = { "code": "VIEW_BASE_INFO_UNAVAILABLE", "message": "view_get_base_info used viewConfig because view baseInfo is unavailable in this permission context.", } if error.backend_code is not None: warning["backend_code"] = error.backend_code if error.http_status is not None: warning["http_status"] = error.http_status if error.request_id is not None: warning["request_id"] = error.request_id if error.details: warning["details"] = error.details warnings.append(warning) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "viewgraph_key": viewgraph_key, "result": result, "warnings": warnings, "verification": verification, } return self._run(profile, runner, tool_name="视图基础信息") def view_update(self, *, profile: str, viewgraph_key: str, payload: JSONObject) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) body = self._require_dict(payload) return self._request( profile, "POST", f"/view/{viewgraph_key}/viewConfig", viewgraph_key=viewgraph_key, tool_name="更新视图配置", json_body=body, risk_operation="update", risk_target="view configuration", ) def view_delete(self, *, profile: str, viewgraph_key: str) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) return self._request( profile, "DELETE", f"/view/{viewgraph_key}", viewgraph_key=viewgraph_key, tool_name="删除视图", risk_operation="delete", risk_target="view configuration", ) def view_copy(self, *, profile: str, viewgraph_key: str) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) return self._request(profile, "POST", f"/view/{viewgraph_key}/copy", viewgraph_key=viewgraph_key, tool_name="复制视图") def view_list_questions(self, *, profile: str, viewgraph_key: str) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) return self._request(profile, "GET", f"/view/{viewgraph_key}/question", viewgraph_key=viewgraph_key, tool_name="视图字段列表") def view_list_associations(self, *, profile: str, viewgraph_key: str) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) return self._request(profile, "GET", f"/view/{viewgraph_key}/association", viewgraph_key=viewgraph_key, tool_name="视图关联列表") def view_update_member_config(self, *, profile: str, viewgraph_key: str, payload: JSONObject) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) body = self._require_dict(payload) return self._request( profile, "POST", f"/view/{viewgraph_key}/member/config", viewgraph_key=viewgraph_key, tool_name="更新视图成员配置", json_body=body, risk_operation="update", risk_target="view member visibility", ) def view_update_apply_config(self, *, profile: str, viewgraph_key: str, payload: JSONObject) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) body = self._require_dict(payload) return self._request( profile, "POST", f"/view/{viewgraph_key}/apply/config", viewgraph_key=viewgraph_key, tool_name="更新视图数据配置", json_body=body, risk_operation="update", risk_target="view apply visibility", ) def view_board_set_lane_config(self, *, profile: str, viewgraph_key: str, payload: JSONObject) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) body = self._require_dict(payload) return self._request( profile, "POST", f"/view/{viewgraph_key}/lane/config", viewgraph_key=viewgraph_key, tool_name="看板泳道配置", json_body=body, risk_operation="update", risk_target="board lane configuration", ) def view_board_get_lane_base_info(self, *, profile: str, viewgraph_key: str, page_num: int, page_size: int) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) return self._request( profile, "GET", f"/view/{viewgraph_key}/lane/baseInfo", viewgraph_key=viewgraph_key, tool_name="看板泳道基础信息", params={"pageNum": page_num, "pageSize": page_size}, ) def view_gantt_switch_auto_calibration(self, *, profile: str, viewgraph_key: str, payload: JSONObject) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) body = self._require_dict(payload) return self._request( profile, "PUT", f"/view/gantt/{viewgraph_key}/auto/calibration", viewgraph_key=viewgraph_key, tool_name="甘特图自动校准", json_body=body, risk_operation="update", risk_target="gantt auto calibration settings", ) def view_gantt_batch_update_time(self, *, profile: str, viewgraph_key: str, payload: JSONObject) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) body = self._require_dict(payload) return self._request( profile, "PUT", f"/view/gantt/{viewgraph_key}/batch/update", viewgraph_key=viewgraph_key, tool_name="甘特图批量更新时间", json_body=body, risk_operation="update", risk_target="gantt task timing", ) def view_get_future_nodes(self, *, profile: str, viewgraph_key: str, apply_id: int, app_key: str) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) self._require_app_key(app_key) self._require_positive("apply_id", apply_id) return self._request( profile, "GET", f"/view/{viewgraph_key}/auditNode/futureList/{apply_id}", viewgraph_key=viewgraph_key, app_key=app_key, apply_id=apply_id, tool_name="视图后续节点", params={"appKey": app_key}, ) def view_get_workflow_status(self, *, profile: str, viewgraph_key: str, row_record_id: int) -> JSONObject: """执行视图相关逻辑。""" self._require_viewgraph_key(viewgraph_key) self._require_positive("row_record_id", row_record_id) return self._request(profile, "GET", f"/view/{viewgraph_key}/workflow/status/{row_record_id}", viewgraph_key=viewgraph_key, row_record_id=row_record_id, tool_name="视图流程状态") def _request( self, profile: str, method: str, path: str, *, json_body: JSONValue = None, params: JSONObject | None = None, risk_operation: str | None = None, risk_target: str | None = None, tool_name: str, **extra: JSONValue, ) -> JSONObject: """执行内部辅助逻辑。""" def runner(session_profile, context): result = self.backend.request(method, context, path, json_body=json_body, params=params) response: JSONObject = {"profile": profile, "ws_id": session_profile.selected_ws_id, "result": result} response.update(extra) if risk_operation and risk_target: return self._attach_human_review_notice(response, operation=risk_operation, target=risk_target) return response return self._run(profile, runner, tool_name=tool_name) def _require_app_key(self, app_key: str) -> None: """执行内部辅助逻辑。""" if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) def _require_viewgraph_key(self, viewgraph_key: str) -> None: """执行内部辅助逻辑。""" if not viewgraph_key: raise_tool_error(QingflowApiError.config_error("viewgraph_key is required")) def _require_positive(self, field_name: str, value: int) -> None: """执行内部辅助逻辑。""" if value <= 0: raise_tool_error(QingflowApiError.config_error(f"{field_name} must be positive")) def _resolve_accessible_system_views(self, context, app_key: str) -> tuple[list[JSONObject], list[JSONObject]]: items: list[JSONObject] = [] warnings: list[JSONObject] = [] for view_id, list_type, name in SYSTEM_VIEW_DEFINITIONS: try: self.backend.request( "POST", context, f"/app/{app_key}/apply/filter", json_body={"type": list_type, "pageNum": 1, "pageSize": 1}, ) except QingflowApiError as exc: if _is_optional_view_list_error(exc): continue raise items.append( { "view_id": view_id, "viewId": view_id, "viewName": name, "name": name, "kind": "system", "list_type": list_type, "analysis_supported": True, } ) return items, warnings def _normalize_reorder_payload(self, payload: list[JSONObject]) -> list[JSONObject]: """执行内部辅助逻辑。""" first_item = payload[0] if "ordinalType" in first_item and "viewKeyList" in first_item: return payload if "ordinalType" in first_item and "viewList" in first_item: normalized_groups: list[JSONObject] = [] for group in payload: view_list = group.get("viewList") view_keys = [ str(view.get("viewKey")) for view in view_list if isinstance(view_list, list) and isinstance(view, dict) and view.get("viewKey") ] if group.get("ordinalType") and view_keys: normalized_groups.append({"ordinalType": group["ordinalType"], "viewKeyList": view_keys}) if normalized_groups: return normalized_groups sorted_items = sorted( payload, key=lambda item: int(item.get("ordinal", payload.index(item))) if isinstance(item.get("ordinal"), int) else payload.index(item), ) view_keys = [ str(item.get("viewgraphKey") or item.get("viewKey")) for item in sorted_items if item.get("viewgraphKey") or item.get("viewKey") ] if not view_keys: raise_tool_error(QingflowApiError.config_error("payload must include viewgraphKey/viewKey or ordinalType/viewKeyList")) return [{"ordinalType": "FIXED_VIEW_LIST", "viewKeyList": view_keys}] def _is_optional_view_list_error(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False backend_code = backend_code_int(error) return backend_code in {40002, 40027, 404} or error.http_status == 404 def _view_list_fallback_warning(code: str, error: QingflowApiError) -> JSONObject: return { "code": code, "message": "custom view list is unavailable; returned accessible system views instead", "backend_code": error.backend_code, "http_status": error.http_status, "request_id": error.request_id, }