import json from constants import DEVICE_INVALID_MESSAGE # Builds AI-safe MCP results for the explicitly supported tool set. class OutputPresenter: TABLE_TOOLS = frozenset(( 'query_order_exact', 'query_track', 'query_customs_declaration_files', )) OPTION_TOOLS = frozenset(( 'list_order_filter_options', 'list_pending_outbound_export_filter_options', )) EXPORT_TOOLS = frozenset(( 'export_pending_outbound_orders', 'export_out_of_province_port_data', )) SAFE_TOOLS = TABLE_TOOLS | OPTION_TOOLS | EXPORT_TOOLS TABLE_COLUMNS = { 'query_order_exact': { 'order_number': ('订单号',), 'reference_number': ('客户参考号',), 'status_txt_name': ('状态',), 'check_status_txt_name': ('是否已查验',), 'customer_name': ('客户名称',), 'customer_account_type_name': ('客户属性',), 'inbound_date': ('入库时间',), 'wo_num': ('未完成工单',), 'product_name': ('物流产品',), 'inbound_pieces': ('件数',), 'inbound_volume': ('体积(CBM)',), 'inbound_weight': ('重量(KG)',), 'pro_cn_name': ('品名',), 'export_declaration_type': ('报关方式',), 'merge_declare_number': ('合并报关单号',), 'delivery_address': ('派送地址',), 'container_code': ('柜号',), 'out_status_txt': ('排舱单状态',), 'hinge_of_destination': ('目的港',), 'etd': ('ETD',), 'atd': ('ATD',), 'eta': ('ETA',), 'ata': ('ATA',), 'release_time': ('清关放行时间',), 'oversea_inbound_date': ('海外入库时间',), 'appt_time': ('APPT时间',), 'est_loading_time': ('预计装柜时间',), 'pickup_time': ('海外提柜时间',), 'delivery_way_title': ('派送方式',), 'tracking_number': ('快递单号', '承运商跟踪号码'), 'shipment_id': ('Shipment ID',), 'goods_attribute': ('商品属性',), 'sku': ('SKU',), 'sales_user': ('商务经理',), 'service_user': ('客户经理',), 'department_name': ('事业部',), 'remark': ('订单备注',), 'importer_name': ('进口商',), 'warehouse_name': ('交货仓库',), 'paid_status_name': ('付款状态',), }, 'query_track': { 'status': ('轨迹节点', '轨迹状态，如：开始集港、离港放行、清关、配送等'), 'location': ('轨迹地点', '发生地点'), 'time': ('时间', '轨迹发生时间（已转换为用户时区）'), 'content': ('轨迹内容', '详细描述'), 'tracking_number': ('跟踪号', '快递单号'), 'shipment_id': ('Shipment ID', '包裹ID'), }, 'query_customs_declaration_files': { 'outbound_number': ('排舱单号',), 'order_number': ('订单号',), 'file_name': ('文件名',), 'file_type': ('文件类型',), 'file_url': ('文件链接',), }, } FIELD_LABELS = { 'query_order_exact': { 'order_number': '订单号', 'order_numbers': '订单号', 'reference_number': '客户参考号', 'reference_numbers': '客户参考号', 'tracking_number': '快递单号', 'tracking_numbers': '快递单号', 'outbound_number': '排舱单号', 'outbound_numbers': '排舱单号', 'container_code': '柜号', 'container_codes': '柜号', 'so_number': 'SO号', 'so_numbers': 'SO号', 'shipment_id': 'Shipment ID', 'receiver_country': '收货国家', 'product_ids': '物流产品', 'customer_ids': '客户', 'sales_id': '销售人员', 'warehouse_ids': '仓库', 'department_id': '事业部', 'inbound_date_start': '入库开始日期', 'inbound_date_end': '入库结束日期', 'outbound_date_start': '出库开始日期', 'outbound_date_end': '出库结束日期', 'page': '页码', 'limit': '每页数量', }, 'query_track': { 'order_id': '订单ID', 'order_number': '订单号', 'tracking_number': '快递单号', 'page': '页码', 'limit': '每页数量', }, 'query_customs_declaration_files': { 'outbound_numbers': '排舱单号', 'order_numbers': '订单号', 'page': '页码', 'limit': '每页数量', }, 'list_order_filter_options': { 'filter_type': '筛选项类型', 'keyword': '关键词', 'page': '页码', 'limit': '每页数量', }, 'list_pending_outbound_export_filter_options': { 'filter_type': '筛选项类型', 'product_type_id': '产品分类', 'keyword': '关键词', 'page': '页码', 'limit': '每页数量', }, 'export_pending_outbound_orders': { 'number': '单号', 'product_type_id': '产品分类', 'product_id': '物流产品', 'order_warehouse_id': '集货仓库', 'receiver_country': '目的国', 'is_remove': '是否装柜剔除', 'address': '派送地址', 'inbound_date': '入库时间', 'status': '订单状态', 'merge_declare_number': '合并报关单号', 'importer_id': '进口商', 'packing_type': '货物类型', 'is_battery': '带电属性', 'is_magnetic': '带磁属性', 'is_wood': '带木属性', 'is_other': '其他商品属性', 'is_fda': 'FDA产品属性', 'is_toy': '玩具属性', 'is_ultra_limit': '超长超重属性', 'is_sensitive': '敏感货属性', 'is_food': '食品属性', 'no_property': '无属性', }, 'export_out_of_province_port_data': { 'outbound_numbers': '排舱单号', 'container_codes': '柜号', 'bl_numbers': '提单号', 'so_numbers': 'SO号', 'file_type': '资料类型', }, } ERROR_MESSAGES = { 'MCP_1101': '设备配置已失效，请重新生成配置', 'MCP_1102': '设备会话已过期，请重新连接', 'MCP_1201': '工具当前不可用', 'MCP_1202': '工具当前不可用', 'MCP_1301': '没有权限使用此工具', 'MCP_9001': '工具调用失败，请稍后重试', } def handles(self, tool_name): return isinstance(tool_name, str) and tool_name in self.SAFE_TOOLS # Convert a backend envelope into text and structured display data. def present(self, tool_name, tool_result): if not self.handles(tool_name) or not isinstance(tool_result, dict): return self._format_error() code = str(tool_result.get('code') or '').strip() meta = self._build_meta(tool_result.get('meta')) if code not in ('MCP_0000', '0'): return self._business_error( tool_name, code or 'MCP_9001', tool_result.get('msg'), meta, ) data = tool_result.get('data') if not isinstance(data, dict): return self._format_error(meta) if tool_name in self.TABLE_TOOLS: return self._present_table( tool_name, data, tool_result.get('meta'), meta, ) if tool_name in self.OPTION_TOOLS: return self._present_options(data, tool_result.get('meta'), meta) return self._present_export(data, meta) # Map local tool exceptions without exposing internal error details. def present_exception(self, tool_name, exception): if not self.handles(tool_name): return self._format_error() message = str(exception or '').strip() if isinstance(exception, ValueError): label = self._find_field_label(tool_name, message) return self._error_result( 'MCP_1401', self._parameter_message(label), False, ) if message == DEVICE_INVALID_MESSAGE: return self._error_result('MCP_1101', DEVICE_INVALID_MESSAGE, False) if message.lower().startswith('tool disabled'): return self._error_result('MCP_1202', '工具当前不可用', False) return self._error_result( 'MCP_9001', '工具调用失败，请稍后重试', True, ) def _present_table(self, tool_name, data, raw_meta, meta): columns = data.get('columns') records = data.get('records') if not isinstance(columns, list) or not columns or not isinstance(records, list): return self._format_error(meta) allowed_columns = self.TABLE_COLUMNS.get(tool_name) if not isinstance(allowed_columns, dict): return self._format_error(meta) headers = [] keys = [] for column in columns: if not isinstance(column, dict): return self._format_error(meta) key = column.get('key') label = column.get('name') if not isinstance(key, str) or not key.strip(): return self._format_error(meta) if not isinstance(label, str) or not label.strip(): return self._format_error(meta) key = key.strip() definition = allowed_columns.get(key) if not isinstance(definition, tuple) or not definition: return self._format_error(meta) keys.append(key) header = {'label': definition[0]} if len(definition) > 1: header['description'] = definition[1] headers.append(header) rows = [] for record in records: if not isinstance(record, dict): return self._format_error(meta) rows.append([ '' if record.get(key) is None else record.get(key, '') for key in keys ]) content = { 'summary': self._safe_text(data.get('summary')), 'headers': headers, 'rows': rows, } tips = self._safe_tips(data.get('tips')) if tips: content['tips'] = tips pagination = self._build_pagination(raw_meta) if pagination: content['pagination'] = pagination return self._success_result(content, self._render_table(content), meta) def _present_options(self, data, raw_meta, meta): records = data.get('records') if not isinstance(records, list): return self._format_error(meta) rows = [] for record in records: if not isinstance(record, dict): return self._format_error(meta) rows.append([ record.get('value', ''), self._safe_text(record.get('label')), self._safe_text(record.get('code')), ]) content = { 'headers': [ {'label': '可传值'}, {'label': '显示名称'}, {'label': '业务编码'}, ], 'rows': rows, } pagination = self._build_pagination(raw_meta) if pagination: content['pagination'] = pagination return self._success_result(content, self._render_table(content), meta) def _present_export(self, data, meta): url = data.get('file_url') if not isinstance(url, str) or not url.strip(): return self._format_error(meta) content = { 'message': '文件已生成', 'files': [{'label': '导出文件', 'url': url.strip()}], } text = '文件已生成\n- 导出文件: {0}'.format(url.strip()) return self._success_result(content, text, meta) def _business_error(self, tool_name, code, raw_message, meta): if code == 'MCP_1401': label = self._find_field_label(tool_name, raw_message) message = self._parameter_message(label) retryable = False else: message = self.ERROR_MESSAGES.get( code, '工具调用失败，请稍后重试', ) retryable = code not in ('MCP_1101', 'MCP_1102', 'MCP_1201', 'MCP_1202', 'MCP_1301') return self._error_result(code, message, retryable, meta) def _find_field_label(self, tool_name, raw_message): message = str(raw_message or '') fields = self.FIELD_LABELS.get(tool_name, {}) for field in sorted(fields, key=len, reverse=True): if field in message: return fields[field] return '' @staticmethod def _parameter_message(label): return '{0}参数不正确'.format(label) if label else '工具参数不正确，请检查后重试' @staticmethod def _build_meta(raw_meta): if not isinstance(raw_meta, dict): return {} request_id = raw_meta.get('request_id') if not isinstance(request_id, str) or not request_id.strip(): return {} return {'request_id': request_id.strip()} @staticmethod def _build_pagination(raw_meta): if not isinstance(raw_meta, dict): return {} return { key: raw_meta[key] for key in ('page', 'limit', 'has_more', 'total') if key in raw_meta } @staticmethod def _safe_text(value): if value is None: return '' return str(value).strip() @classmethod def _safe_tips(cls, value): if not isinstance(value, list): return [] return [cls._safe_text(item) for item in value if cls._safe_text(item)] @classmethod def _render_table(cls, content): headers = content.get('headers') or [] rows = content.get('rows') or [] lines = [] summary = cls._safe_text(content.get('summary')) if summary: lines.append(summary) lines.append('表头共 {0} 列：'.format(len(headers))) for index, header in enumerate(headers, start=1): lines.append('{0}. {1}'.format(index, header['label'])) for index, row in enumerate(rows, start=1): lines.append('记录 {0}：'.format(index)) for header, value in zip(headers, row): if isinstance(value, (dict, list)): value = json.dumps(value, ensure_ascii=False) lines.append('- {0}: {1}'.format(header['label'], value)) tips = content.get('tips') or [] if tips: lines.append('提示：{0}'.format('；'.join(tips))) return '\n'.join(lines) @staticmethod def _success_result(content, text, meta): return { 'structured_content': content, 'text': text, 'is_error': False, 'meta': meta, } @staticmethod def _error_result(code, message, retryable, meta=None): return { 'structured_content': { 'code': code, 'message': message, 'retryable': retryable, }, 'text': '{0}: {1}'.format(code, message), 'is_error': True, 'meta': meta or {}, } @classmethod def _format_error(cls, meta=None): return cls._error_result( 'MCP_9001', '工具返回格式异常', True, meta, )