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