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 PortalTools(ToolBase): """门户工具（中文名：门户页管理）。 类型：门户配置工具。 主要职责： 1. 查询门户列表与详情； 2. 维护门户基础信息与配置； 3. 执行门户发布与删除操作。 """ def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool() def portal_list(profile: str = DEFAULT_PROFILE) -> JSONObject: return self.portal_list(profile=profile) @mcp.tool() def portal_get(profile: str = DEFAULT_PROFILE, dash_key: str = "", being_draft: bool = False) -> JSONObject: return self.portal_get(profile=profile, dash_key=dash_key, being_draft=being_draft) @mcp.tool() def portal_get_base_info(profile: str = DEFAULT_PROFILE, dash_key: str = "") -> JSONObject: return self.portal_get_base_info(profile=profile, dash_key=dash_key) @mcp.tool() def portal_create(profile: str = DEFAULT_PROFILE, payload: JSONObject | None = None) -> JSONObject: return self.portal_create(profile=profile, payload=payload or {}) @mcp.tool(description=self._high_risk_tool_description(operation="update", target="portal base settings")) def portal_update_base_info(profile: str = DEFAULT_PROFILE, dash_key: str = "", payload: JSONObject | None = None) -> JSONObject: return self.portal_update_base_info(profile=profile, dash_key=dash_key, payload=payload or {}) @mcp.tool(description=self._high_risk_tool_description(operation="update", target="portal configuration")) def portal_update(profile: str = DEFAULT_PROFILE, dash_key: str = "", payload: JSONObject | None = None) -> JSONObject: return self.portal_update(profile=profile, dash_key=dash_key, payload=payload or {}) @mcp.tool(description=self._high_risk_tool_description(operation="delete", target="portal configuration")) def portal_delete(profile: str = DEFAULT_PROFILE, dash_key: str = "") -> JSONObject: return self.portal_delete(profile=profile, dash_key=dash_key) @mcp.tool() def portal_publish(profile: str = DEFAULT_PROFILE, dash_key: str = "") -> JSONObject: return self.portal_publish(profile=profile, dash_key=dash_key) def portal_list(self, *, profile: str) -> JSONObject: """执行门户相关逻辑。""" def runner(session_profile, context): result = self.backend.request("GET", context, "/dash") return {"profile": profile, "ws_id": session_profile.selected_ws_id, "items": result} return self._run(profile, runner, tool_name='门户列表') def portal_get(self, *, profile: str, dash_key: str, being_draft: bool = False) -> JSONObject: """执行门户相关逻辑。""" self._require_dash_key(dash_key) def runner(session_profile, context): result = self.backend.request("GET", context, f"/dash/{dash_key}", params={"beingDraft": being_draft}) return {"profile": profile, "ws_id": session_profile.selected_ws_id, "dash_key": dash_key, "result": result} return self._run(profile, runner, tool_name='门户详情') def portal_create(self, *, profile: str, payload: JSONObject) -> JSONObject: """执行门户相关逻辑。""" body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("POST", context, "/dash", json_body=body) return {"profile": profile, "ws_id": session_profile.selected_ws_id, "result": result} return self._run(profile, runner, tool_name='创建门户') def portal_get_base_info(self, *, profile: str, dash_key: str) -> JSONObject: """执行门户相关逻辑。""" self._require_dash_key(dash_key) def runner(session_profile, context): warnings: list[JSONObject] = [] verification: JSONObject = {"base_info_verified": True, "fallback_detail_used": False} try: result = self.backend.request("GET", context, f"/dash/{dash_key}/baseInfo") except QingflowApiError as error: if not _is_optional_portal_base_info_error(error): raise result = self.backend.request("GET", context, f"/dash/{dash_key}", params={"beingDraft": False}) verification["base_info_verified"] = False verification["fallback_detail_used"] = True warning: JSONObject = { "code": "PORTAL_BASE_INFO_UNAVAILABLE", "message": "portal_get_base_info used portal detail because portal 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, "dash_key": dash_key, "result": result, "warnings": warnings, "verification": verification, } return self._run(profile, runner, tool_name='门户基础信息') def portal_update_base_info(self, *, profile: str, dash_key: str, payload: JSONObject) -> JSONObject: """执行门户相关逻辑。""" self._require_dash_key(dash_key) body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("POST", context, f"/dash/{dash_key}/baseInfo", json_body=body) return self._attach_human_review_notice( {"profile": profile, "ws_id": session_profile.selected_ws_id, "dash_key": dash_key, "result": result}, operation="update", target="portal base settings", ) return self._run(profile, runner, tool_name='更新门户基础信息') def portal_update(self, *, profile: str, dash_key: str, payload: JSONObject) -> JSONObject: """执行门户相关逻辑。""" self._require_dash_key(dash_key) body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("POST", context, f"/dash/{dash_key}", json_body=body) return self._attach_human_review_notice( {"profile": profile, "ws_id": session_profile.selected_ws_id, "dash_key": dash_key, "result": result}, operation="update", target="portal configuration", ) return self._run(profile, runner, tool_name='更新门户配置') def portal_delete(self, *, profile: str, dash_key: str) -> JSONObject: """执行门户相关逻辑。""" self._require_dash_key(dash_key) def runner(session_profile, context): result = self.backend.request("DELETE", context, f"/dash/{dash_key}") return self._attach_human_review_notice( {"profile": profile, "ws_id": session_profile.selected_ws_id, "dash_key": dash_key, "result": result}, operation="delete", target="portal configuration", ) return self._run(profile, runner, tool_name='删除门户') def portal_publish(self, *, profile: str, dash_key: str) -> JSONObject: """执行门户相关逻辑。""" self._require_dash_key(dash_key) def runner(session_profile, context): result = self.backend.request("POST", context, f"/dash/{dash_key}/publish") return {"profile": profile, "ws_id": session_profile.selected_ws_id, "dash_key": dash_key, "result": result} return self._run(profile, runner, tool_name='发布门户') def _require_dash_key(self, dash_key: str) -> None: """执行内部辅助逻辑。""" if not dash_key: raise_tool_error(QingflowApiError.config_error("dash_key is required")) def _is_optional_portal_base_info_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