from __future__ import annotations from mcp.server.fastmcp import FastMCP from ..backend_client import BackendRequestContext 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): """流程工具（中文名：流程节点与规则管理）。 类型：流程配置工具。 主要职责： 1. 支持流程运行时校验与调试辅助能力； 2. 保留 legacy 节点写路径供内部/测试直接调用（非 MCP 公开）。 设计期流程配置读取统一走 WorkflowSpec（app_flow_get / GET /api/workflow/spec）。 """ def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @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, ) 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, tool_name="流程节点列表") 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, tool_name="流程节点详情") 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_with_post_fallbacks( profile=profile, app_key=app_key, path=f"/app/{app_key}/auditNodes", json_body=body, alternate_paths=[f"/app/{app_key}/auditNode"], tool_name="新增流程节点", ) 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_with_post_fallbacks( profile=profile, app_key=app_key, path=f"/app/{app_key}/auditNodes/{audit_node_id}", json_body=body, alternate_paths=[f"/app/{app_key}/auditNode/{audit_node_id}"], risk_operation="update", risk_target="workflow node configuration", audit_node_id=audit_node_id, tool_name="更新流程节点", ) 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", tool_name="删除流程节点") 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, tool_name="复制粘贴流程节点") 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, tool_name="剪切粘贴流程节点") 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, tool_name="创建流程分支") 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", tool_name="删除流程分支") 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, tool_name="流程全局设置") 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_with_post_fallbacks( profile=profile, app_key=app_key, path=f"/app/{app_key}/workflow/global/setting", json_body=body, alternate_paths=[], risk_operation="update", risk_target="workflow global settings", tool_name="更新流程全局设置", ) 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_with_post_fallbacks( profile=profile, app_key=app_key, path=f"/app/{app_key}/publish", json_body=body, alternate_paths=[], tool_name="发布流程", ) 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, tool_name="流程后续节点") 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, tool_name="流程后续节点（应用）", ) 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, tool_name="流程 Webhook 测试") 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, tool_name="流程 QSource 查询") 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, tool_name="流程 QSource 测试") 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}, tool_name="流程主动 QSource 配置") 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, tool_name="更新主动 QSource 配置") 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}, tool_name="流程被动 QSource 配置") 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, tool_name="更新被动 QSource 配置") 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, tool_name="切换 QSource 状态") 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", tool_name="删除 QSource 配置") 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, tool_name="节点可编辑字段", ) 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, tool_name="流程打印节点") 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, tool_name: str, **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, tool_name=tool_name) def _request_with_post_fallbacks( self, *, profile: str, app_key: str, path: str, json_body: JSONObject, alternate_paths: list[str], risk_operation: str | None = None, risk_target: str | None = None, tool_name: str, **extra: JSONValue, ) -> JSONObject: """执行内部辅助逻辑。""" def runner(session_profile, context): attempted_contexts = [context] if context.qf_version is not None: attempted_contexts.append( BackendRequestContext( base_url=context.base_url, token=context.token, ws_id=context.ws_id, qf_version=None, qf_version_source="workflow_retry_without_qf_version", ) ) paths = [path, *alternate_paths] last_error: QingflowApiError | None = None for call_context in attempted_contexts: for candidate_path in paths: try: result = self.backend.request("POST", call_context, candidate_path, json_body=json_body) fallback_applied = candidate_path != path or call_context is not context response: JSONObject = { "profile": profile, "ws_id": session_profile.selected_ws_id, "result": result, "request_path": candidate_path, "requested_path": path, "fallback_applied": fallback_applied, "qf_version_source": ( call_context.qf_version_source or ("context" if call_context.qf_version else "unset") ), "verification": { "primary_path_used": candidate_path == path, "qf_version_used": call_context.qf_version is not None, }, } response.update(extra) if risk_operation and risk_target: return self._attach_human_review_notice(response, operation=risk_operation, target=risk_target) return response except QingflowApiError as error: last_error = error if error.http_status != 404: raise assert last_error is not None raise last_error 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_positive(self, field_name: str, value: int) -> None: """执行内部辅助逻辑。""" if value <= 0: raise_tool_error(QingflowApiError.config_error(f"{field_name} must be positive"))