from __future__ import annotations from dataclasses import dataclass from mcp.server.fastmcp import FastMCP from ..backend_client import BackendClient from ..config import get_feedback_app_key, get_feedback_base_url, get_feedback_qsource_token, normalize_base_url from ..errors import QingflowApiError, raise_tool_error from ..json_types import JSONObject CATEGORY_MAP = { "feature_request": "功能需求", "bug_report": "问题反馈", "ux_feedback": "体验建议", "unsupported_scenario": "不支持场景", "other": "其他", } IMPACT_SCOPE_MAP = { "personal": "仅个人", "small_team": "小范围团队", "cross_team": "跨团队", "global": "全局", } @dataclass(slots=True) class FeedbackTools: backend: BackendClient mcp_side: str def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool() def feedback_submit( category: str = "", title: str = "", description: str = "", expected_behavior: str | None = None, actual_behavior: str | None = None, impact_scope: str | None = None, tool_name: str | None = None, app_key: str | None = None, record_id: str | int | None = None, workflow_node_id: str | int | None = None, note: str | None = None, ) -> JSONObject: """Submit product feedback to the Qingflow MCP team. Use this when the current MCP capability is unsupported, awkward, or still cannot satisfy the user's need after reasonable attempts. This helper writes through the internal q-source feedback intake, does not require Qingflow login or workspace selection, and should be called only after explicit user confirmation. """ return self.feedback_submit( category=category, title=title, description=description, expected_behavior=expected_behavior, actual_behavior=actual_behavior, impact_scope=impact_scope, tool_name=tool_name, app_key=app_key, record_id=record_id, workflow_node_id=workflow_node_id, note=note, ) def feedback_submit( self, *, category: str, title: str, description: str, expected_behavior: str | None, actual_behavior: str | None, impact_scope: str | None, tool_name: str | None, app_key: str | None, record_id: str | int | None, workflow_node_id: str | int | None, note: str | None, ) -> JSONObject: """执行工具方法逻辑。""" qsource_token = get_feedback_qsource_token() if not qsource_token: raise_tool_error( QingflowApiError( category="config", message=( "feedback_submit is not configured. Set " "feedback.qsource_token or QINGFLOW_MCP_FEEDBACK_QSOURCE_TOKEN first." ), details={"error_code": "FEEDBACK_NOT_CONFIGURED"}, ) ) base_url = get_feedback_base_url() if not base_url: raise_tool_error( QingflowApiError.config_error( "feedback_submit requires a base_url. Configure feedback.base_url or default_base_url." ) ) normalized_payload = self._build_payload( category=category, title=title, description=description, expected_behavior=expected_behavior, actual_behavior=actual_behavior, impact_scope=impact_scope, tool_name=tool_name, app_key=app_key, record_id=record_id, workflow_node_id=workflow_node_id, note=note, ) try: response = self.backend.public_request_with_meta( "POST", base_url, f"/qsource/{qsource_token}", json_body=normalized_payload, unwrap=True, qf_version=None, ) except QingflowApiError as exc: raise_tool_error(exc) result = response.data if isinstance(response.data, dict) else {} feedback_request_id = result.get("requestId") if isinstance(result, dict) else None return { "ok": True, "request_route": { "base_url": normalize_base_url(base_url) or base_url, "qf_version": None, "qf_version_source": "not_applicable", }, "submission_mode": "qsource_passive", "feedback_target": { "app_key": get_feedback_app_key(), "mcp_side": self.mcp_side, }, "submission_summary": { "category": normalized_payload.get("反馈类型"), "title": normalized_payload.get("title"), "tool_name": normalized_payload.get("关联工具"), "app_key": normalized_payload.get("关联应用"), "record_id": normalized_payload.get("关联记录"), "workflow_node_id": normalized_payload.get("关联节点"), "impact_scope": normalized_payload.get("影响范围"), }, "normalized_payload": normalized_payload, "feedback_request_id": feedback_request_id, } def _build_payload( self, *, category: str, title: str, description: str, expected_behavior: str | None, actual_behavior: str | None, impact_scope: str | None, tool_name: str | None, app_key: str | None, record_id: str | int | None, workflow_node_id: str | int | None, note: str | None, ) -> JSONObject: """执行内部辅助逻辑。""" payload: JSONObject = { "title": self._require_text("title", title), "category": self._normalize_label("category", category, CATEGORY_MAP, required=True), "description": self._require_text("description", description), "submit_method": "AI代提", "status": "待处理", "mcp_side": self.mcp_side, } optional_text = { "expected_result": expected_behavior, "actual_behavior": actual_behavior, "tool_name": tool_name, "app_key": app_key, "note": note, } for key, value in optional_text.items(): normalized = self._normalize_optional_text(value) if normalized is not None: payload[key] = normalized if impact_scope is not None and str(impact_scope).strip(): payload["impact_scope"] = self._normalize_label("impact_scope", impact_scope, IMPACT_SCOPE_MAP, required=False) if record_id is not None and str(record_id).strip(): payload["record_id"] = str(record_id).strip() if workflow_node_id is not None and str(workflow_node_id).strip(): payload["workflow_node_id"] = str(workflow_node_id).strip() return payload def _normalize_label(self, field: str, value: str, mapping: dict[str, str], *, required: bool) -> str: """执行内部辅助逻辑。""" text = str(value or "").strip() if not text: if required: raise_tool_error(QingflowApiError.config_error(f"{field} is required")) return "" canonical = text.lower() if canonical in mapping: return mapping[canonical] if text in mapping.values(): return text supported_values = list(mapping.keys()) + list(mapping.values()) raise_tool_error( QingflowApiError( category="config", message=f"{field} must be one of the supported canonical values or labels", details={ "error_code": "FEEDBACK_INVALID_INPUT", "field": field, "supported_values": supported_values, }, ) ) def _require_text(self, field: str, value: str) -> str: """执行内部辅助逻辑。""" normalized = str(value or "").strip() if not normalized: raise_tool_error(QingflowApiError.config_error(f"{field} is required")) return normalized def _normalize_optional_text(self, value: str | None) -> str | None: """执行内部辅助逻辑。""" if value is None: return None normalized = str(value).strip() return normalized or None