mcp/tools/export_payable_cost_list.py

from .query_payable_cost_list import QueryPayableCostListTool


class ExportPayableCostListTool:
    name = 'export_payable_cost_list'
    route_path = '/mcp/tools/exportPayableCostList'

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

    def metadata(self):
        query_schema = QueryPayableCostListTool().metadata()['input_schema']
        properties = dict(query_schema['properties'])
        properties.pop('page')
        properties.pop('limit')
        return {
            'name': self.name,
            'description': (
                '只提交一个应付费用列表单页签的异步导出任务，不在本次调用内等待或轮询。'
                '成功后返回task_ref、状态和建议等待时间，稍后单独调用query_export_task查询。'
                '筛选值必须与query_payable_cost_list及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, request_id='rq_export_payable_cost_list',
    ):
        if self.api_client is None:
            raise RuntimeError('api client is required for export_payable_cost_list')
        payload = QueryPayableCostListTool._filters(
            QueryPayableCostListTool._integer(business_type, '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['business_type'] = business_type
        if business_type not in QueryPayableCostListTool.BUSINESS_TYPES:
            raise ValueError('business_type is invalid')
        return self.api_client.call_tool(self.name, self.route_path, payload, request_id)