mcp/tools/query_export_task.py

class QueryExportTaskTool:
    name = 'query_export_task'
    route_path = '/mcp/tools/queryExportTask'

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

    def metadata(self):
        return {
            'name': self.name,
            'description': (
                '查询此前由导出工具提交的一个异步导出任务。只能使用导出工具返回的'
                '任务引用，并通过单独的新调用查询。任务仍在等待或执行时，可按返回的'
                '建议等待时间稍后再次查询；不得在同一次调用内等待或循环轮询。'
                '任务完成后返回安全下载链接。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': {
                    'task_ref': {
                        'type': 'string',
                        'minLength': 1,
                        'maxLength': 512,
                        'description': '导出工具返回的任务引用。',
                    },
                },
                'required': ['task_ref'],
                'additionalProperties': False,
            },
        }

    def call(self, task_ref=None, request_id='rq_query_export_task'):
        if self.api_client is None:
            raise RuntimeError('api client is required for query_export_task')
        if not isinstance(task_ref, str):
            raise ValueError('task_ref must be a string')
        task_ref = task_ref.strip()
        if not task_ref or len(task_ref) > 512:
            raise ValueError('task_ref must contain 1 to 512 characters')
        return self.api_client.call_tool(
            self.name,
            self.route_path,
            {'task_ref': task_ref},
            request_id,
        )