mcp/tools/query_receive_volume_list.py

from datetime import datetime


class QueryReceiveVolumeListTool:
    name = 'query_receive_volume_list'
    route_path = '/mcp/tools/queryReceiveVolumeList'

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

    def metadata(self):
        date_field = {
            'type': 'string',
            'format': 'date',
            'pattern': '^\\d{4}-\\d{2}-\\d{2}$',
        }
        id_array = {
            'type': 'array',
            'minItems': 1,
            'maxItems': 200,
            'items': {'type': 'integer', 'minimum': 1},
        }
        return {
            'name': self.name,
            'description': (
                '使用场景：用户要看收货量明细（已入库头程订单一行一单）时，'
                '必须先确认员工时区下最多31个日历日的入库时间闭区间，再调用'
                'list_receive_volume_filter_options取得客户、商务经理、客户经理、'
                '货物类型或物流产品的可传值。对照后台admin/Report/receiveForm的'
                '已入库集合，不是按客户汇总，也不是业务发生时间。'
                '禁止使用：导出、散货/整柜页脚、订单号筛选，不得改走query_order、'
                'query_order_exact或后台收货量导出。参数只用于工具内部调用；'
                '最终回答只能展示中文业务名称；描述筛选条件时不得展示筛选字段的'
                '英文参数名。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': {
                    'inbound_date_start': dict(
                        date_field,
                        description='入库时间开始日期，员工时区 YYYY-MM-DD。',
                    ),
                    'inbound_date_end': dict(
                        date_field,
                        description='入库时间结束日期，员工时区 YYYY-MM-DD，且与开始日期跨度不超过31个日历日。',
                    ),
                    'customer_ids': dict(
                        id_array,
                        description='客户ID数组。必须先调用list_receive_volume_filter_options选择「客户名称」取得value。',
                    ),
                    'sales_ids': dict(
                        id_array,
                        description='商务经理员工ID数组。必须先调用筛选项取得value。',
                    ),
                    'merchandiser_ids': dict(
                        id_array,
                        description='客户经理员工ID数组。必须先调用筛选项取得value。',
                    ),
                    'packing_types': {
                        'type': 'array',
                        'minItems': 1,
                        'maxItems': 200,
                        'items': {
                            'type': 'string',
                            'minLength': 1,
                            'maxLength': 100,
                            'pattern': '.*\\S.*',
                        },
                        'description': '货物类型名称数组。必须先调用筛选项选择「货物类型」取得value。',
                    },
                    'product_ids': dict(
                        id_array,
                        description='物流产品ID数组。必须先调用筛选项选择「物流产品」取得value。',
                    ),
                    'page': {
                        'type': 'integer',
                        'minimum': 1,
                        'maximum': 100,
                        'default': 1,
                    },
                    'limit': {
                        'type': 'integer',
                        'minimum': 1,
                        'maximum': 100,
                        'default': 20,
                    },
                },
                'required': ['inbound_date_start', 'inbound_date_end'],
                'additionalProperties': False,
            },
        }

    def call(
        self,
        inbound_date_start,
        inbound_date_end,
        customer_ids=None,
        sales_ids=None,
        merchandiser_ids=None,
        packing_types=None,
        product_ids=None,
        page=1,
        limit=20,
        request_id='rq_query_receive_volume_list',
    ):
        if self.api_client is None:
            raise RuntimeError(
                'api client is required for query_receive_volume_list'
            )
        payload = {
            'inbound_date_start': self._date(inbound_date_start, 'inbound_date_start'),
            'inbound_date_end': self._date(inbound_date_end, 'inbound_date_end'),
        }
        self._assert_window(
            payload['inbound_date_start'], payload['inbound_date_end']
        )
        if customer_ids is not None:
            payload['customer_ids'] = self._id_list(customer_ids, 'customer_ids')
        if sales_ids is not None:
            payload['sales_ids'] = self._id_list(sales_ids, 'sales_ids')
        if merchandiser_ids is not None:
            payload['merchandiser_ids'] = self._id_list(
                merchandiser_ids, 'merchandiser_ids'
            )
        if packing_types is not None:
            payload['packing_types'] = self._name_list(
                packing_types, 'packing_types'
            )
        if product_ids is not None:
            payload['product_ids'] = self._id_list(product_ids, 'product_ids')
        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 _date(value, field):
        text = str(value or '').strip()
        try:
            parsed = datetime.strptime(text, '%Y-%m-%d').date()
        except ValueError:
            raise ValueError(field + ' is invalid')
        if parsed.strftime('%Y-%m-%d') != text:
            raise ValueError(field + ' is invalid')
        return text

    @staticmethod
    def _assert_window(start, end):
        start_date = datetime.strptime(start, '%Y-%m-%d').date()
        end_date = datetime.strptime(end, '%Y-%m-%d').date()
        if end_date < start_date or (end_date - start_date).days > 30:
            raise ValueError('inbound date window is invalid')

    @staticmethod
    def _id_list(values, field):
        if not isinstance(values, list) or not values:
            raise ValueError(field + ' is invalid')
        result = []
        for value in values:
            if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
                raise ValueError(field + ' is invalid')
            if value not in result:
                result.append(value)
        if len(result) > 200:
            raise ValueError('at most 200 ids are allowed')
        return result

    @staticmethod
    def _name_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 names 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