from typing import Literal from datetime import date class QueryDestinationTrailerListTool: name = 'query_destination_trailer_list' route_path = '/mcp/tools/queryDestinationTrailerList' STAGES = ( '待安排拖车', '待提柜', '待还柜(海)', '已还柜(海)', '已提货(空)', 'DO单制作', ) DO_FORBIDDEN = ( 'providers_id', 'warehouse_id', 'is_direct_send', 'bill_status', 'cabinet_type', 'shipping_method', 'eta_start', 'eta_end', 'pickup_time_start', 'pickup_time_end', 'container_return_time_start', 'container_return_time_end', ) def __init__(self, api_client=None): self.api_client = api_client def metadata(self): number_array = { 'type': 'array', 'minItems': 1, 'maxItems': 200, 'items': { 'type': 'string', 'minLength': 1, 'maxLength': 100, 'pattern': '.*\\S.*', }, } date_field = { 'type': 'string', 'format': 'date', 'pattern': '^\\d{4}-\\d{2}-\\d{2}$', } properties = { 'trailer_stage': { 'type': 'string', 'enum': list[Literal['待安排拖车', '待提柜', '待还柜(海)', '已还柜(海)', '已提货(空)', 'DO单制作']](self.STAGES), 'description': ( '海外提柜类型。必须先调用list_destination_trailer_filter_options并选择' '“海外提柜类型”取得六个中文值之一；用户未明确时必须先询问，不得默认某一类' '或六类一起查，也不得改走排舱列表。' ), }, 'bl_numbers': dict( number_array, description='提单号数组。仅当用户明确说提单号时使用；不得放入柜号或其他号码。', ), 'container_codes': dict( number_array, description='柜号数组。仅当用户明确说柜号时使用；不得放入提单号或其他号码。', ), 'providers_id': { 'type': 'integer', 'minimum': 1, 'description': '拖车行筛选值，必须先通过筛选项工具取得。', }, 'warehouse_id': { 'type': 'integer', 'minimum': 1, 'description': '海外仓筛选值，必须先通过筛选项工具取得；只命中非直送。', }, 'is_direct_send': { 'type': 'integer', 'enum': [0, 1], 'description': '直送柜筛选值，先通过筛选项工具取得。', }, 'bill_status': { 'type': 'integer', 'enum': [10, 20, 30, 35, 40], 'description': '应付费用状态筛选值，先通过筛选项工具取得。', }, 'cabinet_type': { 'type': 'integer', 'enum': [1, 2, 3, 4], 'description': '整拼类型筛选值，先通过筛选项工具取得。', }, 'shipping_method': { 'type': 'integer', 'enum': [1, 2, 3], 'description': '运输方式筛选值，先通过筛选项工具取得。', }, 'eta_start': dict(date_field), 'eta_end': dict(date_field), 'pickup_time_start': dict(date_field), 'pickup_time_end': dict(date_field), 'container_return_time_start': dict(date_field), 'container_return_time_end': dict(date_field), 'page': {'type': 'integer', 'minimum': 1, 'maximum': 100, 'default': 1}, 'limit': {'type': 'integer', 'minimum': 1, 'maximum': 100, 'default': 20}, } return { 'name': self.name, 'description': ( '使用场景：用户要看海外提柜进度或DO单制作列表时，先确认六个类型之一即可查询当前页签；' '提单号、柜号、日期和其他筛选均为可选。传了号码则类型必须明确，' '号码类型不明确时必须先询问用户，' '不得根据号码格式猜测，不得跨字段或跨工具试查，也不得把海外提柜当成排舱列表调用' 'query_outbound_list。日期若传入必须成对且最多31个目的港当地日历日。' '禁止使用：安排拖车、确认还柜、港后数据/应付/入仓单/POD/' '装柜顺序导出，以及混查订单号、拖车单号、排舱单号或SO。参数只用于工具内部调用；' '最终回答只能展示中文业务名称；描述筛选条件时不得展示筛选字段的英文参数名。' ), 'input_schema': { 'type': 'object', 'properties': properties, 'required': ['trailer_stage'], 'additionalProperties': False, }, } def call( self, trailer_stage, bl_numbers=None, container_codes=None, providers_id=None, warehouse_id=None, is_direct_send=None, bill_status=None, cabinet_type=None, shipping_method=None, eta_start=None, eta_end=None, pickup_time_start=None, pickup_time_end=None, container_return_time_start=None, container_return_time_end=None, page=1, limit=20, request_id='rq_query_destination_trailer_list', ): if self.api_client is None: raise RuntimeError( 'api client is required for query_destination_trailer_list' ) stage = str(trailer_stage or '').strip() if stage not in self.STAGES: raise ValueError('trailer_stage is invalid') payload = {'trailer_stage': stage} total = 0 for field, values in ( ('bl_numbers', bl_numbers), ('container_codes', container_codes), ): if values is None: continue cleaned = self._number_list(values, field) payload[field] = cleaned total += len(values) if total > 200: raise ValueError('at most 200 numbers are allowed') for start_name, start_value, end_value in ( ('eta_start', eta_start, eta_end), ('pickup_time_start', pickup_time_start, pickup_time_end), ( 'container_return_time_start', container_return_time_start, container_return_time_end, ), ): end_name = start_name.replace('_start', '_end') start = self._optional_date(start_value, start_name) end = self._optional_date(end_value, end_name) if bool(start) != bool(end): raise ValueError(start_name + ' range is invalid') if start: if (date.fromisoformat(end) - date.fromisoformat(start)).days > 30: raise ValueError(start_name + ' range is invalid') if end < start: raise ValueError(start_name + ' range is invalid') payload[start_name] = start payload[end_name] = end for field, value, allowed in ( ('providers_id', providers_id, None), ('warehouse_id', warehouse_id, None), ('is_direct_send', is_direct_send, (0, 1)), ('bill_status', bill_status, (10, 20, 30, 35, 40)), ('cabinet_type', cabinet_type, (1, 2, 3, 4)), ('shipping_method', shipping_method, (1, 2, 3)), ): if value is None: continue checked = self._integer(value, field, 0 if field == 'is_direct_send' else 1) if allowed is not None and checked not in allowed: raise ValueError(field + ' is invalid') payload[field] = checked if stage == 'DO单制作': for field in self.DO_FORBIDDEN: if field in payload: raise ValueError('DO stage does not accept extra filters') 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) return result @staticmethod def _optional_date(value, field): if value is None or value == '': return '' if not isinstance(value, str): raise ValueError(field + ' is invalid') try: parsed = date.fromisoformat(value) except ValueError: raise ValueError(field + ' is invalid') if parsed.isoformat() != value: raise ValueError(field + ' is invalid') return value @staticmethod def _integer(value, field, minimum=None): if isinstance(value, bool) or not isinstance(value, int): raise ValueError(field + ' is invalid') if minimum is not None and value < minimum: raise ValueError(field + ' is invalid') return value @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