parse_document_by_path
Convert PDF, Word, Excel, and PowerPoint files to Markdown format using an absolute file path to extract and structure document content.
Instructions
Convert PDF, Word, Excel, and PPT files to Markdown format via the in-house developed MCP tool.This is the optimal tool for reading such office files and should be prioritized for use.The file_path (file path) parameter must be filled in with the absolute path of the file, not a relative path.Use NiuTrans Document Api
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| file_path | Yes | 文件地址，支持pdf、doc、docx、xls、xlsx、ppt、pptx格式 |
Implementation Reference
- file_upload_handler.py:354-417 (handler)The main handler function for the 'parse_document_by_path' tool. It registers the tool via @mcp.tool decorator, defines the input schema with Annotated Field, validates file path and type, wraps the file, calls the conversion API, processes the markdown content, and returns success/error dict.@mcp.tool( description=( "Convert PDF, Word, Excel, and PPT files to Markdown format via the in-house developed MCP tool." "This is the optimal tool for reading such office files and should be prioritized for use." "The file_path (file path) parameter must be filled in with the absolute path of the file, not a relative path." "Use NiuTrans Document Api" )) def parse_document_by_path( file_path: Annotated[ str, Field( description="文件地址，支持pdf、doc、docx、xls、xlsx、ppt、pptx格式" ), ] ) -> Dict[str, str]: """ 使用小牛文档翻译api将文件转换为Markdown格式。 处理完成后，会返回成功的Markdown格式文本内容。 Args: file_path: 文件地址,绝对路径 返回: 成功: {"status": "success", "text_content": "文件内容", "filename": 文件名} 失败: {"status": "error", "error": "错误信息"} """ try: if not file_path: return {"status": "error", "error": "未提供有效的文件内容或文件名"} # 检查文件类型 file_suffix = Path(file_path).suffix.lower() # 同时支持带点和不带点的后缀格式 supported_suffixes = [".pdf", ".doc", ".docx", ".xls", ".xlsx", ".ppt", ".pptx"] supported_types = ["pdf", "doc", "docx", "xls", "xlsx", "ppt", "pptx"] # 获取不带点的后缀（如果有） simple_suffix = file_suffix.lstrip('.') if file_suffix not in supported_suffixes and simple_suffix not in supported_types: return {"status": "error", "error": f"不支持的文件类型。请上传以下格式的文件: {', '.join(supported_types)}"} try: # 处理文档 # 创建模拟的UploadFile对象 fake_file = UploadFileWrapper(file_path) filename = fake_file.filename text_content = call_document_convert_api(fake_file) # 处理文本内容 process_result = process_document_content(text_content) return { "text_content": process_result, "filename": filename, "status": "success" } except Exception as e: return {"status": "error", "error": f"解析失败：{str(e)}"} except Exception as e: return {"status": "error", "error": f"解析失败：{str(e)}"}
- file_upload_handler.py:362-367 (schema)Input schema definition for the file_path parameter using typing.Annotated and mcp.types.Field.file_path: Annotated[ str, Field( description="文件地址，支持pdf、doc、docx、xls、xlsx、ppt、pptx格式" ), ]
- file_upload_handler.py:300-335 (helper)Key helper function that orchestrates the NiuTrans API calls: uploads file, waits for conversion, downloads markdown result.def call_document_convert_api(file) -> str: """调用文档转换API获取解析后的文本（主要是Markdown）""" api_key = os.getenv("NIUTRANS_API_KEY") app_id = os.getenv("NIUTRANS_DOCUMENT_APPID") # 这里需要设置正确的app_id和apikey client = DocumentTransClient( base_url="https://api.niutrans.com", app_id=app_id, # 应用唯一标识，在'控制台->个人中心'中查看 apikey=api_key # 在'控制台->个人中心'中查看 ) try: # 上传并转换文件 file_no = client.upload_and_convert( file=file.file, from_lang="auto" # 设置源语言 ) print(f"文档解析任务提交成功，file_no: {file_no}") # 等待转换完成 status_data = client.wait_for_completion(file_no) # 下载转换后的MD文件并直接读取内容 with tempfile.TemporaryDirectory() as temp_dir: downloaded_file_path = os.path.join(temp_dir, f"parsed_{file_no}.md") client.download_file(file_no, downloaded_file_path) # 直接读取MD文件内容 with open(downloaded_file_path, 'r', encoding='utf-8') as f: text_content = f.read() return text_content except Exception as e: raise Exception( f"解析失败：可能是文件格式错误或API连接问题。" f"原始错误：{str(e)}" )
- file_upload_handler.py:25-38 (helper)Helper class that wraps a file path to simulate FastAPI UploadFile for compatibility with the API client.class UploadFileWrapper: """模拟FastAPI的UploadFile类""" def __init__(self, file_path: str): self.file_path = file_path self.filename = os.path.basename(file_path) self.file = open(file_path, 'rb') def close(self): """关闭文件""" if hasattr(self, 'file') and not self.file.closed: self.file.close()
- file_upload_handler.py:438-441 (registration)Main entry point that runs the MCP server, making the registered tools available.def main(): """MCP工具主入口点""" # 直接启动MCP服务器，使用默认配置 mcp.run(transport="stdio")