mcp/tools/save_headhaul_document.py

class SaveHeadhaulDocumentTool:
    name = 'save_headhaul_document'
    route_path = '/mcp/tools/saveHeadhaulDocument'

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

    def metadata(self):
        return {
            'name': self.name,
            'description': (
                '使用场景：上传成功返回ticket_ref之后，或用户只要外链时，'
                '用本工具保存一份新的头程文档。必须带添加文档类型和至少一条已确认种类的录入单号。'
                'ticket_ref和外链不能同时提交。没有行内编辑；要改只能删除后重加。'
                '禁止使用：手填OSS路径、混放号码种类、把文件放进工具参数。'
                '参数只用于工具内部调用；最终回答只能展示中文业务名称。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': {
                    'document_type_id': {
                        'type': 'integer',
                        'minimum': 1,
                        'description': (
                            '要添加的文档类型。必须先调用'
                            'list_headhaul_document_filter_options并选择“添加文档类型”。'
                        ),
                    },
                    'ticket_ref': {
                        'type': 'string',
                        'minLength': 1,
                        'maxLength': 500,
                        'description': '上传成功后的一次性凭证，不能手填。',
                    },
                    'link_url': {
                        'type': 'string',
                        'minLength': 8,
                        'maxLength': 1000,
                        'description': 'http或https外链；与ticket_ref二选一。',
                    },
                    'entries': {
                        'type': 'array',
                        'minItems': 1,
                        'items': {
                            'type': 'object',
                            'properties': {
                                'entry_no_type': {
                                    'type': 'integer',
                                    'minimum': 1,
                                    'maximum': 24,
                                },
                                'numbers': {
                                    'type': 'array',
                                    'minItems': 1,
                                    'maxItems': 50,
                                    'items': {
                                        'type': 'string',
                                        'minLength': 1,
                                        'maxLength': 100,
                                    },
                                },
                            },
                            'required': ['entry_no_type', 'numbers'],
                            'additionalProperties': False,
                        },
                        'description': '录入单号。entry_no_type来自录入单号种类筛选项。',
                    },
                },
                'required': ['document_type_id', 'entries'],
                'additionalProperties': False,
            },
        }

    def call(
        self,
        document_type_id,
        entries,
        ticket_ref=None,
        link_url=None,
        request_id='rq_save_headhaul_document',
    ):
        if self.api_client is None:
            raise RuntimeError(
                'api client is required for save_headhaul_document'
            )
        if (
            isinstance(document_type_id, bool)
            or not isinstance(document_type_id, int)
            or document_type_id < 1
        ):
            raise ValueError('document_type_id is invalid')
        ticket = '' if ticket_ref is None else str(ticket_ref).strip()
        link = '' if link_url is None else str(link_url).strip()
        if (ticket != '' and link != '') or (ticket == '' and link == ''):
            raise ValueError('ticket_ref and link_url are exclusive')
        if link and not (
            link.lower().startswith('http://')
            or link.lower().startswith('https://')
        ):
            raise ValueError('link_url is invalid')
        payload = {
            'document_type_id': document_type_id,
            'entries': self._entries(entries),
        }
        if ticket:
            payload['ticket_ref'] = ticket
        if link:
            payload['link_url'] = link
        return self.api_client.call_tool(
            self.name, self.route_path, payload, request_id,
        )

    @staticmethod
    def _entries(values):
        if not isinstance(values, list) or not values:
            raise ValueError('entries is invalid')
        result = []
        for item in values:
            if not isinstance(item, dict):
                raise ValueError('entries is invalid')
            type_id = item.get('entry_no_type')
            numbers = item.get('numbers')
            if (
                isinstance(type_id, bool)
                or not isinstance(type_id, int)
                or type_id < 1
                or type_id > 24
            ):
                raise ValueError('entry_no_type is invalid')
            if not isinstance(numbers, list) or not numbers or len(numbers) > 50:
                raise ValueError('numbers is invalid')
            clean = []
            for number in numbers:
                if not isinstance(number, str):
                    raise ValueError('numbers is invalid')
                number = number.strip()
                if not number or len(number) > 100:
                    raise ValueError('numbers is invalid')
                if number not in clean:
                    clean.append(number)
            result.append({'entry_no_type': type_id, 'numbers': clean})
        return result