from __future__ import annotations from mcp.server.fastmcp import FastMCP from ..config import DEFAULT_PROFILE from ..errors import QingflowApiError, raise_tool_error from ..json_types import JSONObject, JSONValue from .base import ToolBase class WorkflowTools(ToolBase): def register(self, mcp: FastMCP) -> None: @mcp.tool() def workflow_list_nodes(profile: str = DEFAULT_PROFILE, app_key: str = "") -> JSONObject: return self.workflow_list_nodes(profile=profile, app_key=app_key) @mcp.tool() def workflow_get_node_detail(profile: str = DEFAULT_PROFILE, app_key: str = "", audit_node_id: int = 0) -> JSONObject: return self.workflow_get_node_detail(profile=profile, app_key=app_key, audit_node_id=audit_node_id) @mcp.tool() def workflow_get_global_settings(profile: str = DEFAULT_PROFILE, app_key: str = "") -> JSONObject: return self.workflow_get_global_settings(profile=profile, app_key=app_key) @mcp.tool() def workflow_get_future_nodes(profile: str = DEFAULT_PROFILE, app_key: str = "", apply_id: int = 0) -> JSONObject: return self.workflow_get_future_nodes(profile=profile, app_key=app_key, apply_id=apply_id) @mcp.tool() def workflow_get_future_nodes_app( profile: str = DEFAULT_PROFILE, app_key: str = "", apply_id: int = 0, role: int = 1, audit_node_id: int | None = None, ) -> JSONObject: return self.workflow_get_future_nodes_app( profile=profile, app_key=app_key, apply_id=apply_id, role=role, audit_node_id=audit_node_id, ) @mcp.tool() def workflow_get_qsource_active(profile: str = DEFAULT_PROFILE, app_key: str = "", qsource_id: int = 0) -> JSONObject: return self.workflow_get_qsource_active(profile=profile, app_key=app_key, qsource_id=qsource_id) @mcp.tool() def workflow_get_qsource_passive(profile: str = DEFAULT_PROFILE, app_key: str = "", qsource_id: int = 0) -> JSONObject: return self.workflow_get_qsource_passive(profile=profile, app_key=app_key, qsource_id=qsource_id) @mcp.tool() def workflow_get_editable_question_ids(profile: str = DEFAULT_PROFILE, app_key: str = "", audit_node_id: int = 0) -> JSONObject: return self.workflow_get_editable_question_ids(profile=profile, app_key=app_key, audit_node_id=audit_node_id) @mcp.tool() def workflow_get_print_nodes(profile: str = DEFAULT_PROFILE, app_key: str = "") -> JSONObject: return self.workflow_get_print_nodes(profile=profile, app_key=app_key) def workflow_list_nodes(self, *, profile: str, app_key: str) -> JSONObject: self._require_app_key(app_key) return self._request(profile, "GET", f"/app/{app_key}/auditNodes", app_key=app_key) def workflow_get_node_detail(self, *, profile: str, app_key: str, audit_node_id: int) -> JSONObject: self._require_app_key(app_key) self._require_positive("audit_node_id", audit_node_id) return self._request(profile, "GET", f"/app/{app_key}/auditNodes/{audit_node_id}", app_key=app_key, audit_node_id=audit_node_id) def workflow_add_node(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/auditNodes", app_key=app_key, json_body=body) def workflow_update_node(self, *, profile: str, app_key: str, audit_node_id: int, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) self._require_positive("audit_node_id", audit_node_id) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/auditNodes/{audit_node_id}", app_key=app_key, audit_node_id=audit_node_id, json_body=body, risk_operation="update", risk_target="workflow node configuration") def workflow_delete_node(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "DELETE", f"/app/{app_key}/auditNode", app_key=app_key, json_body=body, risk_operation="delete", risk_target="workflow node configuration") def workflow_copy_paste_node(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/auditNode/copyAndPaste", app_key=app_key, json_body=body) def workflow_cut_paste_node(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/auditNode/cutAndPaste", app_key=app_key, json_body=body) def workflow_create_sub_branch(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/auditNode/subBranch", app_key=app_key, json_body=body) def workflow_delete_sub_branch(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "DELETE", f"/app/{app_key}/auditNode/subBranch", app_key=app_key, json_body=body, risk_operation="delete", risk_target="workflow branch configuration") def workflow_get_global_settings(self, *, profile: str, app_key: str) -> JSONObject: self._require_app_key(app_key) return self._request(profile, "GET", f"/app/{app_key}/workflow/global/setting", app_key=app_key) def workflow_update_global_settings(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/workflow/global/setting", app_key=app_key, json_body=body, risk_operation="update", risk_target="workflow global settings") def workflow_publish(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/publish", app_key=app_key, json_body=body) def workflow_get_future_nodes(self, *, profile: str, app_key: str, apply_id: int) -> JSONObject: self._require_app_key(app_key) self._require_positive("apply_id", apply_id) return self._request(profile, "GET", f"/app/{app_key}/auditNode/futureList/{apply_id}", app_key=app_key, apply_id=apply_id) def workflow_get_future_nodes_app( self, *, profile: str, app_key: str, apply_id: int, role: int, audit_node_id: int | None, ) -> JSONObject: self._require_app_key(app_key) self._require_positive("apply_id", apply_id) params: JSONObject = {"role": role} if audit_node_id is not None: self._require_positive("audit_node_id", audit_node_id) params["auditNodeId"] = audit_node_id return self._request( profile, "GET", f"/app/{app_key}/auditNode/appFutureListV2/{apply_id}", app_key=app_key, apply_id=apply_id, params=params, ) def workflow_webhook_test(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/auditNode/webhookTest", app_key=app_key, json_body=body) def workflow_qsource_query(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/auditNode/qSourceQuery", app_key=app_key, json_body=body) def workflow_qsource_test(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/auditNode/qSourceQueryTest", app_key=app_key, json_body=body) def workflow_get_qsource_active(self, *, profile: str, app_key: str, qsource_id: int) -> JSONObject: self._require_app_key(app_key) self._require_positive("qsource_id", qsource_id) return self._request(profile, "GET", f"/app/{app_key}/auditNode/active/qsource", app_key=app_key, params={"qSourceId": qsource_id}) def workflow_upsert_qsource_active(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/auditNode/active/qsource", app_key=app_key, json_body=body) def workflow_get_qsource_passive(self, *, profile: str, app_key: str, qsource_id: int) -> JSONObject: self._require_app_key(app_key) self._require_positive("qsource_id", qsource_id) return self._request(profile, "GET", f"/app/{app_key}/auditNode/passive/qsource", app_key=app_key, params={"qSourceId": qsource_id}) def workflow_upsert_qsource_passive(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/auditNode/passive/qsource", app_key=app_key, json_body=body) def workflow_switch_qsource_status(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "POST", f"/app/{app_key}/auditNode/source/status", app_key=app_key, json_body=body) def workflow_delete_qsource(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: self._require_app_key(app_key) body = self._require_dict(payload) return self._request(profile, "DELETE", f"/app/{app_key}/auditNode/qsource", app_key=app_key, json_body=body, risk_operation="delete", risk_target="workflow qsource configuration") def workflow_get_editable_question_ids(self, *, profile: str, app_key: str, audit_node_id: int) -> JSONObject: self._require_app_key(app_key) self._require_positive("audit_node_id", audit_node_id) return self._request( profile, "GET", f"/app/{app_key}/auditNode/{audit_node_id}/editableQueIds", app_key=app_key, audit_node_id=audit_node_id, ) def workflow_get_print_nodes(self, *, profile: str, app_key: str) -> JSONObject: self._require_app_key(app_key) return self._request(profile, "GET", f"/app/{app_key}/auditNode/printNodes", app_key=app_key) def _request( self, profile: str, method: str, path: str, *, app_key: str, json_body: JSONValue = None, params: JSONObject | None = None, risk_operation: str | None = None, risk_target: str | None = None, **extra: JSONValue, ) -> JSONObject: def runner(session_profile, context): result = self.backend.request(method, context, path, json_body=json_body, params=params) response: JSONObject = {"profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "result": result} response.update(extra) if risk_operation and risk_target: return self._attach_human_review_notice(response, operation=risk_operation, target=risk_target) return response 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_positive(self, field_name: str, value: int) -> None: if value <= 0: raise_tool_error(QingflowApiError.config_error(f"{field_name} must be positive"))