from datetime import date class QueryPayableCostListTool: name = 'query_payable_cost_list' route_path = '/mcp/tools/queryPayableCostList' NUMBER_FIELDS = ( 'order_numbers', 'tracking_numbers', 'container_codes', 'bl_numbers', 'so_numbers', ) BUSINESS_TYPES = (1, 2, 3, 4, 5, 7) 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 = { 'business_type': {'type': 'integer', 'enum': list(self.BUSINESS_TYPES)}, 'order_numbers': dict(number_array), 'tracking_numbers': dict(number_array), 'container_codes': dict(number_array), 'bl_numbers': dict(number_array), 'so_numbers': dict(number_array), 'business_date_start': dict(date_field), 'business_date_end': dict(date_field), 'cost_date_start': dict(date_field), 'cost_date_end': dict(date_field), 'operation_date_start': dict(date_field), 'operation_date_end': dict(date_field), 'business_node_id': {'type': 'integer', 'minimum': 1}, 'provider_id': {'type': 'integer', 'minimum': 1}, 'cost_type_ids': { 'type': 'array', 'minItems': 1, 'maxItems': 200, 'items': {'type': 'integer', 'minimum': 1}, }, 'billing_status': {'type': 'integer', 'enum': [0, 1]}, 'payment_status': {'type': 'integer', 'enum': [-1, 1]}, 'verification_status': {'type': 'integer', 'enum': [0, 1, 2]}, 'document_type': {'type': 'integer', 'minimum': 0}, 'page': {'type': 'integer', 'minimum': 1, 'maximum': 100, 'default': 1}, 'limit': {'type': 'integer', 'minimum': 1, 'maximum': 100, 'default': 20}, } return { 'name': self.name, 'description': ( '按结算新版页签查询当前员工公司内的应付费用列表。五类号码字段分别精确筛选，' '不同字段与其他条件均为AND；禁止混合单号猜测、跨字段试查或使用未确认的费用项。' '不传号码时至少提供一组不超过31日的业务、费用或操作日期闭区间。' '筛选值必须先调用list_payable_cost_filter_options取得，禁止猜测内部ID。' ), 'input_schema': { 'type': 'object', 'properties': properties, 'required': ['business_type'], 'additionalProperties': False, }, } def call( self, business_type, order_numbers=None, tracking_numbers=None, container_codes=None, bl_numbers=None, so_numbers=None, business_date_start=None, business_date_end=None, cost_date_start=None, cost_date_end=None, operation_date_start=None, operation_date_end=None, business_node_id=None, provider_id=None, cost_type_ids=None, billing_status=None, payment_status=None, verification_status=None, document_type=None, page=1, limit=20, request_id='rq_query_payable_cost_list', ): if self.api_client is None: raise RuntimeError('api client is required for query_payable_cost_list') business_type = self._integer(business_type, 'business_type') if business_type not in self.BUSINESS_TYPES: raise ValueError('business_type is invalid') payload = self._filters( business_type, order_numbers, tracking_numbers, container_codes, bl_numbers, so_numbers, business_date_start, business_date_end, cost_date_start, cost_date_end, operation_date_start, operation_date_end, business_node_id, provider_id, cost_type_ids, billing_status, payment_status, verification_status, document_type, ) 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) @classmethod def _filters( cls, business_type, order_numbers, tracking_numbers, container_codes, bl_numbers, so_numbers, business_date_start, business_date_end, cost_date_start, cost_date_end, operation_date_start, operation_date_end, business_node_id, provider_id, cost_type_ids, billing_status, payment_status, verification_status, document_type, ): values = { 'order_numbers': order_numbers, 'tracking_numbers': tracking_numbers, 'container_codes': container_codes, 'bl_numbers': bl_numbers, 'so_numbers': so_numbers, } payload = {} payload['business_type'] = business_type total = 0 for field in cls.NUMBER_FIELDS: if values[field] is None: continue cleaned = cls._number_list(values[field], field) payload[field] = cleaned total += len(values[field]) if total > 200: raise ValueError('at most 200 numbers are allowed') date_pairs = ( ('business_date_start', business_date_start, business_date_end), ('cost_date_start', cost_date_start, cost_date_end), ('operation_date_start', operation_date_start, operation_date_end), ) complete_dates = False for field, start_value, end_value in date_pairs: start = cls._optional_date(start_value, field) end = cls._optional_date(end_value, field.replace('_start', '_end')) if bool(start) != bool(end): raise ValueError(field + ' range is invalid') if start: if (date.fromisoformat(end) - date.fromisoformat(start)).days > 30: raise ValueError(field + ' range is invalid') if end < start: raise ValueError(field + ' range is invalid') payload[field] = start payload[field.replace('_start', '_end')] = end complete_dates = True if total == 0 and not complete_dates: raise ValueError('number or date filters are required') for field, value, minimum in ( ('business_node_id', business_node_id, 1), ('provider_id', provider_id, 1), ('document_type', document_type, 0), ): if value is not None: payload[field] = cls._integer(value, field, minimum) if cost_type_ids is not None: if not isinstance(cost_type_ids, list) or not cost_type_ids: raise ValueError('cost_type_ids is invalid') if len(cost_type_ids) > 200: raise ValueError('cost_type_ids is invalid') payload['cost_type_ids'] = [cls._integer(item, 'cost_type_ids', 1) for item in cost_type_ids] for field, value, allowed in ( ('billing_status', billing_status, (0, 1)), ('payment_status', payment_status, (-1, 1)), ('verification_status', verification_status, (0, 1, 2)), ): if value is not None: checked = cls._integer(value, field) if checked not in allowed: raise ValueError(field + ' is invalid') payload[field] = checked return payload @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 @classmethod def _bounded_integer(cls, value, field): value = cls._integer(value, field, 1) if value > 100: raise ValueError(field + ' is invalid') return value