mcp/tools/export_out_of_province_port_data.py

class ExportOutOfProvincePortDataTool:
    name = 'export_out_of_province_port_data'
    route_path = '/mcp/tools/exportOutOfProvincePortData'

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

    def metadata(self):
        number_items = {
            'type': 'string',
            'minLength': 1,
            'maxLength': 100,
        }
        return {
            'name': self.name,
            'description': (
                '导出后台排舱单列表中的省外进港资料，并返回可下载文件 URL。'
                '只能在用户明确要求导出省外进港资料，并明确提供排舱单号、'
                '柜号、提单号或SO号中的一种，以及宁波、上海或美森资料类型时调用。'
                '用户没有明确号码类型时必须先询问：'
                '“请确认使用哪种单号导出：排舱单号、柜号、提单号还是SO号？”'
                '确认前不得调用。不得根据号码格式猜测，也不得在查询失败后'
                '切换号码类型重试。一次只能选择一种号码类型。'
                '一次只能选择一种资料类型；用户要求多个类型时应分别调用。'
                '参数名仅用于工具调用；向用户回答时只能使用中文业务名称，'
                '不得展示内部参数名。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': {
                    'outbound_numbers': {
                        'type': 'array',
                        'items': dict(number_items),
                        'minItems': 1,
                        'maxItems': 100,
                        'uniqueItems': True,
                        'description': (
                            '要导出的排舱单号数组。排舱单号是后台排舱单列表中的'
                            '业务单号，不是系统订单号、客户参考号、快递单号、柜号、'
                            'SO号或Shipment ID。只能使用用户明确提供并确认类型为'
                            '排舱单号的值，不得根据号码格式猜测。'
                        ),
                        'examples': [[
                            'PC202607140001',
                            'PC202607140002',
                        ]],
                    },
                    'container_codes': {
                        'type': 'array',
                        'items': dict(number_items),
                        'minItems': 1,
                        'maxItems': 100,
                        'uniqueItems': True,
                        'description': (
                            '要导出的柜号或集装箱号数组。仅当用户明确说明号码类型'
                            '为“柜号”或“集装箱号”时使用，不得放入排舱单号或提单号。'
                        ),
                        'examples': [[
                            'MSCU1234567',
                            'TGHU7654321',
                        ]],
                    },
                    'bl_numbers': {
                        'type': 'array',
                        'items': dict(number_items),
                        'minItems': 1,
                        'maxItems': 100,
                        'uniqueItems': True,
                        'description': (
                            '要导出的提单号数组，对应后台排舱单列表中的“提单号”。'
                            '不是系统提单号；仅当用户明确说明号码类型为“提单号”时使用。'
                        ),
                        'examples': [[
                            'BL202607150001',
                            'BL202607150002',
                        ]],
                    },
                    'so_numbers': {
                        'type': 'array',
                        'items': dict(number_items),
                        'minItems': 1,
                        'maxItems': 100,
                        'uniqueItems': True,
                        'description': (
                            '要导出的SO号数组，对应后台排舱单列表中的“SO号”，'
                            '数据字段为fms_booking_detail.so_number。仅当用户'
                            '明确说明号码类型为“SO号”时使用。'
                        ),
                        'examples': [[
                            'SO202607160001',
                            'SO202607160002',
                        ]],
                    },
                    'file_type': {
                        'type': 'string',
                        'enum': ['NB', 'SH', 'MS'],
                        'description': (
                            '进港资料类型：NB=宁波进港资料，SH=上海进港资料，'
                            'MS=美森进港资料。必须根据用户明确选择传值；'
                            '用户未说明时先询问，不得默认选择。'
                        ),
                        'examples': ['NB'],
                    },
                },
                'required': ['file_type'],
                'oneOf': [
                    {'required': ['outbound_numbers']},
                    {'required': ['container_codes']},
                    {'required': ['bl_numbers']},
                    {'required': ['so_numbers']},
                ],
                'additionalProperties': False,
            },
        }

    def call(
        self,
        outbound_numbers=None,
        file_type=None,
        request_id='rq_export_out_of_province_port_data',
        container_codes=None,
        bl_numbers=None,
        so_numbers=None,
    ):
        if self.api_client is None:
            raise RuntimeError(
                'api client is required for export_out_of_province_port_data'
            )

        number_fields = {
            'outbound_numbers': outbound_numbers,
            'container_codes': container_codes,
            'bl_numbers': bl_numbers,
            'so_numbers': so_numbers,
        }
        provided = [
            field for field, values in number_fields.items()
            if values is not None
        ]
        if len(provided) != 1:
            raise ValueError('provide exactly one number type')
        number_field = provided[0]
        numbers = self._normalize_numbers(number_fields[number_field], number_field)
        normalized_type = str(file_type or '').strip().upper()
        if normalized_type not in ('NB', 'SH', 'MS'):
            raise ValueError('file_type must be NB, SH, or MS')

        return self.api_client.call_tool(
            self.name,
            self.route_path,
            {
                number_field: numbers,
                'file_type': normalized_type,
            },
            request_id,
        )

    @staticmethod
    def _normalize_numbers(values, field):
        if not isinstance(values, list):
            raise ValueError('{0} must be an array'.format(field))
        if len(values) > 100:
            raise ValueError('{0} must contain 1 to 100 values'.format(field))

        numbers = []
        for value in values:
            if not isinstance(value, str):
                raise ValueError('{0} must contain strings'.format(field))
            value = value.strip()
            if not value:
                continue
            if len(value) > 100:
                raise ValueError('{0} contains a number that is too long'.format(field))
            if value not in numbers:
                numbers.append(value)

        if not numbers or len(numbers) > 100:
            raise ValueError('{0} must contain 1 to 100 values'.format(field))
        return numbers