mcp/tools/export_receivable_cost_list.py

from datetime import date


class ExportReceivableCostListTool:
    name = 'export_receivable_cost_list'
    route_path = '/mcp/tools/exportReceivableCostList'
    NUMBER_FIELDS = (
        'reference_numbers',
        'tracking_numbers',
        'order_numbers',
        'bill_numbers',
    )

    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}$',
        }
        return {
            'name': self.name,
            'description': (
                '本工具只提交异步导出任务，不会在本次调用中等待文件生成，也禁止在一次调用内'
                '轮询任务状态。成功后返回任务引用(task_ref)和建议等待时间'
                '(retry_after_seconds)。稍后单独使用 query_export_task 查询任务状态或下载链接。'
                '使用场景：只有用户明确要求导出应收费用单并需要下载文件时才可调用。'
                '筛选值必须与 query_receivable_cost_list / list_receivable_cost_filter_options'
                '已确认的号码类型、日期区间和筛选项 value 完全一致；禁止根据格式猜测号码类型，'
                '禁止勾选 ID，禁止一次调用内轮询 query_export_task。'
                '参数名仅用于工具调用；向用户回答时只能使用中文业务名称，不得展示内部参数名。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': {
                    'reference_numbers': dict(number_array),
                    'tracking_numbers': dict(number_array),
                    'order_numbers': dict(number_array),
                    'bill_numbers': dict(number_array),
                    'business_date_start': dict(date_field),
                    'business_date_end': dict(date_field),
                    'customer_id': {'type': 'integer', 'minimum': 1},
                    'sub_customer_id': {'type': 'integer', 'minimum': 1},
                    'billing_status': {'type': 'integer', 'enum': [0, 1]},
                    'verification_status': {
                        'type': 'integer', 'enum': [-1, 0, 1],
                    },
                    'document_type': {'type': 'integer', 'minimum': 0},
                    'cost_type_id': {'type': 'integer', 'minimum': 1},
                },
                'required': [],
                'allOf': [
                    {
                        'anyOf': [
                            {'required': ['reference_numbers']},
                            {'required': ['tracking_numbers']},
                            {'required': ['order_numbers']},
                            {'required': ['bill_numbers']},
                            {
                                'required': [
                                    'business_date_start',
                                    'business_date_end',
                                ],
                            },
                        ],
                    },
                    {
                        'if': {'required': ['sub_customer_id']},
                        'then': {'required': ['customer_id']},
                    },
                ],
                'additionalProperties': False,
            },
        }

    def call(
        self,
        reference_numbers=None,
        tracking_numbers=None,
        order_numbers=None,
        bill_numbers=None,
        business_date_start=None,
        business_date_end=None,
        customer_id=None,
        sub_customer_id=None,
        billing_status=None,
        verification_status=None,
        document_type=None,
        cost_type_id=None,
        request_id='rq_export_receivable_cost_list',
    ):
        if self.api_client is None:
            raise RuntimeError(
                'api client is required for export_receivable_cost_list'
            )

        supplied_numbers = {}
        total_numbers = 0
        for field, values in (
            ('reference_numbers', reference_numbers),
            ('tracking_numbers', tracking_numbers),
            ('order_numbers', order_numbers),
            ('bill_numbers', bill_numbers),
        ):
            if values is None:
                continue
            if not isinstance(values, list) or not values:
                raise ValueError(field + ' must be a non-empty list')
            cleaned = []
            for value in values:
                if not isinstance(value, str):
                    raise ValueError(field + ' items must be strings')
                item = value.strip()
                if not item or len(item) > 100:
                    raise ValueError(field + ' items must be 1 to 100 chars')
                if item not in cleaned:
                    cleaned.append(item)
            supplied_numbers[field] = cleaned
            total_numbers += len(cleaned)
        if total_numbers > 200:
            raise ValueError('at most 200 numbers across all number fields')

        has_start = business_date_start is not None
        has_end = business_date_end is not None
        if has_start != has_end:
            raise ValueError('business date range requires both start and end')
        if has_start:
            start = date.fromisoformat(str(business_date_start))
            end = date.fromisoformat(str(business_date_end))
            if end < start or (end - start).days > 30:
                raise ValueError('business date range must be within 31 days')

        if not supplied_numbers and not has_start:
            raise ValueError(
                'provide number filters or a business date range'
            )
        if sub_customer_id is not None and customer_id is None:
            raise ValueError('sub_customer_id requires customer_id')

        payload = dict(supplied_numbers)
        if has_start:
            payload['business_date_start'] = str(business_date_start)
            payload['business_date_end'] = str(business_date_end)
        for field, value in (
            ('customer_id', customer_id),
            ('sub_customer_id', sub_customer_id),
            ('billing_status', billing_status),
            ('verification_status', verification_status),
            ('document_type', document_type),
            ('cost_type_id', cost_type_id),
        ):
            if value is not None:
                payload[field] = value
        return self.api_client.call_tool(
            self.name, self.route_path, payload, request_id
        )