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 from .base import ToolBase class NavigationTools(ToolBase): """导航工具（中文名：导航菜单管理）。 类型：门户导航配置工具。 主要职责： 1. 查询导航结构与发布状态； 2. 读取指定导航详情； 3. 应用导航配置变更。 """ def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool() def navigation_list_published(profile: str = DEFAULT_PROFILE, page_num: int = 1, page_size: int = 50) -> JSONObject: return self.navigation_list_published(profile=profile, page_num=page_num, page_size=page_size) @mcp.tool() def navigation_list_draft_page( profile: str = DEFAULT_PROFILE, page_num: int = 1, page_size: int = 50, query_key: str | None = None, ) -> JSONObject: return self.navigation_list_draft_page(profile=profile, page_num=page_num, page_size=page_size, query_key=query_key) @mcp.tool() def navigation_list_draft_all(profile: str = DEFAULT_PROFILE, query_key: str | None = None) -> JSONObject: return self.navigation_list_draft_all(profile=profile, query_key=query_key) @mcp.tool() def navigation_get_detail(profile: str = DEFAULT_PROFILE, navigation_item_id: int = 0, status: str = "draft") -> JSONObject: return self.navigation_get_detail(profile=profile, navigation_item_id=navigation_item_id, status=status) @mcp.tool() def navigation_get_status(profile: str = DEFAULT_PROFILE) -> JSONObject: return self.navigation_get_status(profile=profile) @mcp.tool() def navigation_set_visible(profile: str = DEFAULT_PROFILE, payload: JSONObject | None = None) -> JSONObject: return self.navigation_set_visible(profile=profile, payload=payload or {}) @mcp.tool() def navigation_create(profile: str = DEFAULT_PROFILE, payload: JSONObject | None = None) -> JSONObject: return self.navigation_create(profile=profile, payload=payload or {}) @mcp.tool(description=self._high_risk_tool_description(operation="update", target="navigation item configuration")) def navigation_update(profile: str = DEFAULT_PROFILE, navigation_item_id: int = 0, payload: JSONObject | None = None) -> JSONObject: return self.navigation_update(profile=profile, navigation_item_id=navigation_item_id, payload=payload or {}) @mcp.tool(description=self._high_risk_tool_description(operation="delete", target="navigation item configuration")) def navigation_delete(profile: str = DEFAULT_PROFILE, navigation_item_id: int = 0) -> JSONObject: return self.navigation_delete(profile=profile, navigation_item_id=navigation_item_id) @mcp.tool() def navigation_publish(profile: str = DEFAULT_PROFILE, navigation_id: int = 0) -> JSONObject: return self.navigation_publish(profile=profile, navigation_id=navigation_id) @mcp.tool() def navigation_reorder(profile: str = DEFAULT_PROFILE, payload: list[JSONObject] | None = None) -> JSONObject: return self.navigation_reorder(profile=profile, payload=payload or []) def navigation_list_published(self, *, profile: str, page_num: int = 1, page_size: int = 50) -> JSONObject: """执行工具方法逻辑。""" def runner(session_profile, context): result = self.backend.request("GET", context, "/navigation", params={"pageNum": page_num, "pageSize": page_size}) return {"profile": profile, "ws_id": session_profile.selected_ws_id, "page": result} return self._run(profile, runner, tool_name='导航已发布列表') def navigation_list_draft_page(self, *, profile: str, page_num: int = 1, page_size: int = 50, query_key: str | None = None) -> JSONObject: """执行工具方法逻辑。""" def runner(session_profile, context): params: JSONObject = {"pageNum": page_num, "pageSize": page_size} if query_key: params["queryCondition"] = query_key fallback_error: QingflowApiError | None = None try: result = self.backend.request("GET", context, "/navigation/page", params=params) response = {"profile": profile, "ws_id": session_profile.selected_ws_id, "status": "draft", "page": result} except QingflowApiError as exc: if not _is_optional_draft_navigation_read_error(exc): raise fallback_error = exc result = self.backend.request("GET", context, "/navigation", params={"pageNum": page_num, "pageSize": page_size}) response = { "profile": profile, "ws_id": session_profile.selected_ws_id, "status": "published", "requested_status": "draft", "page": result, } if fallback_error is not None: response["warnings"] = [_navigation_fallback_warning("NAVIGATION_DRAFT_PAGE_UNAVAILABLE", fallback_error)] response["verification"] = {"draft_readable": False, "published_fallback_used": True} return response return self._run(profile, runner, tool_name='导航草稿分页') def navigation_list_draft_all(self, *, profile: str, query_key: str | None = None) -> JSONObject: """执行工具方法逻辑。""" def runner(session_profile, context): params: JSONObject = {} if query_key: params["queryCondition"] = query_key fallback_error: QingflowApiError | None = None try: result = self.backend.request("GET", context, "/navigation/all", params=params) response = {"profile": profile, "ws_id": session_profile.selected_ws_id, "status": "draft", "items": result} except QingflowApiError as exc: if not _is_optional_draft_navigation_read_error(exc): raise fallback_error = exc result = self.backend.request("GET", context, "/navigation", params={"pageNum": 1, "pageSize": 1000}) response = { "profile": profile, "ws_id": session_profile.selected_ws_id, "status": "published", "requested_status": "draft", "items": result, } if fallback_error is not None: response["warnings"] = [_navigation_fallback_warning("NAVIGATION_DRAFT_ALL_UNAVAILABLE", fallback_error)] response["verification"] = {"draft_readable": False, "published_fallback_used": True} return response return self._run(profile, runner, tool_name='导航草稿全量') def navigation_get_detail(self, *, profile: str, navigation_item_id: int, status: str = "draft") -> JSONObject: """执行工具方法逻辑。""" self._require_navigation_item_id(navigation_item_id) def runner(session_profile, context): requested_status = str(status or "draft") fallback_error: QingflowApiError | None = None try: result = self.backend.request( "GET", context, f"/navigation/detail/{navigation_item_id}", params={"status": requested_status}, ) effective_status = requested_status except QingflowApiError as exc: if requested_status != "draft" or not _is_optional_draft_navigation_read_error(exc): raise fallback_error = exc effective_status = "published" result = self.backend.request( "GET", context, f"/navigation/detail/{navigation_item_id}", params={"status": effective_status}, ) response = { "profile": profile, "ws_id": session_profile.selected_ws_id, "navigation_item_id": navigation_item_id, "status": effective_status, "requested_status": requested_status, "result": result, } if fallback_error is not None: response["warnings"] = [_navigation_fallback_warning("NAVIGATION_DRAFT_DETAIL_UNAVAILABLE", fallback_error)] response["verification"] = {"draft_readable": False, "published_fallback_used": True} return response return self._run(profile, runner, tool_name='导航详情') def navigation_get_status(self, *, profile: str) -> JSONObject: """执行工具方法逻辑。""" def runner(session_profile, context): result = self.backend.request("GET", context, "/navigation/status") return {"profile": profile, "ws_id": session_profile.selected_ws_id, "result": result} return self._run(profile, runner, tool_name='导航发布状态') def navigation_set_visible(self, *, profile: str, payload: JSONObject) -> JSONObject: """执行工具方法逻辑。""" body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("POST", context, "/navigation/visible", json_body=body) return {"profile": profile, "ws_id": session_profile.selected_ws_id, "result": result} return self._run(profile, runner, tool_name='设置导航可见性') def navigation_create(self, *, profile: str, payload: JSONObject) -> JSONObject: """执行工具方法逻辑。""" body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("POST", context, "/navigation", json_body=body) return {"profile": profile, "ws_id": session_profile.selected_ws_id, "result": result} return self._run(profile, runner, tool_name='创建导航项') def navigation_update(self, *, profile: str, navigation_item_id: int, payload: JSONObject) -> JSONObject: """执行工具方法逻辑。""" self._require_navigation_item_id(navigation_item_id) body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("PUT", context, f"/navigation/{navigation_item_id}", json_body=body) return self._attach_human_review_notice( {"profile": profile, "ws_id": session_profile.selected_ws_id, "navigation_item_id": navigation_item_id, "result": result}, operation="update", target="navigation item configuration", ) return self._run(profile, runner, tool_name='更新导航项') def navigation_delete(self, *, profile: str, navigation_item_id: int) -> JSONObject: """执行工具方法逻辑。""" self._require_navigation_item_id(navigation_item_id) def runner(session_profile, context): result = self.backend.request("DELETE", context, f"/navigation/{navigation_item_id}") return self._attach_human_review_notice( {"profile": profile, "ws_id": session_profile.selected_ws_id, "navigation_item_id": navigation_item_id, "result": result}, operation="delete", target="navigation item configuration", ) return self._run(profile, runner, tool_name='删除导航项') def navigation_publish(self, *, profile: str, navigation_id: int) -> JSONObject: """执行工具方法逻辑。""" if navigation_id <= 0: raise_tool_error(QingflowApiError.config_error("navigation_id must be positive")) def runner(session_profile, context): result = self.backend.request("POST", context, f"/navigation/publish/{navigation_id}") return {"profile": profile, "ws_id": session_profile.selected_ws_id, "navigation_id": navigation_id, "result": result} return self._run(profile, runner, tool_name='发布导航') def navigation_reorder(self, *, profile: str, payload: list[JSONObject]) -> JSONObject: """执行工具方法逻辑。""" if not isinstance(payload, list) or not payload: raise_tool_error(QingflowApiError.config_error("payload must be a non-empty array")) def runner(session_profile, context): result = self.backend.request("POST", context, "/navigation/ordinal", json_body=payload) return {"profile": profile, "ws_id": session_profile.selected_ws_id, "result": result} return self._run(profile, runner, tool_name='导航排序') def _require_navigation_item_id(self, navigation_item_id: int) -> None: """执行内部辅助逻辑。""" if navigation_item_id <= 0: raise_tool_error(QingflowApiError.config_error("navigation_item_id must be positive")) def _is_optional_draft_navigation_read_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 _navigation_fallback_warning(code: str, error: QingflowApiError) -> JSONObject: return { "code": code, "message": "draft navigation data is unavailable; returned published navigation data instead", "backend_code": error.backend_code, "http_status": error.http_status, "request_id": error.request_id, }