import json import logging from constants import DEVICE_INVALID_MESSAGE logger = logging.getLogger(__name__) # 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', 'query_outbound_list', )) DETAIL_TOOLS = frozenset(('query_outbound_detail',)) ORDER_DETAIL_TOOLS = frozenset(('query_order_detail',)) OPTION_TOOLS = frozenset(( 'list_order_filter_options', 'list_outbound_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 | DETAIL_TOOLS | ORDER_DETAIL_TOOLS | OPTION_TOOLS | EXPORT_TOOLS ) ORDER_DETAIL_SECTIONS = { 'overview': '订单概览', 'packages': '箱单信息', 'package_items': '箱单商品', 'dw_auth': 'DW授权信息', 'attachments': '附件信息', 'inbound': '入库信息', 'checks': '查验信息', 'tracks': '订单轨迹', 'operation_logs': '操作日志', 'cost_logs': '应收与结算日志', 'delivery': '派送信息', 'all': '全部', } ORDER_DETAIL_FIELDS = { 'overview': { 'order_info': { 'order_number': '订单号', 'reference_number': '客户参考号', 'customer_name': '客户名称', 'departure_country': '起运国', 'destination_country': '目的国', 'package_method': '包装类型', 'product_name': '物流产品', 'forecast_pwv': '预报件重体', 'inbound_pwv': '入库件重体', 'receivable_charge_weight': '应收计费重/体积', 'settlement_charge_weight': '结算计费重/体积', 'declaration_type': '报关类型', 'pre_recording': '是否需要预录单', 'idcard_name': '企业/个人姓名', 'idcard_number': '社会信用代码/身份证号', 'outbound_number': '排舱单号', 'container_mode': '直送柜/拼柜', 'track_map_available': '是否可查看轨迹地图', 'insurance': '购买保险', 'delivery_way': '派送方式', 'return_shipping_surcharge': '退运附加', 'work_order_count': '工单数量', 'bill_remark': '账单备注', 'remark': '订单备注', }, 'freight_info': { 'estimated_inbound_time': '预计入库时间', 'cargo_type': '货物类型', 'pickup_way': '交货方式', 'warehouse': '交货仓库', 'pickup_address': '提货地址', 'pickup_contact': '提货联系人', 'pickup_phone': '提货联系电话', 'delivery_type': '派送类型', 'delivery_address': '派送地址', 'delivery_remark': '派送备注', 'oversea_warehouse': '海外仓', }, 'package_summary': { 'box_count': '箱数', 'total_weight': '重量', 'weight_unit': '重量单位', 'total_volume': '体积', 'sku_count': 'SKU数量', 'clearance_total_price': '清关总价', 'clearance_currency': '清关币种', 'purchase_total_price': '采购总价', 'purchase_currency': '采购币种', }, 'trader_info': { 'exporter': '出口商', 'importer': '进口商', 'coo_required': '是否做产地证', }, }, 'packages': { 'shipment_id': 'SHIPMENT ID', 'reference_id': 'REFERENCE ID', 'sku_summary': 'SKU', 'product_name_summary': '商品名', 'box_quantity': '箱数', 'box_length_cm': '单箱长(CM)', 'box_width_cm': '单箱宽(CM)', 'box_height_cm': '单箱高(CM)', 'gross_weight_kg': '单箱毛重量(KG)', 'net_weight_kg': '单箱净重量(KG)', 'is_over_length': '长度是否超限', 'is_over_width': '宽度是否超限', 'is_over_height': '高度是否超限', 'is_over_weight': '重量是否超限', }, 'package_items': { 'shipment_id': 'SHIPMENT ID', 'reference_id': 'REFERENCE ID', 'box_specification': '箱规', 'box_specification_unit': '箱规单位', 'package_gross_weight_kg': '箱单单箱毛重(KG)', 'package_net_weight_kg': '箱单单箱净重(KG)', 'inspection_result': '查验结果', 'exception_reason': '异常原因', 'inspection_photo_count': '查验图片数量', 'sku': 'SKU', 'units_per_box': '单箱个数', 'declaration_type': '报关类型', 'box_number': '箱号', 'item_gross_weight_kg': '商品单箱毛重量(KG)', 'item_net_weight_kg': '商品单箱净重量(KG)', 'chinese_name': '中文品名', 'english_name': '英文名称', 'brand_type': '品牌类型', 'brand': '品牌', 'model': '型号', 'material_cn': '材质（CN）', 'material_en': '材质（EN）', 'purpose_cn': '用途', 'goods_attributes': '商品属性', 'declaration_hs_code': '报关编码', 'clearance_hs_code': '清关编码', 'material_ratio': '材质占比', 'clearance_unit_price': '清关单价', 'clearance_currency': '清关币种', 'purchase_unit_price': '采购单价', 'purchase_currency': '采购币种', 'product_image_count': '商品图片数量', 'remark': '备注', }, 'dw_auth': { 'fba_id': 'FBAID', 'dw_start': 'DW开始时间', 'dw_end': 'DW结束时间', 'dw_display': 'DW', 'authorization_status': '客户授权状态', }, 'attachments': { 'category': '附件分类', 'file_name': '文件名称', 'file_extension': '文件类型', 'is_image': '是否图片', 'preview_url': '预览链接', 'download_url': '下载链接', 'cost_name': '收费项目', }, 'inbound_summary': { 'inbound_status': '入库状态', 'inbound_time': '入库时间', 'inbound_operator': '操作人', 'warehouse': '仓库', 'abnormal_reason': '异常原因', 'inbound_photo_count': '入库图片数量', 'forecast_pwv_summary': '预报件重体', 'inbound_pwv_summary': '入库件重体', }, 'inbound': { 'has_label': '是否贴标', 'shipment_id': 'SHIPMENT ID', 'inbound_quantity': '入库件数', 'measurement_photo_count': '测量图片数量', 'box_specification': '入库箱规', 'weight_mode': None, 'weight': None, 'weight_unit': '重量单位', 'is_over_length': '长度是否超限', 'is_over_width': '宽度是否超限', 'is_over_height': '高度是否超限', 'is_over_weight': '重量是否超限', 'operated_at': '操作时间', }, 'checks': { 'shipment_id': '货件编号', 'sku': 'SKU', 'box_number': '箱号', 'box_quantity': '箱数', 'box_specification': '箱规', 'inspection_result': '查验结果', 'inspection_photo_count': '查验图片数量', 'remark': '备注', 'inspector': '查验人', 'inspected_at': '查验时间', }, 'tracks': { 'track_type': '轨迹类型', 'status': '轨迹节点', 'location': '轨迹地点', 'occurred_at': '时间', 'tracking_number': '跟踪号', 'shipment_id': 'SHIPMENT ID', 'content': '轨迹内容', }, 'operation_logs': { 'operated_at': '操作时间', 'content': '操作内容', 'operator': '操作人', }, 'cost_logs': { 'operated_at': '操作时间', 'content': '操作内容', 'operator': '操作人', }, 'delivery_summary': { 'master_tracking_number': '主跟踪号', 'sub_order_count': '子单数量', }, 'delivery': {'sub_tracking_number': '子跟踪号'}, } ORDER_DETAIL_STATUS_FIELDS = { 'stage': '阶段', 'label': '节点', 'state': '状态', 'occurred_at': '时间', 'time_kind': '时间类型', } ORDER_DETAIL_STATES = { 'completed': '已完成', 'in_progress': '进行中', 'pending': '待完成', 'hidden': '不展示', } ORDER_DETAIL_TIME_KINDS = { 'actual': '实际', 'estimated': '预计', 'empty': '暂无', } 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': ('文件链接',), }, 'query_outbound_list': { 'outbound_number': ('排舱单号',), 'direct_send': ('直送柜',), 'remark': ('备注',), 'cargo_type': ('超长超重',), 'warehouse_name': ('集货仓库',), 'status': ('状态',), 'so_number': ('SO号',), 'container_code': ('柜号',), 'seal_number': ('封条号',), 'container_type': ('柜型',), 'shipping_method': ('运输方式',), 'total_volume': ('体积(CBM)',), 'total_weight': ('重量(KG)',), 'ship_schedule': ('船期',), 'closing_time': ('截关时间',), 'est_loading_time': ('预计装柜时间',), 'operator': ('操作人',), 'operation_modes': ('拖报清方式',), 'operation_time': ('操作时间',), }, 'query_outbound_detail': { 'order_number': ('订单号',), 'cargo_type': ('超长超重',), 'customs_files': ('报关资料',), 'reference_number': ('客户参考号',), 'customer_name': ('客户名称',), 'declaration_type': ('报关方式',), 'merge_declare_number': ('合并报关单号',), 'order_remark': ('订单备注',), 'bl_number': ('提单号',), 'container_code': ('柜号',), 'customer_service': ('客户经理',), 'est_inbound_date': ('预计入库时间',), 'inbound_date': ('实际入库时间',), 'status': ('状态',), 'pieces': ('预报件数 / 入库件数',), 'weight': ('预报实重(KG) / 入库实重(KG)',), 'volume': ('预报体积(m³) / 入库体积(m³)',), 'pickup_place': ('交货地',), 'product_name': ('物流产品',), 'goods_name': ('品名',), 'closing_time': ('截关时间',), 'clearance_remark': ('报关备注',), 'import_clearance_remark': ('清关备注',), 'delivery_address': ('派送地址',), 'delivery_type': ('派送类型',), 'delivery_way': ('派送方式',), 'channel': ('派送渠道',), 'country_name': ('目的国',), 'importer_name': ('进口商',), 'vat_type': ('清关税号',), 'packing_type': ('货物类型',), 'is_abnormal': ('是否问题件',), 'package_method': ('包装类型',), 'order_reply': ('到货回复',), }, } OUTBOUND_DETAIL_SUMMARY = { 'bl_number': '提单号', 'container_code': '柜号', 'container_type': '柜型', 'total_volume': '总体积', 'total_weight': '总重量', 'total_pieces': '总件数', 'sku': 'SKU', 'buy_declaration_count': '买单报关数量', 'general_declaration_count': '一般贸易报关数量', 'must_load': '必装单量/体积', 'backup_load': '备装单量/体积', } 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': '每页数量', }, 'query_outbound_list': { 'outbound_numbers': '排舱单号', 'order_numbers': '订单号', 'container_codes': '柜号', 'so_numbers': 'SO号', 'bl_numbers': '提单号', 'outbound_status': '排舱状态', 'shipping_method': '运输方式', 'warehouse_id': '集货仓库', 'is_direct_send': '是否直送柜', 'page': '页码', 'limit': '每页数量', }, 'query_outbound_detail': { 'outbound_number': '排舱单号', 'page': '页码', 'limit': '每页数量', }, 'query_order_detail': { 'order_number': '订单号', 'section': '详情模块', 'page': '页码', 'limit': '每页数量', }, 'list_order_filter_options': { 'filter_type': '筛选项类型', 'keyword': '关键词', 'page': '页码', 'limit': '每页数量', }, 'list_outbound_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_1103': '员工账号不可用，请联系管理员', 'MCP_1104': '身份信息已失效，请重新登录或重新生成设备配置', 'MCP_1201': '工具当前不可用', 'MCP_1202': '工具当前不可用', 'MCP_1301': '没有权限使用此工具', 'MCP_1501': '目标数据不可用', 'MCP_1502': '结果过多，请缩小查询范围', 'MCP_1601': '没有可导出的数据', 'MCP_9001': '工具调用失败，请稍后重试', } NON_RETRYABLE_CODES = frozenset(( 'MCP_1101', 'MCP_1102', 'MCP_1103', 'MCP_1104', 'MCP_1201', 'MCP_1202', 'MCP_1301', 'MCP_1401', 'MCP_1501', 'MCP_1502', 'MCP_1601', )) 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.DETAIL_TOOLS: return self._present_outbound_detail( tool_name, data, tool_result.get('meta'), meta, ) if tool_name in self.ORDER_DETAIL_TOOLS: return self._present_order_detail(data, tool_result.get('meta'), 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) def _present_order_detail(self, data, raw_meta, meta): if set(data) != {'section', 'order_number', 'payload'}: return self._format_error(meta) section = data.get('section') order_number = data.get('order_number') payload = data.get('payload') if ( section not in self.ORDER_DETAIL_SECTIONS or not isinstance(order_number, str) or not order_number.strip() or not isinstance(payload, dict) ): return self._format_error(meta) content = { '订单号': order_number.strip(), '详情模块': self.ORDER_DETAIL_SECTIONS[section], } if section == 'all': all_content = self._present_order_detail_all(payload) if all_content is None: return self._format_error(meta) content.update(all_content) elif section == 'overview': expected = { 'status_nodes', 'order_info', 'freight_info', 'package_summary', 'trader_info', } if set(payload) != expected or not isinstance(payload['status_nodes'], list): return self._format_error(meta) nodes = [] for row in payload['status_nodes']: translated = self._translate_exact(row, self.ORDER_DETAIL_STATUS_FIELDS) if translated is None: return self._format_error(meta) state = row.get('state') time_kind = row.get('time_kind') if state not in self.ORDER_DETAIL_STATES or time_kind not in self.ORDER_DETAIL_TIME_KINDS: return self._format_error(meta) translated['状态'] = self.ORDER_DETAIL_STATES[state] translated['时间类型'] = self.ORDER_DETAIL_TIME_KINDS[time_kind] nodes.append(translated) content['状态节点'] = nodes for raw_key, chinese_key in ( ('order_info', '订单信息'), ('freight_info', '货运信息'), ('package_summary', '箱单汇总'), ('trader_info', '进出口商'), ): translated = self._translate_exact( payload.get(raw_key), self.ORDER_DETAIL_FIELDS['overview'][raw_key], ) if translated is None: return self._format_error(meta) content[chinese_key] = translated elif section == 'inbound': if set(payload) != {'summary', 'records'}: return self._format_error(meta) summary = self._translate_exact( payload.get('summary'), self.ORDER_DETAIL_FIELDS['inbound_summary'] ) records = self._translate_order_detail_rows(section, payload.get('records')) if summary is None or records is None: return self._format_error(meta) content['入库概况'] = summary content['明细'] = records pagination = self._order_detail_pagination(raw_meta) if pagination is None: return self._format_error(meta) content['分页'] = pagination elif section == 'delivery': if set(payload) != {'summary', 'records'}: return self._format_error(meta) summary = self._translate_exact( payload.get('summary'), self.ORDER_DETAIL_FIELDS['delivery_summary'] ) records = self._translate_order_detail_rows(section, payload.get('records')) if summary is None or records is None: return self._format_error(meta) content['派送汇总'] = summary content['明细'] = records pagination = self._order_detail_pagination(raw_meta) if pagination is None: return self._format_error(meta) content['分页'] = pagination else: if set(payload) != {'records'}: return self._format_error(meta) records = self._translate_order_detail_rows(section, payload.get('records')) pagination = self._order_detail_pagination(raw_meta) if records is None or pagination is None: return self._format_error(meta) content['明细'] = records content['分页'] = pagination text = json.dumps(content, ensure_ascii=False, indent=2) return self._success_result(content, text, meta) def _present_order_detail_all(self, payload): expected_sections = set(self.ORDER_DETAIL_SECTIONS) - {'all'} if set(payload) != expected_sections: return None overview = payload.get('overview') if not isinstance(overview, dict) or set(overview) != { 'status_nodes', 'order_info', 'freight_info', 'package_summary', 'trader_info', } or not isinstance(overview['status_nodes'], list): return None overview_content = {'状态节点': []} for row in overview['status_nodes']: translated = self._translate_exact(row, self.ORDER_DETAIL_STATUS_FIELDS) if translated is None: return None state = row.get('state') time_kind = row.get('time_kind') if state not in self.ORDER_DETAIL_STATES or time_kind not in self.ORDER_DETAIL_TIME_KINDS: return None translated['状态'] = self.ORDER_DETAIL_STATES[state] translated['时间类型'] = self.ORDER_DETAIL_TIME_KINDS[time_kind] overview_content['状态节点'].append(translated) for raw_key, chinese_key in ( ('order_info', '订单信息'), ('freight_info', '货运信息'), ('package_summary', '箱单汇总'), ('trader_info', '进出口商'), ): translated = self._translate_exact( overview.get(raw_key), self.ORDER_DETAIL_FIELDS['overview'][raw_key] ) if translated is None: return None overview_content[chinese_key] = translated result = {'订单概览': overview_content} for section, label in self.ORDER_DETAIL_SECTIONS.items(): if section in ('overview', 'all'): continue translated = self._present_order_detail_all_section(section, payload.get(section)) if translated is None: return None result[label] = translated return result def _present_order_detail_all_section(self, section, payload): if not isinstance(payload, dict) or 'pagination' not in payload: return None pagination = self._order_detail_pagination(payload.get('pagination')) if pagination is None: return None if section == 'inbound': if set(payload) != {'summary', 'records', 'pagination'}: return None summary = self._translate_exact( payload.get('summary'), self.ORDER_DETAIL_FIELDS['inbound_summary'] ) records = self._translate_order_detail_rows(section, payload.get('records')) if summary is None or records is None: return None return {'入库概况': summary, '明细': records, '分页': pagination} if section == 'delivery': if set(payload) != {'summary', 'records', 'pagination'}: return None summary = self._translate_exact( payload.get('summary'), self.ORDER_DETAIL_FIELDS['delivery_summary'] ) records = self._translate_order_detail_rows(section, payload.get('records')) if summary is None or records is None: return None return {'派送汇总': summary, '明细': records, '分页': pagination} if set(payload) != {'records', 'pagination'}: return None records = self._translate_order_detail_rows(section, payload.get('records')) if records is None: return None return {'明细': records, '分页': pagination} def _translate_order_detail_rows(self, section, records): if not isinstance(records, list): return None mapping = self.ORDER_DETAIL_FIELDS.get(section) if not isinstance(mapping, dict): return None translated_rows = [] for row in records: translated = self._translate_exact(row, mapping) if translated is None: return None if section == 'inbound': mode = row.get('weight_mode') if mode not in ('single_box', 'total'): return None translated[ '单箱入库重量' if mode == 'single_box' else '总重量KG' ] = row.get('weight', '') translated_rows.append(translated) return translated_rows @staticmethod def _translate_exact(value, mapping): if not isinstance(value, dict) or set(value) != set(mapping): return None result = {} for key, label in mapping.items(): if label is None: continue item = value.get(key, '') if isinstance(item, (dict, list)): return None result[label] = '' if item is None else item return result @staticmethod def _order_detail_pagination(raw_meta): if not isinstance(raw_meta, dict): return None if not all(key in raw_meta for key in ('page', 'limit', 'has_more')): return None page = raw_meta.get('page') limit = raw_meta.get('limit') has_more = raw_meta.get('has_more') if isinstance(page, bool) or not isinstance(page, int) or page < 1: return None if isinstance(limit, bool) or not isinstance(limit, int) or limit < 1: return None if not isinstance(has_more, bool): return None return {'页码': page, '每页数量': limit, '是否还有更多': has_more} # 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_outbound_detail(self, tool_name, data, raw_meta, meta): summary = data.get('summary') columns = data.get('columns') records = data.get('records') if not isinstance(summary, dict): return self._format_error(meta) if not isinstance(columns, list) or not columns or not isinstance(records, list): return self._format_error(meta) if any(key not in summary for key in self.OUTBOUND_DETAIL_SUMMARY): return self._format_error(meta) allowed_columns = self.TABLE_COLUMNS.get(tool_name) headers = [] keys = [] for column in columns: if not isinstance(column, dict): return self._format_error(meta) key = column.get('key') definition = allowed_columns.get(key) if isinstance(key, str) else None if not isinstance(definition, tuple) or not definition: return self._format_error(meta) keys.append(key) headers.append({'label': definition[0]}) rows = [] for record in records: if not isinstance(record, dict): return self._format_error(meta) row = [] for key in keys: value = '' if record.get(key) is None else record.get(key, '') if key == 'customs_files': if not isinstance(value, list): return self._format_error(meta) files = [] for item in value: if not isinstance(item, dict): return self._format_error(meta) files.append({ '文件名称': item.get('file_name', ''), '文件类型': item.get('file_type', ''), '文件链接': item.get('file_url', ''), }) value = files row.append(value) rows.append(row) summary_keys = list(self.OUTBOUND_DETAIL_SUMMARY.keys()) content = { 'summary': { 'headers': [ {'label': self.OUTBOUND_DETAIL_SUMMARY[key]} for key in summary_keys ], 'row': [summary.get(key, '') for key in summary_keys], }, 'details': { 'headers': headers, 'rows': rows, }, } tips = self._safe_tips(data.get('tips')) if tips: content['details']['tips'] = tips pagination = self._build_pagination(raw_meta) if pagination: content['details']['pagination'] = pagination return self._success_result( content, self._render_outbound_detail(content), meta, ) @classmethod def _render_outbound_detail(cls, content): lines = ['排舱汇总：'] summary = content['summary'] for header, value in zip(summary['headers'], summary['row']): lines.append('- {0}: {1}'.format(header['label'], value)) lines.append('订单明细：') details = content['details'] lines.append('表头共 {0} 列：'.format(len(details['headers']))) for index, header in enumerate(details['headers'], start=1): lines.append('{0}. {1}'.format(index, header['label'])) for index, row in enumerate(details['rows'], start=1): lines.append('记录 {0}：'.format(index)) for header, value in zip(details['headers'], row): if isinstance(value, (dict, list)): value = json.dumps(value, ensure_ascii=False) lines.append('- {0}: {1}'.format(header['label'], value)) tips = details.get('tips') or [] if tips: lines.append('提示：{0}'.format('；'.join(tips))) return '\n'.join(lines) 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 not in self.ERROR_MESSAGES and code != 'MCP_1401': logger.warning( 'Unknown MCP business error code', extra={ 'request_id': meta.get('request_id', ''), 'tool_code': tool_name, 'backend_code': code, 'response_code': 'MCP_9001', 'diagnostic_reason': 'UNEXPECTED_EXCEPTION', }, ) code = 'MCP_9001' if code == 'MCP_1401': label = self._find_field_label(tool_name, raw_message) message = self._parameter_message(label) else: message = self.ERROR_MESSAGES[code] retryable = code not in self.NON_RETRYABLE_CODES 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, )