mcp/tools/query_order_exact.py

class QueryOrderExactTool:
    name = 'query_order_exact'
    route_path = '/mcp/tools/queryOrderExact'

    STRING_FILTERS = (
        'order_number',
        'reference_number',
        'tracking_number',
        'outbound_number',
        'container_code',
        'so_number',
        'shipment_id',
        'receiver_country',
        'inbound_date_start',
        'inbound_date_end',
        'outbound_date_start',
        'outbound_date_end',
    )
    ARRAY_FILTERS = ('product_ids', 'customer_ids', 'warehouse_ids')
    NUMBER_ARRAY_FILTERS = (
        'order_numbers',
        'reference_numbers',
        'tracking_numbers',
        'outbound_numbers',
        'container_codes',
        'so_numbers',
    )
    NUMBER_FILTER_MAP = {
        'order_numbers': 'order_number',
        'reference_numbers': 'reference_number',
        'tracking_numbers': 'tracking_number',
        'outbound_numbers': 'outbound_number',
        'container_codes': 'container_code',
        'so_numbers': 'so_number',
    }
    SCALAR_ID_FILTERS = ('sales_id', 'department_id')
    MAX_NUMBER_FILTER_COUNT = 200
    FIELD_GUIDANCE = {
        'order_number': (
            '系统订单号，用户明确说“订单号”且只有一个值时使用；'
            '不是客户参考号或快递单号。',
            ['USC26070917207'],
        ),
        'order_numbers': (
            '多个系统订单号的批量精准查询；用户明确给出多个订单号时使用。',
            [['USC26070917207', 'USC26070917208']],
        ),
        'reference_number': (
            '客户参考号，用户明确说“客户参考号”且只有一个值时使用；'
            '不是系统订单号。',
            ['REF-20260713-001'],
        ),
        'reference_numbers': (
            '多个客户参考号的批量精准查询；用户明确给出多个客户参考号时使用。',
            [['REF-001', 'REF-002']],
        ),
        'tracking_number': (
            '快递单号、物流跟踪号或承运商跟踪号码；用户明确说“快递单号”、'
            '“物流跟踪号”或“承运商跟踪号”且只有一个值时使用；'
            '不是系统订单号。',
            ['1Z999AA10123456784'],
        ),
        'tracking_numbers': (
            '多个快递单号或承运商跟踪号码的批量精准查询；'
            '用户明确给出多个此类号码时使用。',
            [['TRACK-001', 'TRACK-002']],
        ),
        'outbound_number': (
            '排舱单号，用户明确说“排舱单号”且只有一个值时使用；'
            '不是海外仓出库单号。',
            ['PC20210331070002001'],
        ),
        'outbound_numbers': (
            '多个排舱单号的批量精准查询；用户明确给出多个排舱单号时使用。',
            [['PC20210331070002001', 'PC20210331070002002']],
        ),
        'container_code': (
            '柜号或集装箱号，用户明确说“柜号”或“集装箱号”且只有一个值时使用。',
            ['MSCU1234567'],
        ),
        'container_codes': (
            '多个柜号或集装箱号的批量精准查询；用户明确给出多个此类号码时使用。',
            [['MSCU1234567', 'TGHU7654321']],
        ),
        'so_number': (
            'SO号，即 Shipping Order 编号，格式不固定；'
            '用户明确说“SO号”且只有一个值时使用。',
            ['97964454'],
        ),
        'so_numbers': (
            '多个格式不固定的 SO 号或 Shipping Order 编号的批量精准查询；'
            '用户明确给出多个 SO 号时使用。',
            [['97964454', 'OOLU12345678']],
        ),
        'shipment_id': (
            'Shipment ID，用户明确说“Shipment ID”且只有一个值时使用。',
            ['FBA123456789'],
        ),
        'receiver_country': (
            '收货国家代码；先调用 list_order_filter_options 的 country 类型'
            '取得可用值。',
            ['US'],
        ),
        'product_ids': (
            '物流产品 ID 数组；必须先调用 list_order_filter_options 的 '
            'product 类型取得 ID。',
            [[101, 102]],
        ),
        'customer_ids': (
            '客户 ID 数组；必须先调用 list_order_filter_options 的 customer '
            '类型取得 ID。',
            [[201, 202]],
        ),
        'sales_id': (
            '销售人员 ID；必须先调用 list_order_filter_options 的 sales 类型'
            '取得 ID。',
            [301],
        ),
        'warehouse_ids': (
            '仓库 ID 数组；必须先调用 list_order_filter_options 的 warehouse '
            '类型取得 ID，-1 表示客户仓。',
            [[401, -1]],
        ),
        'department_id': (
            '事业部 ID；必须先调用 list_order_filter_options 的 department '
            '类型取得 ID。',
            [501],
        ),
        'inbound_date_start': (
            '入库日期范围开始日期，格式 YYYY-MM-DD。',
            ['2026-07-01'],
        ),
        'inbound_date_end': (
            '入库日期范围结束日期，格式 YYYY-MM-DD。',
            ['2026-07-13'],
        ),
        'outbound_date_start': (
            '出库日期范围开始日期，格式 YYYY-MM-DD。',
            ['2026-07-01'],
        ),
        'outbound_date_end': (
            '出库日期范围结束日期，格式 YYYY-MM-DD。',
            ['2026-07-13'],
        ),
        'page': ('结果页码，从 1 开始。', [1]),
        'limit': ('每页记录数，范围 1 到 100。', [20]),
    }

    def __init__(self, api_client=None):
        self.api_client = api_client

    def metadata(self):
        properties = {
            field: {'type': 'string'}
            for field in self.STRING_FILTERS
        }
        for field in self.ARRAY_FILTERS:
            properties[field] = {
                'type': 'array',
                'items': {'type': 'integer'},
            }
        for field in self.NUMBER_ARRAY_FILTERS:
            properties[field] = {
                'type': 'array',
                'items': {'type': 'string'},
            }
        for field in self.SCALAR_ID_FILTERS:
            properties[field] = {'type': 'integer', 'minimum': 1}
        properties.update({
            'page': {'type': 'integer', 'minimum': 1, 'maximum': 100},
            'limit': {'type': 'integer', 'minimum': 1, 'maximum': 100},
        })
        for field, guidance in self.FIELD_GUIDANCE.items():
            properties[field]['description'] = guidance[0]
            properties[field]['examples'] = guidance[1]

        return {
            'name': self.name,
            'description': (
                '使用场景：用户明确要查看订单资料或订单列表，并且已经明确每个号码的'
                '业务类型时，按指定字段精准查询订单；即使使用排舱单号、柜号或SO号筛选，'
                '返回目标仍是订单资料。'
                '订单号、客户参考号、快递单号、排舱单号、柜号和 SO 号必须'
                '使用各自对应字段。'
                '所有单号格式均为开放格式，只能根据用户明确说出的业务类型'
                '选择字段，不得根据号码的前缀、长度、字符组合或示例猜测。'
                '不得根据号码格式猜测。'
                '号码类型不明确时必须先询问用户；单号类型不明确时先询问用户，'
                '确认前不得调用。'
                '一个值使用单数字段，多个同类值使用复数字段数组；'
                '同一字段使用 IN，不同字段使用 AND。'
                '禁止使用：用户要看排舱列表、排舱详情、物流轨迹、报关资料文件或导出'
                '结果时不得使用本工具。不得跨字段或跨工具试查；查询无结果时不得改用'
                '其他字段重试。'
                '产品、客户、销售、仓库、事业部和国家条件先调用 '
                'list_order_filter_options。'
                '参数名仅用于工具调用；向用户回答时只能使用对应的中文业务名称，'
                '不得展示内部参数名。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': properties,
            },
        }

    def call(
        self,
        order_number='',
        reference_number='',
        tracking_number='',
        outbound_number='',
        container_code='',
        so_number='',
        shipment_id='',
        receiver_country='',
        product_ids=None,
        customer_ids=None,
        sales_id=None,
        warehouse_ids=None,
        department_id=None,
        inbound_date_start='',
        inbound_date_end='',
        outbound_date_start='',
        outbound_date_end='',
        page=1,
        limit=20,
        request_id='rq_query_order_exact',
        order_numbers=None,
        reference_numbers=None,
        tracking_numbers=None,
        outbound_numbers=None,
        container_codes=None,
        so_numbers=None,
    ):
        if self.api_client is None:
            raise RuntimeError('api client is required for query_order_exact')

        values = locals()
        payload = {}
        for field in self.STRING_FILTERS:
            value = str(values[field] or '').strip()
            if value:
                payload[field] = value

        number_count = 0
        for field in self.NUMBER_ARRAY_FILTERS:
            value = self._normalize_string_list(values[field], field)
            if value:
                payload[field] = value
            singular = self.NUMBER_FILTER_MAP[field]
            merged = list(value)
            singular_value = payload.get(singular)
            if singular_value and singular_value not in merged:
                merged.insert(0, singular_value)
            number_count += len(merged)
        if number_count > self.MAX_NUMBER_FILTER_COUNT:
            raise ValueError('number filters must not exceed 200 items')

        for field in self.ARRAY_FILTERS:
            allow_customer_warehouse = field == 'warehouse_ids'
            value = self._normalize_int_list(
                values[field],
                allow_customer_warehouse=allow_customer_warehouse,
            )
            if value:
                payload[field] = value

        for field in self.SCALAR_ID_FILTERS:
            value = values[field]
            if value is None or value == '':
                continue
            value = int(value)
            if value <= 0:
                raise ValueError('{0} must be greater than 0'.format(field))
            payload[field] = value

        if not payload:
            raise ValueError('at least one exact order filter is required')

        payload['page'] = max(1, min(100, int(page)))
        payload['limit'] = max(1, min(100, int(limit)))
        return self.api_client.call_tool(
            self.name,
            self.route_path,
            payload,
            request_id,
        )

    @staticmethod
    def _normalize_int_list(value, allow_customer_warehouse=False):
        if value is None or value == '':
            return []
        if isinstance(value, str):
            value = [item.strip() for item in value.split(',') if item.strip()]

        result = []
        for item in value:
            item = int(item)
            if item <= 0 and not (allow_customer_warehouse and item == -1):
                continue
            if item not in result:
                result.append(item)
        return result

    @staticmethod
    def _normalize_string_list(value, field):
        if value is None or value == '':
            return []
        if not isinstance(value, (list, tuple)):
            raise ValueError('{0} must be an array'.format(field))

        result = []
        for item in value:
            if not isinstance(item, str):
                raise ValueError('{0} items must be strings'.format(field))
            item = item.strip()
            if not item or len(item) > 100:
                raise ValueError('{0} contains an invalid number'.format(field))
            if item not in result:
                result.append(item)
        return result