parse_document_by_path

Instructions

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")