from __future__ import annotations 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, 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 class QingbiReportTools(ToolBase): def register(self, mcp: FastMCP) -> None: @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") 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) def qingbi_report_get_base(self, *, profile: str, chart_id: str) -> JSONObject: self._require_chart_id(chart_id) return self._request(profile, "GET", f"/qingbi/charts/baseinfo/{chart_id}", chart_id=chart_id) 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") def qingbi_report_get_config(self, *, profile: str, chart_id: str) -> JSONObject: self._require_chart_id(chart_id) return self._request(profile, "GET", f"/qingbi/charts/{chart_id}/configs", chart_id=chart_id) 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") 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} if page_num is not None: params["pageNum"] = page_num if page_size is not None: params["pageSize"] = page_size if page_num_y is not None: params["pageNumY"] = page_num_y if page_size_y is not None: params["pageSizeY"] = page_size_y return self._request(profile, "POST", f"/qingbi/charts/data/{chart_id}", chart_id=chart_id, params=params, json_body=payload) 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") 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}, ) 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, ) 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") 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, **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) 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 []