from __future__ import annotations from typing import Any from mcp.server.fastmcp import FastMCP from ..config import DEFAULT_PROFILE from ..errors import QingflowApiError, backend_code_int, is_auth_like_error, raise_tool_error from ..json_types import JSONObject from .base import ToolBase class RoleTools(ToolBase): """角色工具（中文名：角色检索与创建）。 类型：组织权限工具。 主要职责： 1. 搜索角色并提供候选列表； 2. 创建角色并返回标准角色信息； 3. 服务流程节点与权限配置场景。 """ def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool(description="Search workspace roles for builder permission and workflow configuration. Requires contact/role management permission and a non-empty keyword; do not use for record member/department field candidates.") def role_search( profile: str = DEFAULT_PROFILE, keyword: str = "", page_num: int = 1, page_size: int = 50, ) -> JSONObject: return self.role_search(profile=profile, keyword=keyword, page_num=page_num, page_size=page_size) @mcp.tool() def role_create(profile: str = DEFAULT_PROFILE, payload: JSONObject | None = None) -> JSONObject: return self.role_create(profile=profile, payload=payload or {}) @mcp.tool(description=self._high_risk_tool_description(operation="update", target="role configuration")) def role_update( profile: str = DEFAULT_PROFILE, role_id: int = 0, payload: JSONObject | None = None, ) -> JSONObject: return self.role_update(profile=profile, role_id=role_id, payload=payload or {}) @mcp.tool(description=self._high_risk_tool_description(operation="delete", target="roles and assignments")) def role_delete(profile: str = DEFAULT_PROFILE, role_ids: list[int] | None = None) -> JSONObject: return self.role_delete(profile=profile, role_ids=role_ids or []) def role_search(self, *, profile: str, keyword: str, page_num: int, page_size: int) -> JSONObject: """执行角色相关逻辑。""" if page_num <= 0 or page_size <= 0: raise_tool_error(QingflowApiError.config_error("page_num and page_size must be positive")) normalized_keyword = keyword.strip() if not normalized_keyword: raise_tool_error( QingflowApiError.config_error( "keyword is required for role_search; role lookup is a contact-management path, not a record candidate fallback" ) ) def runner(session_profile, context): try: result = self.backend.request( "GET", context, "/contact/roleByPage", params={"keyword": normalized_keyword, "pageNum": page_num, "pageSize": page_size}, ) except QingflowApiError as exc: if _is_role_permission_denied(exc): return _contact_role_permission_denied_payload( profile=profile, ws_id=session_profile.selected_ws_id, error=exc, operation="role_search", selection={"keyword": normalized_keyword, "page_num": page_num, "page_size": page_size}, ) raise return { "profile": profile, "ws_id": session_profile.selected_ws_id, "keyword": normalized_keyword, "page": result, } return self._run(profile, runner, tool_name='角色搜索') def role_create(self, *, profile: str, payload: JSONObject) -> JSONObject: """执行角色相关逻辑。""" body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("POST", context, "/contact/role", json_body=body) return {"profile": profile, "ws_id": session_profile.selected_ws_id, "result": result} return self._run(profile, runner, tool_name='创建角色') def role_update(self, *, profile: str, role_id: int, payload: JSONObject) -> JSONObject: """执行角色相关逻辑。""" if role_id <= 0: raise_tool_error(QingflowApiError.config_error("role_id must be positive")) body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("POST", context, f"/contact/role/{role_id}", json_body=body) return self._attach_human_review_notice( {"profile": profile, "ws_id": session_profile.selected_ws_id, "role_id": role_id, "result": result}, operation="update", target="role configuration", ) return self._run(profile, runner, tool_name='更新角色') def role_delete(self, *, profile: str, role_ids: list[int]) -> JSONObject: """执行角色相关逻辑。""" if not role_ids or any(role_id <= 0 for role_id in role_ids): raise_tool_error(QingflowApiError.config_error("role_ids must be a non-empty array of positive integers")) def runner(session_profile, context): result = self.backend.request("DELETE", context, "/contact/role", json_body=role_ids) return self._attach_human_review_notice( {"profile": profile, "ws_id": session_profile.selected_ws_id, "role_ids": role_ids, "result": result}, operation="delete", target="roles and assignments", ) return self._run(profile, runner, tool_name='删除角色') def _is_role_permission_denied(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False return backend_code_int(error) in {40002, 40027} def _contact_role_permission_denied_payload( *, profile: str, ws_id: Any, error: QingflowApiError, operation: str, selection: JSONObject, ) -> JSONObject: return { "profile": profile, "ws_id": ws_id, "ok": False, "status": "failed", "error_code": "CONTACT_ROLE_PERMISSION_DENIED", "message": ( "Contact role-management data is not readable for the current user. " "This is a role/contact-management permission boundary, not proof that record " "member or department field candidates are unavailable." ), "backend_code": error.backend_code, "request_id": error.request_id, "http_status": error.http_status, "keyword": selection.get("keyword"), "page": {"list": [], "total": 0, "pageAmount": 0}, "data": { "items": [], "pagination": { "page": selection.get("page_num"), "page_size": selection.get("page_size"), "returned_items": 0, "reported_total": 0, "page_amount": 0, }, "selection": selection, }, "warnings": [ { "code": "CONTACT_ROLE_PERMISSION_DENIED", "message": "This is a role/contact-management permission boundary, not a record field candidate failure.", "operation": operation, } ], }