mcp/tools/query_order_abnormal_list.py

from typing import Literal


class QueryOrderAbnormalListTool:
    name = 'query_order_abnormal_list'
    route_path = '/mcp/tools/queryOrderAbnormalList'
    STATUSES = (
        '待处理', '放行待审核', '包裹放行', '退件待审核', '退件驳回', '已退件',
    )

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

    def metadata(self):
        return {
            'name': self.name,
            'description': (
                '使用场景：用户要看头程问题件列表时，先调用'
                'list_order_abnormal_filter_options取得六个中文问题件状态之一即可查询当前页签；'
                '订单号可选。对照后台admin/Order/abnormal，一行一条问题件记录。'
                '号码类型不明确时必须先询问用户，'
                '不得根据号码格式猜测，不得跨字段或跨工具试查，也不得改走query_order或作废审核。'
                '禁止使用：关闭异常、已读、重新取号、编辑、导出，以及混查参考号、跟踪号、柜号或SO。'
                '参数只用于工具内部调用；最终回答只能展示中文业务名称；'
                '描述筛选条件时不得展示筛选字段的英文参数名。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': {
                    'abnormal_status': {
                        'type': 'string',
                        'enum': list[Literal[
                            '待处理', '放行待审核', '包裹放行',
                            '退件待审核', '退件驳回', '已退件',
                        ]](self.STATUSES),
                        'description': (
                            '问题件状态。必须先调用list_order_abnormal_filter_options并选择'
                            '“问题件状态”取得六个中文页签之一；用户未明确时必须先询问，'
                            '不得默认待处理或六类一起查。'
                        ),
                    },
                    'order_numbers': {
                        'type': 'array',
                        'minItems': 1,
                        'maxItems': 200,
                        'items': {
                            'type': 'string',
                            'minLength': 1,
                            'maxLength': 100,
                            'pattern': '.*\\S.*',
                        },
                        'description': (
                            '订单号数组。仅当用户明确说订单号时使用；不得放入参考号、跟踪号、柜号或SO。'
                        ),
                    },
                    'page': {
                        'type': 'integer',
                        'minimum': 1,
                        'maximum': 100,
                        'default': 1,
                    },
                    'limit': {
                        'type': 'integer',
                        'minimum': 1,
                        'maximum': 100,
                        'default': 20,
                    },
                },
                'required': ['abnormal_status'],
                'additionalProperties': False,
            },
        }

    def call(
        self,
        abnormal_status,
        order_numbers=None,
        page=1,
        limit=20,
        request_id='rq_query_order_abnormal_list',
    ):
        if self.api_client is None:
            raise RuntimeError(
                'api client is required for query_order_abnormal_list'
            )
        status = str(abnormal_status or '').strip()
        if status not in self.STATUSES:
            raise ValueError('abnormal_status is invalid')
        payload = {'abnormal_status': status}
        if order_numbers is not None:
            payload['order_numbers'] = self._number_list(
                order_numbers, 'order_numbers'
            )
        payload['page'] = self._bounded_integer(page, 'page')
        payload['limit'] = self._bounded_integer(limit, 'limit')
        return self.api_client.call_tool(
            self.name, self.route_path, payload, request_id,
        )

    @staticmethod
    def _number_list(values, field):
        if not isinstance(values, list) or not values:
            raise ValueError(field + ' is invalid')
        result = []
        for value in values:
            if not isinstance(value, str):
                raise ValueError(field + ' is invalid')
            value = value.strip()
            if not value or len(value) > 100:
                raise ValueError(field + ' is invalid')
            if value not in result:
                result.append(value)
        if len(result) > 200:
            raise ValueError('at most 200 numbers are allowed')
        return result

    @staticmethod
    def _bounded_integer(value, field):
        if isinstance(value, bool) or not isinstance(value, int):
            raise ValueError(field + ' is invalid')
        if value < 1 or value > 100:
            raise ValueError(field + ' is invalid')
        return value