mcp/tools/arrange_destination_trailer.py

class ArrangeDestinationTrailerTool:
    name = 'arrange_destination_trailer'
    route_path = '/mcp/tools/arrangeDestinationTrailer'
    CHANGE_REASONS = (
        '直送爆仓被拒',
        '客户HOLD货',
        '拿不到APPT',
        '集装箱装箱不当，亚马逊无法卸货',
        '货物不符合直送标准',
    )

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

    def metadata(self):
        text = {
            'type': 'string',
            'minLength': 1,
            'maxLength': 200,
            'pattern': '.*\\S.*',
        }
        number = dict(text, maxLength=100)
        return {
            'name': self.name,
            'description': (
                '首次安排海外拖车行，对齐后台待安排拖车页的「安排拖车行」。'
                '只接受提单号或柜号，可以只给一个，也可以两个都给并且同时满足；'
                '必须正好命中当前员工能看到的一柜。不得传内部编号，不得修改已有提柜单。'
                '拖车行可以说「名称-代码」、名称或代码。海外仓只说仓库代码。'
                '港后跟进人说员工姓名；不说则保持现在的人，说「空」才清空。'
                '直送柜不说则保持现状，只能说是或否。'
                '选否时要说海外仓；联系人和电话不说则用仓库主数据。'
                '选是时不能再说海外仓、联系人或电话。'
                '本来就是直送、预选海外仓有外部仓库代码时，仍要按订单给重量和体积；'
                '重量体积不合法时整次拒绝，不会先生成拖车单。'
                '本来是直送、这次改为否时，必须说直送改拆；选是还要说页面上的改拆原因原文。'
                '预计提柜时间可选，格式 YYYY-MM-DD HH:MM:SS，按目的港当地时间。'
                '需要上传 BOL 的柜子不要调用，请到页面安排。'
                '有外部仓库代码但不要 BOL 时，可以按订单号改重量或体积。'
                '不该出现的字段不要传。失败时不会返回可选列表。'
                '成功只返回拖车单号、提单号、柜号。'
                '参数名仅用于工具调用；向用户回答时只用中文业务名称。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': {
                    'bl_number': dict(number, description='提单号。与柜号至少给一个。'),
                    'container_code': dict(number, description='柜号。与提单号至少给一个。'),
                    'trailer_provider': dict(
                        text,
                        description='拖车行。可说名称-代码、名称或代码，必须正好命中一家。',
                    ),
                    'is_direct_send': {
                        'type': 'string',
                        'enum': ['是', '否'],
                        'description': '直送柜。不传则保持这一柜现在的值。',
                    },
                    'warehouse_code': dict(number, description='海外仓仓库代码。直送为否时必填。'),
                    'warehouse_contact': dict(text, description='仓库联系人。不传则用海外仓主数据。'),
                    'warehouse_tel': dict(text, description='仓库联系电话。不传则用海外仓主数据。'),
                    'follow_up_user': dict(
                        text,
                        description='港后跟进人姓名。说「空」表示清空。不传则保持现在的人。',
                    ),
                    'est_pickup_time': {
                        'type': 'string',
                        'pattern': '^\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}$',
                        'description': '预计提柜时间，目的港当地时间。',
                    },
                    'is_direct_change': {
                        'type': 'string',
                        'enum': ['是', '否'],
                        'description': '直送改拆。仅本来是直送、这次改为否时填写。',
                    },
                    'direct_change_reason': {
                        'type': 'string',
                        'enum': list(self.CHANGE_REASONS),
                        'description': '改拆原因，必须是页面上的整句原文。',
                    },
                    'order_measures': {
                        'type': 'array',
                        'maxItems': 500,
                        'items': {
                            'type': 'object',
                            'properties': {
                                'order_number': dict(number, description='这一柜上的订单号。'),
                                'weight': {
                                    'type': 'string',
                                    'minLength': 1,
                                    'maxLength': 20,
                                    'description': '重量，千克，大于 0 小于 50000，最多三位小数。',
                                },
                                'volume': {
                                    'type': 'string',
                                    'minLength': 1,
                                    'maxLength': 20,
                                    'description': '体积，CBM，大于 0 小于 10000，最多三位小数。',
                                },
                            },
                            'required': ['order_number'],
                            'additionalProperties': False,
                        },
                    },
                },
                'required': ['trailer_provider'],
                'additionalProperties': False,
            },
        }

    def call(
        self,
        trailer_provider,
        bl_number=None,
        container_code=None,
        is_direct_send=None,
        warehouse_code=None,
        warehouse_contact=None,
        warehouse_tel=None,
        follow_up_user=None,
        est_pickup_time=None,
        is_direct_change=None,
        direct_change_reason=None,
        order_measures=None,
        request_id='rq_arrange_destination_trailer',
    ):
        if self.api_client is None:
            raise RuntimeError('api client is required for arrange_destination_trailer')
        payload = {
            'trailer_provider': self._text(trailer_provider, 'trailer_provider', 200),
        }
        if bl_number is not None:
            payload['bl_number'] = self._text(bl_number, 'bl_number', 100)
        if container_code is not None:
            payload['container_code'] = self._text(container_code, 'container_code', 100)
        if bl_number is None and container_code is None:
            raise ValueError('bl number or container code is required')
        self._choice(payload, 'is_direct_send', is_direct_send, ('是', '否'))
        self._optional(payload, 'warehouse_code', warehouse_code, 100)
        self._optional(payload, 'warehouse_contact', warehouse_contact, 200)
        self._optional(payload, 'warehouse_tel', warehouse_tel, 200)
        self._optional(payload, 'follow_up_user', follow_up_user, 200)
        if est_pickup_time is not None:
            text = self._text(est_pickup_time, 'est_pickup_time', 19)
            if len(text) != 19:
                raise ValueError('est_pickup_time is invalid')
            payload['est_pickup_time'] = text
        self._choice(payload, 'is_direct_change', is_direct_change, ('是', '否'))
        self._choice(payload, 'direct_change_reason', direct_change_reason, self.CHANGE_REASONS)
        if order_measures is not None:
            payload['order_measures'] = self._measures(order_measures)
        return self.api_client.call_tool(
            self.name, self.route_path, payload, request_id,
        )

    def _optional(self, payload, field, value, limit):
        if value is not None:
            payload[field] = self._text(value, field, limit)

    def _choice(self, payload, field, value, allowed):
        if value is None:
            return
        if value not in allowed:
            raise ValueError(field + ' is invalid')
        payload[field] = value

    @staticmethod
    def _text(value, field, limit):
        if not isinstance(value, str):
            raise ValueError(field + ' is invalid')
        text = value.strip()
        if not text or len(text) > limit:
            raise ValueError(field + ' is invalid')
        return text

    def _measures(self, rows):
        if not isinstance(rows, list) or len(rows) > 500:
            raise ValueError('order_measures is invalid')
        result = []
        for row in rows:
            if not isinstance(row, dict) or 'order_number' not in row:
                raise ValueError('order_measures is invalid')
            item = {'order_number': self._text(row['order_number'], 'order_number', 100)}
            for key in ('weight', 'volume'):
                if key in row and row[key] is not None:
                    item[key] = self._text(row[key], key, 20)
            result.append(item)
        return result