mcp/services/output_presenter.py

import json
import logging
import math
from urllib.parse import urlparse

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',
        'query_customer_list',
    ))
    PAYMENT_FOLLOWUP_TOOLS = frozenset(('query_customer_payment_followup',))
    PAYMENT_DETAIL_TOOLS = frozenset((
        'query_customer_unverified_bill_details',
    ))
    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',
        'list_customer_filter_options',
    ))
    EXPORT_TOOLS = frozenset((
        'export_pending_outbound_orders',
        'export_out_of_province_port_data',
    ))
    TASK_TOOLS = frozenset(('query_export_task',))
    SAFE_TOOLS = (
        TABLE_TOOLS | DETAIL_TOOLS | ORDER_DETAIL_TOOLS | PAYMENT_FOLLOWUP_TOOLS
        | PAYMENT_DETAIL_TOOLS
        | OPTION_TOOLS | EXPORT_TOOLS | TASK_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': ('到货回复',),
        },
        'query_customer_list': {
            'customer_name': ('客户名称',),
            'customer_code': ('客户代码',),
            'create_time': ('开户时间',),
            'business_type': ('业务类型',),
            'customer_attribute': ('客户属性',),
            'customer_source': ('客户来源',),
            'first_inbound_date': ('首次成交时间',),
            'last_inbound_date': ('最后一次走货时间',),
            'active_status': ('活跃状态',),
            'contract_status': ('合同状态',),
            'contract_validity': ('合同有效期',),
            'credit_limit': ('信用额度',),
            'currency_code': ('结算币种',),
            'billing_modes': ('结算模式（分业务类型）',),
            'sales_name': ('商务经理',),
            'merchandiser_name': ('客户经理',),
            'department_name': ('事业部',),
        },
    }

    PAYMENT_FOLLOWUP_COLUMNS = {
        'customer_name': '客户名称',
        'settlement_currency': '结算币种',
        'billed_unverified_amount': '已出账未核销金额',
        'unbilled_amount': '未出账金额',
        'overdue_unpaid_amount': '逾期未回款金额',
        'billed_unverified_amount_cny': '已出账未核销金额（人民币）',
        'unbilled_amount_cny': '未出账金额（人民币）',
        'overdue_unpaid_amount_cny': '逾期未回款金额（人民币）',
        'unverified_receivable_monthly_summary': '未回款月份汇总',
        'receipt_unverified_amount': '收款单未核销金额',
        'current_balance': '当前余额',
        'credit_limit': '信用额度',
        'bad_debt_total': '坏账合计',
        'contract_status': '合同状态',
    }
    PAYMENT_FOLLOWUP_DETAIL_COLUMNS = {
        'bill_month': '账单月份',
        'bill_no': '账单号',
        'business_type': '业务类型',
        'settlement_mode': '结算模式',
        'unverified_amount': '未核销金额',
        'customer_receivable_date': '客户应收款日期',
    }
    PAYMENT_FOLLOWUP_MONTHLY_COLUMNS = {
        'receivable_month': '应收月份',
        'unverified_amount': '未核销金额',
        'is_overdue': '是否逾期',
    }

    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': '每页数量',
        },
        'query_customer_list': {
            'customer_id': '客户名称',
            'department_id': '事业部',
            'sales_id': '商务经理',
            'merchandiser_id': '客户经理',
            'page': '页码',
            'limit': '每页数量',
        },
        'query_customer_payment_followup': {
            'customer_id': '客户名称',
            'department_id': '事业部',
            'sales_id': '商务经理',
            'merchandiser_id': '客户经理',
            'has_unverified_receivable_only': '只看有应收未核销费用的客户',
            'page': '页码',
            'limit': '每页数量',
        },
        'query_customer_unverified_bill_details': {
            'customer_id': '客户名称',
            'page': '页码',
            'limit': '每页数量',
        },
        'list_customer_filter_options': {
            'filter_type': '筛选项类型',
            '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': '资料类型',
        },
        'query_export_task': {
            'task_ref': '导出任务引用',
        },
    }

    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.PAYMENT_FOLLOWUP_TOOLS:
            return self._present_customer_payment_followup(
                data, tool_result.get('meta'), meta
            )
        if tool_name in self.PAYMENT_DETAIL_TOOLS:
            return self._present_customer_unverified_bill_details(
                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:
            if tool_name == 'list_customer_filter_options':
                return self._present_customer_options(
                    data, tool_result.get('meta'), meta
                )
            return self._present_options(data, tool_result.get('meta'), meta)
        if tool_name in self.EXPORT_TOOLS:
            return self._present_export_submission(data, meta)
        return self._present_export_task(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)

        if tool_name == 'query_customer_list':
            expected = list(allowed_columns.keys())
            actual = [
                column.get('key') if isinstance(column, dict) else None
                for column in columns
            ]
            if actual != expected or self._customer_pagination(raw_meta) is None:
                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)
            if tool_name == 'query_customer_list' and set(record) != set(keys):
                return self._format_error(meta)
            row = []
            for key in keys:
                value = '' if record.get(key) is None else record.get(key, '')
                if tool_name == 'query_customer_list' and key == 'billing_modes':
                    value = self._customer_billing_modes(value)
                    if value is None:
                        return self._format_error(meta)
                elif tool_name == 'query_customer_list' and isinstance(value, (dict, list)):
                    return self._format_error(meta)
                row.append(value)
            rows.append(row)

        content = {
            'summary': self._safe_text(data.get('summary')),
            'headers': headers,
            'rows': rows,
        }
        if tool_name == 'query_customer_list':
            content['display_rules'] = {
                'mode': 'complete',
                'allow_summary': False,
                'allow_omit_records': False,
                'allow_omit_empty_fields': False,
                'allow_rename_fields': False,
                'preserve_record_order': True,
                'required_field_count': len(headers),
                'returned_record_count': len(rows),
                'required_value_count': len(headers) * len(rows),
                'instruction': '最终回复必须逐条展示全部记录及全部17个字段，不得摘要、省略或改写',
            }
        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,
                require_complete=tool_name == 'query_customer_list',
            ),
            meta,
        )

    def _present_customer_payment_followup(self, data, raw_meta, meta):
        if set(data) != {'columns', 'records'}:
            return self._format_error(meta)
        columns = data.get('columns')
        records = data.get('records')
        pagination = self._customer_pagination(raw_meta)
        expected_keys = list(self.PAYMENT_FOLLOWUP_COLUMNS)
        if not isinstance(columns, list) or not isinstance(records, list):
            return self._format_error(meta)
        actual_keys = []
        for column in columns:
            if (
                not isinstance(column, dict)
                or set(column) != {'key', 'name'}
                or not isinstance(column.get('key'), str)
                or not isinstance(column.get('name'), str)
                or not column.get('name').strip()
            ):
                return self._format_error(meta)
            actual_keys.append(column['key'])
        if actual_keys != expected_keys or pagination is None:
            return self._format_error(meta)

        amount_keys = {
            'billed_unverified_amount', 'unbilled_amount',
            'overdue_unpaid_amount', 'billed_unverified_amount_cny',
            'unbilled_amount_cny', 'overdue_unpaid_amount_cny',
            'receipt_unverified_amount', 'current_balance',
            'credit_limit', 'bad_debt_total',
        }
        text_keys = {'customer_name', 'settlement_currency', 'contract_status'}
        rows = []
        for record in records:
            if not isinstance(record, dict) or set(record) != set(expected_keys):
                return self._format_error(meta)
            row = []
            for key in expected_keys:
                value = record[key]
                if key in amount_keys:
                    if (
                        isinstance(value, bool)
                        or not isinstance(value, (int, float))
                        or not math.isfinite(float(value))
                    ):
                        return self._format_error(meta)
                elif key in text_keys:
                    if not isinstance(value, str):
                        return self._format_error(meta)
                else:
                    value = self._payment_followup_monthly_summaries(value)
                    if value is None:
                        return self._format_error(meta)
                row.append(value)
            rows.append(row)

        headers = [
            {'label': self.PAYMENT_FOLLOWUP_COLUMNS[key]}
            for key in expected_keys
        ]
        content = {
            'headers': headers,
            'rows': rows,
            'pagination': pagination,
            'display_rules': {
                'mode': 'complete',
                'allow_summary': False,
                'allow_omit_records': False,
                'allow_omit_empty_fields': False,
                'allow_rename_fields': False,
                'preserve_record_order': True,
                'required_field_count': len(headers),
                'returned_record_count': len(rows),
                'required_value_count': len(headers) * len(rows),
                'instruction': '最终回复必须逐条展示全部客户、全部字段和全部未回款月份汇总',
            },
        }
        return self._success_result(
            content,
            self._render_table(content, require_complete=True),
            meta,
        )

    def _payment_followup_monthly_summaries(self, summaries):
        if not isinstance(summaries, list):
            return None
        expected = set(self.PAYMENT_FOLLOWUP_MONTHLY_COLUMNS)
        translated = []
        for summary in summaries:
            if not isinstance(summary, dict) or set(summary) != expected:
                return None
            amount = summary.get('unverified_amount')
            if (
                isinstance(amount, bool)
                or not isinstance(amount, (int, float))
                or not math.isfinite(float(amount))
            ):
                return None
            if (
                not isinstance(summary.get('receivable_month'), str)
                or not isinstance(summary.get('is_overdue'), bool)
            ):
                return None
            item = {}
            for key, label in self.PAYMENT_FOLLOWUP_MONTHLY_COLUMNS.items():
                value = summary[key]
                item[label] = value
            translated.append(item)
        return translated

    def _present_customer_unverified_bill_details(self, data, raw_meta, meta):
        if set(data) != {'columns', 'records'}:
            return self._format_error(meta)
        columns = data.get('columns')
        records = data.get('records')
        pagination = self._customer_pagination(raw_meta)
        expected_keys = list(self.PAYMENT_FOLLOWUP_DETAIL_COLUMNS)
        if not isinstance(columns, list) or not isinstance(records, list):
            return self._format_error(meta)
        actual_keys = []
        for column in columns:
            if (
                not isinstance(column, dict)
                or set(column) != {'key', 'name'}
                or not isinstance(column.get('key'), str)
                or not isinstance(column.get('name'), str)
                or not column.get('name').strip()
            ):
                return self._format_error(meta)
            actual_keys.append(column['key'])
        if actual_keys != expected_keys or pagination is None:
            return self._format_error(meta)

        translated = self._payment_followup_details(records)
        if translated is None:
            return self._format_error(meta)
        headers = [
            {'label': self.PAYMENT_FOLLOWUP_DETAIL_COLUMNS[key]}
            for key in expected_keys
        ]
        rows = [[item[header['label']] for header in headers] for item in translated]
        content = {'headers': headers, 'rows': rows, 'pagination': pagination}
        return self._success_result(content, self._render_table(content), meta)

    def _payment_followup_details(self, details):
        expected = set(self.PAYMENT_FOLLOWUP_DETAIL_COLUMNS)
        translated = []
        for detail in details:
            if not isinstance(detail, dict) or set(detail) != expected:
                return None
            amount = detail.get('unverified_amount')
            if (
                isinstance(amount, bool)
                or not isinstance(amount, (int, float))
                or not math.isfinite(float(amount))
            ):
                return None
            item = {}
            for key, label in self.PAYMENT_FOLLOWUP_DETAIL_COLUMNS.items():
                value = detail[key]
                if key != 'unverified_amount' and not isinstance(value, str):
                    return None
                item[label] = value
            translated.append(item)
        return translated

    @staticmethod
    def _customer_billing_modes(value):
        if not isinstance(value, list):
            return None
        result = []
        business_types = set()
        for item in value:
            if not isinstance(item, dict) or set(item) != {
                'business_type', 'settlement_mode',
            }:
                return None
            business_type = item.get('business_type')
            settlement_mode = item.get('settlement_mode')
            if (
                not isinstance(business_type, str) or not business_type.strip()
                or not isinstance(settlement_mode, str) or not settlement_mode.strip()
                or business_type in business_types
            ):
                return None
            business_types.add(business_type)
            result.append({
                '业务类型': business_type.strip(),
                '结算模式': settlement_mode.strip(),
            })
        return result

    def _present_customer_options(self, data, raw_meta, meta):
        if set(data) != {'records'}:
            return self._format_error(meta)
        records = data.get('records')
        pagination = self._customer_pagination(raw_meta)
        if not isinstance(records, list) or pagination is None:
            return self._format_error(meta)
        rows = []
        for record in records:
            if not isinstance(record, dict) or set(record) != {'value', 'label', 'code'}:
                return self._format_error(meta)
            value = record.get('value')
            label = record.get('label')
            code = record.get('code')
            if (
                isinstance(value, bool) or not isinstance(value, int) or value < 1
                or not isinstance(label, str) or not label.strip()
                or not isinstance(code, str)
            ):
                return self._format_error(meta)
            rows.append([value, label.strip(), code.strip()])
        content = {
            'headers': [
                {'label': '可传值'}, {'label': '显示名称'}, {'label': '业务编码'},
            ],
            'rows': rows,
            'pagination': pagination,
        }
        return self._success_result(content, self._render_table(content), meta)

    @staticmethod
    def _customer_pagination(raw_meta):
        if not isinstance(raw_meta, dict):
            return None
        required = {'page', 'limit', 'has_more'}
        if (
            not required.issubset(raw_meta)
            or not set(raw_meta).issubset(required | {'request_id'})
        ):
            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 or page > 100
            or isinstance(limit, bool) or not isinstance(limit, int)
            or limit < 1 or limit > 100
            or not isinstance(has_more, bool)
        ):
            return None
        return {'page': page, 'limit': limit, 'has_more': has_more}

    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_submission(self, data, meta):
        expected = {'task_ref', 'status', 'retry_after_seconds'}
        if set(data) != expected:
            return self._format_error(meta)
        task_ref = data.get('task_ref')
        retry_after = data.get('retry_after_seconds')
        if (
            not isinstance(task_ref, str) or not task_ref.strip()
            or data.get('status') != 'queued'
            or isinstance(retry_after, bool)
            or not isinstance(retry_after, int)
            or retry_after <= 0
        ):
            return self._format_error(meta)
        task = {
            'task_ref': task_ref.strip(),
            'status': 'queued',
            'retry_after_seconds': retry_after,
        }
        content = {
            'message': '导出任务已提交',
            'task': task,
        }
        text = (
            '导出任务已提交\n'
            '- 任务引用: {0}\n'
            '- 建议 {1} 秒后单独查询任务状态'
        ).format(task['task_ref'], retry_after)
        return self._success_result(content, text, meta)

    def _present_export_task(self, data, meta):
        task_ref = data.get('task_ref')
        status = data.get('status')
        if (
            not isinstance(task_ref, str)
            or not task_ref.strip()
            or status not in ('queued', 'running', 'completed', 'failed')
        ):
            return self._format_error(meta)

        task = {'task_ref': task_ref.strip(), 'status': status}
        if status in ('queued', 'running'):
            if set(data) != {'task_ref', 'status', 'retry_after_seconds'}:
                return self._format_error(meta)
            retry_after = data.get('retry_after_seconds')
            if (
                isinstance(retry_after, bool)
                or not isinstance(retry_after, int)
                or retry_after <= 0
            ):
                return self._format_error(meta)
            task['retry_after_seconds'] = retry_after
            label = '等待中' if status == 'queued' else '生成中'
            content = {'message': '导出任务{0}'.format(label), 'task': task}
            text = (
                '导出任务{0}\n- 任务引用: {1}\n- 建议 {2} 秒后再次查询'
            ).format(label, task['task_ref'], retry_after)
            return self._success_result(content, text, meta)

        if status == 'failed':
            if set(data) != {'task_ref', 'status'}:
                return self._format_error(meta)
            content = {'message': '导出任务失败，请重新提交', 'task': task}
            text = '导出任务失败，请重新提交'
            return self._success_result(content, text, meta)

        if set(data) != {'task_ref', 'status', 'files'}:
            return self._format_error(meta)
        files = data.get('files')
        if not isinstance(files, list) or not files:
            return self._format_error(meta)
        safe_files = []
        for item in files:
            if not isinstance(item, dict) or set(item) != {'label', 'url'}:
                return self._format_error(meta)
            label = item.get('label')
            url = item.get('url')
            if (
                not isinstance(label, str) or not label.strip()
                or not self._valid_http_url(url)
            ):
                return self._format_error(meta)
            safe_files.append({'label': label.strip(), 'url': url.strip()})
        content = {
            'message': '文件已生成',
            'task': task,
            'files': safe_files,
        }
        lines = ['文件已生成']
        for item in safe_files:
            lines.append('- {0}: {1}'.format(item['label'], item['url']))
        return self._success_result(content, '\n'.join(lines), meta)

    def _valid_http_url(self, value):
        if not isinstance(value, str) or not value.strip():
            return False
        if value != value.strip() or '\\' in value or any(
            character.isspace() or ord(character) < 32 or ord(character) == 127
            for character in value
        ):
            return False
        try:
            parsed = urlparse(value)
            port = parsed.port
            return (
                parsed.scheme in ('http', 'https')
                and bool(parsed.hostname)
                and parsed.username is None
                and parsed.password is None
                and (port is None or 0 < port <= 65535)
            )
        except ValueError:
            return False

    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, require_complete=False):
        headers = content.get('headers') or []
        rows = content.get('rows') or []
        lines = []
        if require_complete:
            lines.append('完整客户数据，禁止摘要、合并、隐藏字段或省略空字段。')
            lines.append(
                '本页记录数：{0}；每条字段数：{1}；应展示字段值总数：{2}。'.format(
                    len(rows), len(headers), len(rows) * len(headers)
                )
            )
        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)))
        if require_complete:
            lines.append(
                '展示完整性校验：已提供{0}条客户的全部{1}个字段。'.format(
                    len(rows), len(headers)
                )
            )
        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,
        )