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之后,或用户只要外链时,' '用本工具保存一份新的头程文档。必须带添加文档类型和至少一条已确认种类的录入单号。' '添加文档类型的显示名是「文档类型名称 - 文件维度」。' '用户只说了类型名称且不止一条对得上时,先列出完整显示名请用户选择,选定前不要保存。' '只有一条对得上时使用该条 value。' '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