from __future__ import annotations from typing import Any from mcp.server.fastmcp import FastMCP from ..backend_client import BackendRequestContext from ..cloud_context import current_cloud_context from ..config import DEFAULT_PROFILE from ..errors import QingflowApiError, backend_code_int, is_auth_like_error, raise_tool_error from .base import ToolBase _WORKSPACE_DETAIL_WARNING_KEY = "_qingflow_workspace_detail_warning" class WorkspaceTools(ToolBase): """工作区工具（中文名：工作区与插件管理）。 类型：基础上下文工具。 主要职责： 1. 查询当前用户可见工作区列表； 2. 管理工作区插件安装状态。 """ def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool() def workspace_list( profile: str = DEFAULT_PROFILE, page_num: int = 1, page_size: int = 20, include_external: bool = False, ) -> dict[str, Any]: return self.workspace_list( profile=profile, page_num=page_num, page_size=page_size, include_external=include_external, ) @mcp.tool() def workspace_get( profile: str = DEFAULT_PROFILE, ws_id: int = 0, ) -> dict[str, Any]: return self.workspace_get( profile=profile, ws_id=ws_id if ws_id > 0 else None, ) @mcp.tool() def workspace_select( profile: str = DEFAULT_PROFILE, ws_id: int = 0, ) -> dict[str, Any]: return self.workspace_select( profile=profile, ws_id=ws_id, ) @mcp.tool() def workspace_set_plugin_status( profile: str = DEFAULT_PROFILE, plugin_id: int = 0, being_installed: bool = True, ) -> dict[str, Any]: return self.workspace_set_plugin_status( profile=profile, plugin_id=plugin_id, being_installed=being_installed, ) def workspace_list( self, *, profile: str = DEFAULT_PROFILE, page_num: int = 1, page_size: int = 20, include_external: bool = False, ) -> dict[str, Any]: """执行工作区相关逻辑。""" if page_num <= 0 or page_size <= 0: raise_tool_error(QingflowApiError.config_error("page_num and page_size must be positive")) def runner(_, context): path = "/user/allWorkspaceList/pageQuery" if include_external else "/user/workspaceList/pageQuery" result = self.backend.request( "POST", context, path, json_body={"pageNum": page_num, "pageSize": page_size, "authList": [0, 1, 2]}, ) normalized = result if isinstance(result, dict) else {"list": result if isinstance(result, list) else []} workspace_list = normalized.get("list") if isinstance(normalized.get("list"), list) else [] selected_ws_id = _.selected_ws_id if _ is not None else None warnings: list[dict[str, Any]] = [] if selected_ws_id and not any(isinstance(item, dict) and item.get("wsId") == selected_ws_id for item in workspace_list): try: selected_workspace = self.backend.request("GET", context, f"/user/workspace/{selected_ws_id}") except QingflowApiError as exc: if not _is_optional_workspace_detail_error(exc): raise warnings.append(_workspace_detail_fallback_warning(exc, ws_id=selected_ws_id)) selected_workspace = { "wsId": selected_ws_id, "workspaceName": _.selected_ws_name if _ is not None else None, } if isinstance(selected_workspace, dict): workspace_list.append(selected_workspace) normalized["list"] = workspace_list total = normalized.get("total") if isinstance(total, int): normalized["total"] = total + 1 return { "profile": profile, "include_external": include_external, "page": normalized, "warnings": warnings, "verification": { "selected_workspace_detail_verified": not warnings, "selected_workspace_included": bool( selected_ws_id and any(isinstance(item, dict) and item.get("wsId") == selected_ws_id for item in workspace_list) ), }, } return self._run(profile, runner, require_workspace=False, tool_name='工作区列表') def workspace_get( self, *, profile: str = DEFAULT_PROFILE, ws_id: int | None = None, ) -> dict[str, Any]: """读取单个工作区详情，并尽量补齐真实 systemVersion。""" def runner(session_profile, context): target_ws_id = ws_id or (session_profile.selected_ws_id if session_profile is not None else None) if target_ws_id is None or target_ws_id <= 0: raise_tool_error(QingflowApiError.workspace_not_selected(profile)) workspace = self._fetch_workspace_with_fallback(context, ws_id=target_ws_id) workspace, warnings = _strip_workspace_detail_warning(workspace) system_version = self._workspace_system_version(workspace) return { "profile": profile, "ws_id": target_ws_id, "qf_version": system_version, "qf_version_source": "workspace_system_version" if system_version else "unverified", "workspace": workspace, "warnings": warnings, "verification": { "workspace_detail_verified": not warnings, "workspace_list_fallback_used": bool(warnings), }, } return self._run(profile, runner, require_workspace=False, tool_name='工作区详情') def workspace_select( self, *, profile: str = DEFAULT_PROFILE, ws_id: int, ) -> dict[str, Any]: """切换当前 profile 选中的工作区，并尽量同步真实 systemVersion。""" if ws_id <= 0: raise_tool_error(QingflowApiError.config_error("ws_id must be positive")) def runner(_, context): cloud_context = current_cloud_context.get() if cloud_context is not None and ws_id != cloud_context.ws_id: raise_tool_error( QingflowApiError.config_error( "Cloud MCP workspace is bound to the configured MCP credential. Use a credential for the target workspace." ) ) workspace = self._fetch_workspace_with_fallback(context, ws_id=ws_id) workspace, warnings = _strip_workspace_detail_warning(workspace) workspace_name = str(workspace.get("workspaceName") or workspace.get("wsName") or workspace.get("remark") or "").strip() or None if cloud_context is not None: system_version = self._workspace_system_version(workspace) qf_version_source = "workspace_system_version" if system_version else "unverified" return { "profile": profile, "ws_id": ws_id, "qf_version": system_version or cloud_context.qf_version, "qf_version_source": qf_version_source if system_version else cloud_context.qf_version_source, "workspace": workspace, "selected": { "ws_id": cloud_context.ws_id, "workspace_name": workspace_name or cloud_context.ws_name, }, "warnings": warnings, "verification": { "workspace_detail_verified": not warnings, "workspace_list_fallback_used": bool(warnings), }, } selected = self.sessions.select_workspace(profile, ws_id=ws_id, ws_name=workspace_name) system_version = self._workspace_system_version(workspace) qf_version_source = "workspace_system_version" if system_version else "unverified" if system_version: selected = self.sessions.update_route( profile, qf_version=system_version, qf_version_source=qf_version_source, ) return { "profile": profile, "ws_id": ws_id, "qf_version": selected.qf_version, "qf_version_source": selected.qf_version_source or qf_version_source, "workspace": workspace, "selected": { "ws_id": selected.selected_ws_id, "workspace_name": selected.selected_ws_name, }, "warnings": warnings, "verification": { "workspace_detail_verified": not warnings, "workspace_list_fallback_used": bool(warnings), }, } return self._run(profile, runner, require_workspace=False, tool_name='切换工作区') def workspace_set_plugin_status( self, *, profile: str = DEFAULT_PROFILE, plugin_id: int, being_installed: bool = True, ) -> dict[str, Any]: """执行工作区相关逻辑。""" if plugin_id <= 0: raise_tool_error(QingflowApiError.config_error("plugin_id must be positive")) def runner(_, context): result = self.backend.request( "POST", context, "/ws/plugin", json_body={ "pluginId": plugin_id, "beingInstalled": being_installed, }, ) return { "profile": profile, "plugin_id": plugin_id, "being_installed": being_installed, "result": result, } return self._run(profile, runner, tool_name='设置工作区插件状态') def _fetch_workspace_with_fallback( self, context: BackendRequestContext, *, ws_id: int, ) -> dict[str, Any]: try: workspace = self.backend.request("GET", context, f"/user/workspace/{ws_id}") except QingflowApiError as error: if not _is_optional_workspace_detail_error(error): raise fallback = self._fetch_workspace_from_list(context, ws_id=ws_id) if isinstance(fallback, dict): return _with_workspace_detail_warning(fallback, error, ws_id=ws_id) raise if not isinstance(workspace, dict): raise_tool_error(QingflowApiError(category="workspace", message=f"Workspace {ws_id} is not accessible")) if self._workspace_needs_list_fallback(workspace): fallback = self._fetch_workspace_from_list(context, ws_id=ws_id) if isinstance(fallback, dict): merged = dict(workspace) for key, value in fallback.items(): if merged.get(key) in (None, "") and value not in (None, ""): merged[key] = value workspace = merged return workspace def _fetch_workspace_from_list(self, context: BackendRequestContext, *, ws_id: int) -> dict[str, Any] | None: payload = self.backend.request( "POST", BackendRequestContext( base_url=context.base_url, token=context.token, ws_id=None, qf_version=context.qf_version, qf_version_source=context.qf_version_source, ), "/user/workspaceList/pageQuery", json_body={"pageNum": 1, "pageSize": 100, "authList": [0, 1, 2, 3]}, ) workspaces = payload.get("list") if isinstance(payload, dict) else [] if not isinstance(workspaces, list): return None found = next( ( item for item in workspaces if isinstance(item, dict) and item.get("wsId") == ws_id ), None, ) return found if isinstance(found, dict) else None def _workspace_needs_list_fallback(self, workspace: dict[str, Any]) -> bool: workspace_name = str(workspace.get("workspaceName") or workspace.get("wsName") or "").strip() system_version = self._workspace_system_version(workspace) return not workspace_name or system_version is None def _workspace_system_version(self, workspace: Any) -> str | None: if not isinstance(workspace, dict): return None value = workspace.get("systemVersion") if value is None: return None normalized = str(value).strip() return normalized or None def _is_optional_workspace_detail_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, 59004} or error.http_status == 404 def _with_workspace_detail_warning(workspace: dict[str, Any], error: QingflowApiError, *, ws_id: int) -> dict[str, Any]: enriched = dict(workspace) enriched[_WORKSPACE_DETAIL_WARNING_KEY] = _workspace_detail_fallback_warning(error, ws_id=ws_id) return enriched def _strip_workspace_detail_warning(workspace: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]: public_workspace = dict(workspace) warning = public_workspace.pop(_WORKSPACE_DETAIL_WARNING_KEY, None) warnings = [warning] if isinstance(warning, dict) else [] return public_workspace, warnings def _workspace_detail_fallback_warning(error: QingflowApiError, *, ws_id: int) -> dict[str, Any]: return { "code": "WORKSPACE_DETAIL_UNAVAILABLE", "message": ( f"workspace detail route for ws_id {ws_id} was unavailable; " "the tool used the visible workspace list as fallback." ), "backend_code": error.backend_code, "http_status": error.http_status, "request_id": error.request_id, }