mcp/tools/query_outbound_list.py

class QueryOutboundListTool:
    name = 'query_outbound_list'
    route_path = '/mcp/tools/queryOutboundList'
    MAX_NUMBERS = 100
    STATUS_VALUES = (20, 30, 60, 70, 80, 90, 100, 110, 120)

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

    def metadata(self):
        number_array = {
            'type': 'array',
            'items': {'type': 'string', 'minLength': 1, 'maxLength': 100},
            'minItems': 1,
            'maxItems': self.MAX_NUMBERS,
        }
        mode_array = {
            'type': 'array',
            'items': {'type': 'integer', 'enum': [1, 2]},
            'minItems': 1,
            'maxItems': 2,
            'uniqueItems': True,
        }
        properties = {
            'outbound_numbers': dict(
                number_array,
                description='排舱单号数组。仅当用户明确说排舱单号时使用；不得放入其他类型号码。',
            ),
            'order_numbers': dict(
                number_array,
                description='订单号数组。仅当用户明确说订单号时使用；不得放入其他类型号码。',
            ),
            'container_codes': dict(
                number_array,
                description='柜号或集装箱号数组。仅当用户明确说柜号时使用；不得放入其他类型号码。',
            ),
            'so_numbers': dict(
                number_array,
                description='SO号数组。仅当用户明确说SO号时使用；不得放入其他类型号码。',
            ),
            'bl_numbers': dict(
                number_array,
                description='提单号数组。仅当用户明确说提单号时使用；不得放入其他类型号码。',
            ),
            'outbound_status': {
                'type': 'integer', 'enum': list(self.STATUS_VALUES),
                'description': (
                    '排舱阶段筛选值。先调用list_outbound_filter_options并选择“排舱阶段”'
                    '取得后台中文阶段和可传值；可选阶段为国内排舱、国内拖车、出库装柜、'
                    '出口报关、干线运输、目的国清关、海外拖车、海外仓入库、完成。'
                    '用户未明确时必须先询问，不得默认选择。'
                ),
            },
            'shipping_method': {
                'type': 'integer', 'enum': [1, 2, 3],
                'description': (
                    '运输方式筛选值。明确指定时先通过统一筛选选项工具取得；未指定时查询全部，'
                    '调用工具时不传该筛选。'
                ),
            },
            'warehouse_id': {
                'oneOf': [
                    {'type': 'integer', 'const': -1},
                    {'type': 'integer', 'minimum': 1},
                ],
                'description': (
                    '集货仓库筛选值。必须先调用list_outbound_filter_options并选择'
                    '“集货仓库”取得；禁止猜测内部值。'
                ),
            },
            'is_direct_send': dict(
                {'type': 'integer', 'enum': [0, 1]},
                description='是否直送柜筛选值，先通过统一筛选选项工具取得。',
            ),
            'trailer_types': dict(mode_array, description='拖车方式筛选值，先通过统一筛选选项工具取得。'),
            'declaration_types': dict(mode_array, description='报关方式筛选值，先通过统一筛选选项工具取得。'),
            'clearance_types': dict(mode_array, description='清关方式筛选值，先通过统一筛选选项工具取得。'),
            'page': {'type': 'integer', 'minimum': 1, 'maximum': 100, 'default': 1},
            'limit': {'type': 'integer', 'minimum': 1, 'maximum': 100, 'default': 20},
        }
        for field, label in (
            ('closing_time_start', '截关开始时间'),
            ('closing_time_end', '截关结束时间'),
            ('est_loading_time_start', '预计装柜开始时间'),
            ('est_loading_time_end', '预计装柜结束时间'),
            ('create_date_start', '创建开始时间'),
            ('create_date_end', '创建结束时间'),
            ('loading_time_start', '出库开始时间'),
            ('loading_time_end', '出库结束时间'),
        ):
            properties[field] = {
                'type': 'string', 'maxLength': 19,
                'description': label + '，格式与后台筛选一致。',
            }

        return {
            'name': self.name,
            'description': (
                '使用场景：用户明确要查看或筛选排舱列表时使用，返回后台排舱单列表权限'
                '范围内固定的19个字段。运输方式未指定时查询全部；排舱阶段必须由用户明确，未明确时'
                '必须先询问，不得默认选择。号码类型不明确时必须先'
                '询问用户，确认是排舱单号、订单号、柜号、SO号还是提单号后再调用。'
                '只有用户明确分别提供多个不同类型条件时才按AND组合。阶段、运输方式、仓库、'
                '直送柜、拖车、报关和清关筛选均先调用list_outbound_filter_options取得可传值。'
                '禁止使用：用户要查看订单资料、排舱详情、物流轨迹、报关资料文件或导出'
                '结果时不得使用本工具。不得根据号码格式猜测，不得跨字段或跨工具试查，'
                '查询无结果后不得换号码字段重试。'
                '公司、员工和权限范围由当前设备会话确定，调用方不得覆盖。'
                '参数只用于工具内部调用；描述筛选条件时只能使用中文业务名称，不得展示筛选'
                '字段的英文参数名，不得展示内部参数名，不得使用“筛选字段=内部值”形式，'
                '不得展示筛选项内部数字'
                '代码，不得使用“API固定值”“接口参数”“默认参数”等技术表述。查询结果中的'
                '件数、重量、体积、日期等正常业务数值必须保留。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': properties,
                'required': ['outbound_status'],
                'additionalProperties': False,
            },
        }

    def call(
        self,
        outbound_numbers=None,
        order_numbers=None,
        container_codes=None,
        so_numbers=None,
        bl_numbers=None,
        outbound_status=None,
        shipping_method=None,
        warehouse_id=None,
        is_direct_send=None,
        trailer_types=None,
        declaration_types=None,
        clearance_types=None,
        closing_time_start='',
        closing_time_end='',
        est_loading_time_start='',
        est_loading_time_end='',
        create_date_start='',
        create_date_end='',
        loading_time_start='',
        loading_time_end='',
        page=1,
        limit=20,
        request_id='rq_query_outbound_list',
    ):
        if self.api_client is None:
            raise RuntimeError('api client is required for query_outbound_list')

        payload = {
            'outbound_status': self._enum_integer(
                outbound_status, 'outbound_status', self.STATUS_VALUES
            ),
            'page': self._bounded_integer(page, 'page', 100),
            'limit': self._bounded_integer(limit, 'limit', 100),
        }
        if shipping_method is not None:
            payload['shipping_method'] = self._enum_integer(
                shipping_method, 'shipping_method', (1, 2, 3)
            )
        for field, values in (
            ('outbound_numbers', outbound_numbers),
            ('order_numbers', order_numbers),
            ('container_codes', container_codes),
            ('so_numbers', so_numbers),
            ('bl_numbers', bl_numbers),
        ):
            if values is not None:
                payload[field] = self._string_array(values, field)
        for field, values in (
            ('trailer_types', trailer_types),
            ('declaration_types', declaration_types),
            ('clearance_types', clearance_types),
        ):
            if values is not None:
                payload[field] = self._mode_array(values, field)
        if warehouse_id is not None:
            payload['warehouse_id'] = self._warehouse_id(warehouse_id)
        if is_direct_send is not None:
            payload['is_direct_send'] = self._enum_integer(
                is_direct_send, 'is_direct_send', (0, 1)
            )
        for field, value in (
            ('closing_time_start', closing_time_start),
            ('closing_time_end', closing_time_end),
            ('est_loading_time_start', est_loading_time_start),
            ('est_loading_time_end', est_loading_time_end),
            ('create_date_start', create_date_start),
            ('create_date_end', create_date_end),
            ('loading_time_start', loading_time_start),
            ('loading_time_end', loading_time_end),
        ):
            value = str(value or '').strip()
            if len(value) > 19:
                raise ValueError('{0} is too long'.format(field))
            if value:
                payload[field] = value

        return self.api_client.call_tool(
            self.name, self.route_path, payload, request_id
        )

    @classmethod
    def _string_array(cls, values, field):
        if not isinstance(values, list) or len(values) > cls.MAX_NUMBERS:
            raise ValueError('{0} must contain 1 to 100 values'.format(field))
        result = []
        for value in values:
            if not isinstance(value, str):
                raise ValueError('{0} must contain strings'.format(field))
            value = value.strip()
            if not value or len(value) > 100:
                raise ValueError('{0} contains an invalid value'.format(field))
            if value not in result:
                result.append(value)
        if not result:
            raise ValueError('{0} must contain 1 to 100 values'.format(field))
        return result

    @classmethod
    def _mode_array(cls, values, field):
        if not isinstance(values, list) or not values or len(values) > 2:
            raise ValueError('{0} must contain 1 or 2'.format(field))
        result = []
        for value in values:
            value = cls._enum_integer(value, field, (1, 2))
            if value not in result:
                result.append(value)
        return result

    @staticmethod
    def _bounded_integer(value, field, maximum):
        if isinstance(value, bool):
            raise ValueError('{0} is invalid'.format(field))
        try:
            value = int(value)
        except (TypeError, ValueError):
            raise ValueError('{0} is invalid'.format(field))
        if value < 1 or value > maximum:
            raise ValueError('{0} is invalid'.format(field))
        return value

    @staticmethod
    def _enum_integer(value, field, allowed):
        if isinstance(value, bool):
            raise ValueError('{0} is invalid'.format(field))
        try:
            value = int(value)
        except (TypeError, ValueError):
            raise ValueError('{0} is invalid'.format(field))
        if value not in allowed:
            raise ValueError('{0} is invalid'.format(field))
        return value

    @staticmethod
    def _warehouse_id(value):
        if isinstance(value, bool):
            raise ValueError('warehouse_id is invalid')
        try:
            value = int(value)
        except (TypeError, ValueError):
            raise ValueError('warehouse_id is invalid')
        if value != -1 and value < 1:
            raise ValueError('warehouse_id is invalid')
        return value