from __future__ import annotations import json from uuid import uuid4 from mcp.server.fastmcp import FastMCP from ..backend_client import BackendRequestContext from ..config import DEFAULT_PROFILE, normalize_base_url from ..errors import QingflowApiError, backend_code_int, is_auth_like_error, raise_tool_error from ..json_types import JSONObject, JSONValue from .base import ToolBase def _qingbi_base_url(base_url: str) -> str: normalized = normalize_base_url(base_url) if not normalized: raise QingflowApiError.config_error("base_url is required") return normalized[:-4] if normalized.endswith("/api") else normalized def _should_retry_qflow_base(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False backend_code = backend_code_int(error) http_status = getattr(error, "http_status", None) return backend_code in {40002, 40027, 404, 81007} or http_status == 404 def _should_retry_asos_data(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False backend_code = backend_code_int(error) http_status = getattr(error, "http_status", None) return backend_code in {40002, 40027, 404, 44011, 81007} or http_status == 404 def _should_fallback_config_from_data(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False backend_code = backend_code_int(error) http_status = getattr(error, "http_status", None) return backend_code in {40002, 40027, 404, 81007} or http_status == 404 def _coerce_tool_error(error: RuntimeError | QingflowApiError) -> QingflowApiError | None: if isinstance(error, QingflowApiError): return error if not isinstance(error, RuntimeError): return None try: payload = json.loads(str(error)) except Exception: return None if not isinstance(payload, dict): return None return QingflowApiError( category=str(payload.get("category") or "runtime"), message=str(payload.get("message") or str(error)), backend_code=payload.get("backend_code"), request_id=payload.get("request_id"), http_status=payload.get("http_status"), details=payload.get("details") if isinstance(payload.get("details"), dict) else None, ) class QingbiReportTools(ToolBase): """轻报表工具（中文名：图表与报表读取）。 类型：报表分析工具。 主要职责： 1. 查询报表列表与基础信息； 2. 获取报表配置与图表数据； 3. 为门户与任务上下文提供报表侧读取能力。 """ def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool() def qingbi_report_list( profile: str = DEFAULT_PROFILE, app_key: str = "", ) -> JSONObject: return self.qingbi_report_list(profile=profile, app_key=app_key) @mcp.tool() def qingbi_report_get_base( profile: str = DEFAULT_PROFILE, chart_id: str = "", ) -> JSONObject: return self.qingbi_report_get_base(profile=profile, chart_id=chart_id) @mcp.tool() def qingbi_report_get_config( profile: str = DEFAULT_PROFILE, chart_id: str = "", ) -> JSONObject: return self.qingbi_report_get_config(profile=profile, chart_id=chart_id) @mcp.tool() def qingbi_report_get_data( profile: str = DEFAULT_PROFILE, chart_id: str = "", payload: JSONObject | None = None, page_num: int | None = None, page_size: int | None = None, page_num_y: int | None = None, page_size_y: int | None = None, ) -> JSONObject: return self.qingbi_report_get_data( profile=profile, chart_id=chart_id, payload=payload or {}, page_num=page_num, page_size=page_size, page_num_y=page_num_y, page_size_y=page_size_y, ) @mcp.tool(description=self._high_risk_tool_description(operation="delete", target="report chart")) def qingbi_report_delete( profile: str = DEFAULT_PROFILE, chart_id: str = "", ) -> JSONObject: return self.qingbi_report_delete(profile=profile, chart_id=chart_id) @mcp.tool() def qingbi_report_reorder( profile: str = DEFAULT_PROFILE, app_key: str = "", chart_ids: list[str] | None = None, ) -> JSONObject: return self.qingbi_report_reorder(profile=profile, app_key=app_key, chart_ids=chart_ids or []) @mcp.tool() def qingbi_report_list_fields( profile: str = DEFAULT_PROFILE, app_key: str = "", ) -> JSONObject: return self.qingbi_report_list_fields(profile=profile, app_key=app_key) @mcp.tool() def qingbi_report_list_field_options( profile: str = DEFAULT_PROFILE, field_id: str = "", chart_id: str = "", ) -> JSONObject: return self.qingbi_report_list_field_options(profile=profile, field_id=field_id, chart_id=chart_id) def qingbi_report_list(self, *, profile: str, app_key: str) -> JSONObject: """执行工具方法逻辑。""" self._require_app_key(app_key) return self._request(profile, "GET", f"/qingbi/charts/data/bichart/{app_key}", app_key=app_key, items_key="items", tool_name="报表列表") def qingbi_report_list_sorted( self, *, profile: str, app_key: str, page_num: int = 1, page_size: int = 500, search_key: str | None = None, ) -> JSONObject: """执行工具方法逻辑。""" self._require_app_key(app_key) body: JSONObject = {"appKey": app_key, "pageNum": page_num, "pageSize": page_size} if search_key: body["searchKey"] = search_key return self._request( profile, "POST", "/qingbi/charts/chart/list", app_key=app_key, page_num=page_num, page_size=page_size, search_key=search_key, items_key="items", result_transform=_extract_sorted_chart_items, json_body=body, tool_name="报表排序列表", ) def qingbi_report_create(self, *, profile: str, payload: JSONObject) -> JSONObject: """执行工具方法逻辑。""" body = self._require_dict(payload) return self._request(profile, "POST", "/qingbi/charts", json_body=body, tool_name="创建报表") def qingbi_report_get_base(self, *, profile: str, chart_id: str) -> JSONObject: """执行工具方法逻辑。""" self._require_chart_id(chart_id) source_error: QingflowApiError | None = None try: return self._request(profile, "GET", f"/qingbi/charts/qflow/baseinfo/{chart_id}", chart_id=chart_id, tool_name="报表基础信息") except (QingflowApiError, RuntimeError) as raw_error: error = _coerce_tool_error(raw_error) if error is None or not _should_retry_qflow_base(error): raise source_error = error try: fallback = self._request(profile, "GET", f"/qingbi/charts/baseinfo/{chart_id}", chart_id=chart_id, tool_name="报表基础信息") except (QingflowApiError, RuntimeError): raise if not _has_chart_base_identity(fallback.get("result"), chart_id=chart_id): raise source_error fallback.setdefault( "warnings", [ { "code": "CHART_BASE_INFO_FALLBACK_FROM_LEGACY", "message": "qingbi_report_get_base used the legacy chart baseInfo route because qflow baseInfo was unavailable in this permission context.", "backend_code": source_error.backend_code, "http_status": source_error.http_status, "request_id": source_error.request_id, } ], ) fallback.setdefault( "verification", {"qflow_base_route_loaded": False, "legacy_base_route_loaded": True}, ) return fallback def qingbi_report_update_base(self, *, profile: str, chart_id: str, payload: JSONObject) -> JSONObject: """执行工具方法逻辑。""" self._require_chart_id(chart_id) body = self._require_dict(payload) return self._request(profile, "PUT", f"/qingbi/charts/baseinfo/{chart_id}", chart_id=chart_id, json_body=body, risk_operation="update", risk_target="report base settings", tool_name="更新报表基础信息") def qingbi_report_get_config(self, *, profile: str, chart_id: str) -> JSONObject: """执行工具方法逻辑。""" self._require_chart_id(chart_id) original_error: QingflowApiError | RuntimeError | None = None try: return self._request(profile, "GET", f"/qingbi/charts/{chart_id}/configs", chart_id=chart_id, tool_name="报表配置") except (QingflowApiError, RuntimeError) as raw_error: original_error = raw_error error = _coerce_tool_error(raw_error) if error is None or not _should_fallback_config_from_data(error): raise source_error = error try: data_payload = self.qingbi_report_get_data(profile=profile, chart_id=chart_id, payload={}) except (QingflowApiError, RuntimeError): if original_error is not None: raise original_error raise data_result = data_payload.get("result") if isinstance(data_payload, dict) else None if isinstance(data_result, dict) and isinstance(data_result.get("config"), dict) and data_result["config"]: return { "profile": data_payload.get("profile", profile), "ws_id": data_payload.get("ws_id"), "chart_id": chart_id, "result": data_result["config"], "warnings": [ { "code": "CHART_CONFIG_FALLBACK_FROM_DATA", "message": "qingbi_report_get_config used config embedded in qflow chart data because the chart config endpoint was unavailable in this permission context.", "backend_code": source_error.backend_code, "http_status": source_error.http_status, "request_id": source_error.request_id, } ], "verification": { "config_endpoint_loaded": False, "data_config_loaded": True, }, } if original_error is not None: raise original_error raise_tool_error(QingflowApiError.config_error("chart config fallback did not return config")) def qingbi_report_update_config(self, *, profile: str, chart_id: str, payload: JSONObject) -> JSONObject: """执行工具方法逻辑。""" self._require_chart_id(chart_id) body = self._require_dict(payload) return self._request(profile, "PUT", f"/qingbi/charts/{chart_id}/configs", chart_id=chart_id, json_body=body, risk_operation="update", risk_target="report chart config", tool_name="更新报表配置") def qingbi_report_get_data( self, *, profile: str, chart_id: str, payload: JSONObject, page_num: int | None = None, page_size: int | None = None, page_num_y: int | None = None, page_size_y: int | None = None, ) -> JSONObject: """执行工具方法逻辑。""" self._require_chart_id(chart_id) params = { "qfUUID": uuid4().hex, # Match Qingflow's real BI/qflow read path defaults so a caller can # fetch chart data without having to synthesize pagination first. "pageNum": page_num if page_num is not None else 1, "pageSize": page_size if page_size is not None else 20, } if page_num_y is not None: params["pageNumY"] = page_num_y if page_size_y is not None: params["pageSizeY"] = page_size_y try: if payload: return self._request( profile, "POST", f"/qingbi/charts/data/qflow/{chart_id}/detail", chart_id=chart_id, params=params, json_body=payload, tool_name="报表数据", ) return self._request( profile, "GET", f"/qingbi/charts/data/qflow/{chart_id}", chart_id=chart_id, params=params, tool_name="报表数据", ) except (QingflowApiError, RuntimeError) as raw_error: error = _coerce_tool_error(raw_error) if error is None or not _should_retry_asos_data(error): raise warning: JSONObject = { "code": "CHART_DATA_FALLBACK_FROM_ASOS", "message": "qingbi_report_get_data used the asos data route because the qflow chart data route was 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: warning["request_id"] = error.request_id return self._request( profile, "POST", f"/qingbi/charts/data/qflow/{chart_id}/asos", chart_id=chart_id, params=params, json_body=payload or {}, warnings=[warning], verification={ "qflow_data_route_loaded": False, "asos_data_route_loaded": True, }, tool_name="报表数据", ) def qingbi_report_delete(self, *, profile: str, chart_id: str) -> JSONObject: """执行工具方法逻辑。""" self._require_chart_id(chart_id) return self._request(profile, "DELETE", f"/qingbi/charts/{chart_id}", chart_id=chart_id, risk_operation="delete", risk_target="report chart", tool_name="删除报表") def qingbi_report_reorder(self, *, profile: str, app_key: str, chart_ids: list[str]) -> JSONObject: """执行工具方法逻辑。""" self._require_app_key(app_key) if not isinstance(chart_ids, list) or not chart_ids: raise_tool_error(QingflowApiError.config_error("chart_ids must be a non-empty array")) if any(not chart_id for chart_id in chart_ids): raise_tool_error(QingflowApiError.config_error("chart_ids cannot contain empty values")) return self._request( profile, "POST", "/qingbi/charts/sort/chart", app_key=app_key, chart_ids=chart_ids, json_body={"appKey": app_key, "sortChartIdList": chart_ids}, tool_name="报表排序", ) def qingbi_report_list_fields(self, *, profile: str, app_key: str) -> JSONObject: """执行工具方法逻辑。""" self._require_app_key(app_key) return self._request( profile, "GET", "/qingbi/datasets/datasource/fields", app_key=app_key, params={"dataSourceId": app_key, "dataSourceType": "qingflow"}, items_key="items", result_transform=_extract_dataset_fields, tool_name="报表字段列表", ) def qingbi_report_list_field_options(self, *, profile: str, field_id: str, chart_id: str) -> JSONObject: """执行工具方法逻辑。""" if not field_id: raise_tool_error(QingflowApiError.config_error("field_id is required")) path = f"/qingbi/charts/datasource/{chart_id}/question/option" if chart_id else "/qingbi/charts/datasource/question/option" params = {"fieldId": field_id} if chart_id: self._require_chart_id(chart_id) return self._request(profile, "GET", path, chart_id=chart_id or None, field_id=field_id, params=params, items_key="items", tool_name="报表字段选项") def _request( self, profile: str, method: str, path: str, *, json_body: JSONValue = None, params: JSONObject | None = None, items_key: str | None = None, result_transform: object | None = None, risk_operation: str | None = None, risk_target: str | None = None, tool_name: str, **extra: JSONValue, ) -> JSONObject: """执行内部辅助逻辑。""" def runner(session_profile, context): qingbi_context = BackendRequestContext( base_url=_qingbi_base_url(context.base_url), token=context.token, ws_id=context.ws_id, qf_request_id=context.qf_request_id, qf_version=context.qf_version, qf_version_source=context.qf_version_source, ) result = self.backend.request(method, qingbi_context, path, json_body=json_body, params=params) if callable(result_transform): result = result_transform(result) payload: JSONObject = {"profile": profile, "ws_id": session_profile.selected_ws_id} if items_key: payload[items_key] = result else: payload["result"] = result payload.update(extra) if risk_operation and risk_target: return self._attach_human_review_notice(payload, operation=risk_operation, target=risk_target) return payload 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_chart_id(self, chart_id: str) -> None: """执行内部辅助逻辑。""" if not chart_id: raise_tool_error(QingflowApiError.config_error("chart_id is required")) def _extract_dataset_fields(result: JSONValue) -> list[JSONObject]: if isinstance(result, dict) and isinstance(result.get("fields"), list): return result["fields"] if isinstance(result, list): return result return [] def _extract_sorted_chart_items(result: JSONValue) -> list[JSONObject]: if isinstance(result, dict) and isinstance(result.get("list"), list): return result["list"] if isinstance(result, list): return result return [] def _has_chart_base_identity(result: JSONValue, *, chart_id: str) -> bool: if not isinstance(result, dict): return False expected = str(chart_id or "").strip() id_candidates = ( result.get("chartId"), result.get("biChartId"), result.get("id"), result.get("chart_id"), ) if expected and any(str(value or "").strip() == expected for value in id_candidates): return True return any( str(result.get(key) or "").strip() for key in ("chartName", "chart_name", "name", "title") )