from __future__ import annotations from copy import deepcopy 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 from ..solution.compiler.icon_utils import workspace_icon_config from .base import ToolBase class PackageTools(ToolBase): """分组工具（中文名：分组与归档管理）。 类型：资源组织工具。 主要职责： 1. 查询、解析、创建分组； 2. 管理应用与分组的挂载关系； 3. 为应用搭建提供分组定位能力。 """ def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool() def package_list(profile: str = DEFAULT_PROFILE, trial_status: str = "all", include_raw: bool = False) -> JSONObject: return self.package_list(profile=profile, trial_status=trial_status, include_raw=include_raw) @mcp.tool() def package_get(profile: str = DEFAULT_PROFILE, tag_id: int = 0, include_raw: bool = False) -> JSONObject: return self.package_get(profile=profile, tag_id=tag_id, include_raw=include_raw) @mcp.tool() def package_get_base(profile: str = DEFAULT_PROFILE, tag_id: int = 0, include_raw: bool = False) -> JSONObject: return self.package_get_base(profile=profile, tag_id=tag_id, include_raw=include_raw) @mcp.tool() def package_create(profile: str = DEFAULT_PROFILE, payload: JSONObject | None = None) -> JSONObject: return self.package_create(profile=profile, payload=payload or {}) @mcp.tool(description=self._high_risk_tool_description(operation="update", target="app package settings")) def package_update(profile: str = DEFAULT_PROFILE, tag_id: int = 0, payload: JSONObject | None = None) -> JSONObject: return self.package_update(profile=profile, tag_id=tag_id, payload=payload or {}) @mcp.tool(description=self._high_risk_tool_description(operation="delete", target="app package and linked data")) def package_delete(profile: str = DEFAULT_PROFILE, tag_id: int = 0, deleted_all_data: bool = False) -> JSONObject: return self.package_delete(profile=profile, tag_id=tag_id, deleted_all_data=deleted_all_data) def package_list(self, *, profile: str, trial_status: str = "all", include_raw: bool = False) -> JSONObject: """执行分组与包相关逻辑。""" def runner(session_profile, context): result = self.backend.request("GET", context, "/tag", params={"trialStatus": trial_status}) raw_items, source_shape = self._extract_package_items(result) compact_items = [self._compact_package(item) for item in raw_items if isinstance(item, dict)] response = { "profile": profile, "ws_id": session_profile.selected_ws_id, "trial_status": trial_status, "items": result if include_raw else compact_items, "count": len(compact_items), "compact": not include_raw, "source_shape": source_shape, "retried": False, } if include_raw: response["summary"] = compact_items return response return self._run(profile, runner, tool_name='分组列表') def package_get(self, *, profile: str, tag_id: int, include_raw: bool = False) -> JSONObject: """执行分组与包相关逻辑。""" self._require_tag_id(tag_id) def runner(session_profile, context): warnings: list[JSONObject] = [] verification: JSONObject = {"base_info_verified": True, "fallback_visible_package_used": False} try: base_info = self.backend.request("GET", context, f"/tag/{tag_id}/baseInfo") except QingflowApiError as exc: if not _is_optional_package_metadata_error(exc): raise base_info = self._resolve_visible_package(context, tag_id) if base_info is None: raise warnings.append( { "code": "PACKAGE_BASE_INFO_UNAVAILABLE", "message": "package_get used the current user's visible package list because package baseInfo is unavailable in this permission context.", "backend_code": exc.backend_code, "http_status": exc.http_status, "request_id": exc.request_id, } ) verification = {"base_info_verified": False, "fallback_visible_package_used": True} result: JSONObject = base_info if isinstance(base_info, dict) else {} if verification["base_info_verified"]: try: detail = self.backend.request("GET", context, f"/tag/{tag_id}") if isinstance(detail, dict): result = detail except QingflowApiError as exc: if not _is_optional_package_metadata_error(exc): raise warnings.append( { "code": "PACKAGE_DETAIL_READ_DEGRADED", "message": "package_get used package baseInfo because the detail endpoint requires package edit/add-app permission.", "backend_code": exc.backend_code, "http_status": exc.http_status, "request_id": exc.request_id, } ) compact = self._compact_package( result if isinstance(result, dict) else {}, base_info=base_info if isinstance(base_info, dict) else None, ) response = { "profile": profile, "ws_id": session_profile.selected_ws_id, "tag_id": tag_id, "result": result if include_raw else compact, "compact": not include_raw, "detail_degraded": bool(warnings), "verification": verification, } if warnings: response["warnings"] = warnings if include_raw: response["summary"] = compact return response return self._run(profile, runner, tool_name='分组详情') def package_get_base(self, *, profile: str, tag_id: int, include_raw: bool = False) -> JSONObject: """执行分组与包相关逻辑。""" self._require_readable_tag_id(tag_id) def runner(session_profile, context): warnings: list[JSONObject] = [] verification: JSONObject = {"base_info_verified": True, "fallback_visible_package_used": False} try: result = self.backend.request("GET", context, f"/tag/{tag_id}/baseInfo") except QingflowApiError as exc: if not _is_optional_package_metadata_error(exc): raise visible_package = self._resolve_visible_package(context, tag_id) if visible_package is None: raise result = visible_package warnings.append( { "code": "PACKAGE_BASE_INFO_UNAVAILABLE", "message": "package_get_base used the current user's visible package list because package baseInfo is unavailable in this permission context.", "backend_code": exc.backend_code, "http_status": exc.http_status, "request_id": exc.request_id, } ) verification = {"base_info_verified": False, "fallback_visible_package_used": True} compact = self._compact_package(result if isinstance(result, dict) else {}) response = { "profile": profile, "ws_id": session_profile.selected_ws_id, "tag_id": tag_id, "result": result if include_raw else compact, "compact": not include_raw, "verification": verification, } if warnings: response["warnings"] = warnings if include_raw: response["summary"] = compact return response return self._run(profile, runner, tool_name='分组基础信息') def package_create(self, *, profile: str, payload: JSONObject) -> JSONObject: """执行分组与包相关逻辑。""" body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("POST", context, "/tag", json_body=self._normalize_package_payload(body)) return {"profile": profile, "ws_id": session_profile.selected_ws_id, "result": result} return self._run(profile, runner, tool_name='创建分组') def package_update(self, *, profile: str, tag_id: int, payload: JSONObject) -> JSONObject: """执行分组与包相关逻辑。""" self._require_tag_id(tag_id) body = self._require_dict(payload) def runner(session_profile, context): warnings: list[JSONObject] = [] try: current = self.backend.request("GET", context, f"/tag/{tag_id}") except QingflowApiError as exc: if not _is_optional_package_metadata_error(exc): raise try: current = self.backend.request("GET", context, f"/tag/{tag_id}/baseInfo") fallback_source = "baseInfo" except QingflowApiError as base_exc: if not _is_optional_package_metadata_error(base_exc): raise visible_package = self._resolve_visible_package(context, tag_id) if visible_package is None: raise base_exc current = visible_package fallback_source = "visible_package_list" warnings.append( { "code": "PACKAGE_DETAIL_READ_DEGRADED", "message": "package_update used package baseInfo or visible package metadata because package detail is unavailable in this permission context.", "source": fallback_source, "backend_code": exc.backend_code, "http_status": exc.http_status, "request_id": exc.request_id, } ) result = self.backend.request( "PUT", context, f"/tag/{tag_id}", json_body=self._normalize_package_payload(body, existing=current), ) response = {"profile": profile, "ws_id": session_profile.selected_ws_id, "tag_id": tag_id, "result": result} if warnings: response["warnings"] = warnings return self._attach_human_review_notice( response, operation="update", target="app package settings", ) return self._run(profile, runner, tool_name='更新分组') def package_sort_items(self, *, profile: str, tag_id: int, tag_items: list[JSONObject]) -> JSONObject: """执行分组与包相关逻辑。""" self._require_tag_id(tag_id) def runner(session_profile, context): result = self.backend.request( "POST", context, f"/tag/{tag_id}/ordinal", json_body={"tagItems": deepcopy(tag_items)}, ) return {"profile": profile, "ws_id": session_profile.selected_ws_id, "tag_id": tag_id, "result": result} return self._run(profile, runner, tool_name='分组排序') def package_group_create(self, *, profile: str, tag_id: int, group_name: str) -> JSONObject: """执行分组与包相关逻辑。""" self._require_tag_id(tag_id) normalized_name = str(group_name or "").strip() if not normalized_name: raise_tool_error(QingflowApiError.config_error("group_name is required")) def runner(session_profile, context): result = self.backend.request( "POST", context, f"/tag/{tag_id}/group", json_body={"groupName": normalized_name}, ) return {"profile": profile, "ws_id": session_profile.selected_ws_id, "tag_id": tag_id, "result": result} return self._run(profile, runner, tool_name='创建分组分栏') def package_group_update(self, *, profile: str, tag_id: int, group_id: int, group_name: str) -> JSONObject: """执行分组与包相关逻辑。""" self._require_tag_id(tag_id) self._require_group_id(group_id) normalized_name = str(group_name or "").strip() if not normalized_name: raise_tool_error(QingflowApiError.config_error("group_name is required")) def runner(session_profile, context): result = self.backend.request( "POST", context, f"/tag/{tag_id}/group/{group_id}", json_body={"groupName": normalized_name}, ) return {"profile": profile, "ws_id": session_profile.selected_ws_id, "tag_id": tag_id, "group_id": group_id, "result": result} return self._run(profile, runner, tool_name='更新分组分栏') def package_group_delete(self, *, profile: str, tag_id: int, group_id: int) -> JSONObject: """执行分组与包相关逻辑。""" self._require_tag_id(tag_id) self._require_group_id(group_id) def runner(session_profile, context): result = self.backend.request("DELETE", context, f"/tag/{tag_id}/group/{group_id}") return {"profile": profile, "ws_id": session_profile.selected_ws_id, "tag_id": tag_id, "group_id": group_id, "result": result} return self._run(profile, runner, tool_name='删除分组分栏') def package_delete(self, *, profile: str, tag_id: int, deleted_all_data: bool = False) -> JSONObject: """执行分组与包相关逻辑。""" self._require_tag_id(tag_id) def runner(session_profile, context): result = self.backend.request( "DELETE", context, "/tag", json_body={"tagId": tag_id, "deletedAllData": deleted_all_data}, ) return self._attach_human_review_notice( {"profile": profile, "ws_id": session_profile.selected_ws_id, "tag_id": tag_id, "deleted_all_data": deleted_all_data, "result": result}, operation="delete", target="app package and linked data", ) return self._run(profile, runner, tool_name='删除分组') def _require_tag_id(self, tag_id: int) -> None: """执行内部辅助逻辑。""" if tag_id <= 0: raise_tool_error(QingflowApiError.config_error("tag_id must be positive")) def _require_group_id(self, group_id: int) -> None: """执行内部辅助逻辑。""" if group_id <= 0: raise_tool_error(QingflowApiError.config_error("group_id must be positive")) def _require_readable_tag_id(self, tag_id: int) -> None: """执行内部辅助逻辑。""" if tag_id < 0: raise_tool_error(QingflowApiError.config_error("tag_id must be non-negative")) def _resolve_visible_package(self, context, tag_id: int) -> JSONObject | None: # type: ignore[no-untyped-def] """Resolve package metadata from the current user's visible package list.""" payload = self.backend.request("GET", context, "/tag", params={"trialStatus": "all"}) items, _source_shape = self._extract_package_items(payload) for item in items: try: candidate = int(item.get("tagId")) if item.get("tagId") is not None else None except (TypeError, ValueError): candidate = None if candidate == tag_id: return dict(item) return None def _normalize_package_payload(self, payload: JSONObject, existing: JSONObject | None = None) -> JSONObject: """执行内部辅助逻辑。""" data = deepcopy(existing) if isinstance(existing, dict) else {} data.update(deepcopy(payload)) data.pop("tagId", None) data.pop("createTime", None) data.pop("creator", None) data.pop("publishStatus", None) data.pop("beingTrial", None) data.pop("trialExpireDate", None) data.setdefault("tagIcon", "") data.setdefault("tagItems", []) data.setdefault("auth", _default_package_auth()) return data def _compact_package(self, result: dict[str, object], *, base_info: dict[str, object] | None = None) -> JSONObject: """执行内部辅助逻辑。""" tag_items = result.get("tagItems") if isinstance(result.get("tagItems"), list) else [] permission_source = base_info if isinstance(base_info, dict) else result preview = [] for item in tag_items[:10]: if not isinstance(item, dict): continue preview.append( { "title": item.get("title"), "itemType": item.get("itemType"), "appKey": item.get("appKey"), "dashKey": item.get("dashKey"), } ) item_count = None raw_item_count = result.get("itemCount") try: if raw_item_count is not None: coerced_count = int(raw_item_count) if coerced_count >= 0: item_count = coerced_count except (TypeError, ValueError): item_count = None if item_count is None and "tagItems" in result: item_count = len(tag_items) tag_icon = result.get("tagIcon") if result.get("tagIcon") is not None else permission_source.get("tagIcon") compact = { "tagId": result.get("tagId"), "tagName": result.get("tagName"), "tagIcon": tag_icon, "iconConfig": workspace_icon_config(str(tag_icon).strip() if tag_icon not in (None, "") else None), "publishStatus": result.get("publishStatus") if result.get("publishStatus") is not None else permission_source.get("publishStatus"), "beingTrial": result.get("beingTrial"), "itemCount": item_count, "itemPreview": preview, "addAppStatus": permission_source.get("addAppStatus"), "editAppStatus": permission_source.get("editAppStatus"), "delAppStatus": permission_source.get("delAppStatus"), "editTagStatus": permission_source.get("editTagStatus"), } return {key: value for key, value in compact.items() if value is not None} def _extract_package_items(self, result: object) -> tuple[list[dict[str, object]], str]: """执行内部辅助逻辑。""" if isinstance(result, list): return [item for item in result if isinstance(item, dict)], "list" if isinstance(result, dict): tags = result.get("tags") if isinstance(tags, list): return [item for item in tags if isinstance(item, dict)], "dict.tags" items = result.get("items") if isinstance(items, list): return [item for item in items if isinstance(item, dict)], "dict.items" return [], "dict.unknown" return [], type(result).__name__ def _default_package_auth() -> JSONObject: members = { "depart": [], "dynamic": [], "includeSubDeparts": None, "member": [], "role": [], } return { "type": "WORKSPACE", "contactAuth": {"type": "WORKSPACE_ALL", "authMembers": deepcopy(members)}, "externalMemberAuth": {"type": "NOT", "authMembers": deepcopy(members)}, } def _is_optional_package_metadata_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