mcp-pdf2md

Instructions

Convert local PDF file to Markdown, supports single file or file list

Args:
    file_path: PDF file local path or path list, can be separated by spaces, commas, or newlines
    enable_ocr: Whether to enable OCR (default: True)

Returns:
    dict: Conversion result information

Implementation Reference

• src/pdf2md/server.py:394-504 (handler)

  The core handler function for the 'convert_pdf_file' tool. It handles uploading local PDF files to the MinerU API using presigned URLs, polling for processing completion, and downloading the resulting Markdown ZIP files which are automatically extracted.

  @mcp.tool()  
  async def convert_pdf_file(file_path: str, enable_ocr: bool = True) -> Dict[str, Any]:
      """
      Convert local PDF file to Markdown, supports single file or file list
      
      Args:
          file_path: PDF file local path or path list, can be separated by spaces, commas, or newlines
          enable_ocr: Whether to enable OCR (default: True)
  
      Returns:
          dict: Conversion result information
      """
      if not MINERU_API_KEY:
          return {"success": False, "error": "Missing API key, please set environment variable MINERU_API_KEY"}
      
      if isinstance(file_path, str):
          file_paths = parse_path_string(file_path)
      else:
          file_paths = [file_path]  
      
      for path in file_paths:
          if not os.path.exists(path):
              return {"success": False, "error": f"File does not exist: {path}"}
          else:
              if not path.lower().endswith('.pdf'):
                  return {"success": False, "error": f"File is not in PDF format: {path}"}
      
      async with httpx.AsyncClient(timeout=300.0) as client:
          try:
              file_names = [os.path.basename(path) for path in file_paths]
              
              files_data = []
              for i, name in enumerate(file_names):
                  files_data.append({
                      "name": name,
                      "is_ocr": enable_ocr,
                      "data_id": f"file_convert_{i+1}_{int(time.time())}"
                  })
              
              file_url_data = {
                  "enable_formula": True,
                  "language": "auto",
                  "layout_model": "doclayout_yolo",
                  "enable_table": True,
                  "files": files_data
              }
              
              file_url_response = await client.post(
                  MINERU_FILE_URLS_API,
                  headers=HEADERS,
                  json=file_url_data,
                  timeout=60.0
              )
              
              if file_url_response.status_code != 200:
                  return {"success": False, "error": f"Failed to get upload link: {file_url_response.status_code}"}
              
              file_url_result = file_url_response.json()
              
              if file_url_result.get("code") != 0 and file_url_result.get("code") != 200:
                  error_msg = file_url_result.get("msg", "Unknown error")
                  return {"success": False, "error": f"Failed to get upload link: {error_msg}"}
              
              batch_id = file_url_result.get("data", {}).get("batch_id", "")
              file_urls = file_url_result.get("data", {}).get("file_urls", [])
              
              if not batch_id or not file_urls or len(file_urls) != len(file_paths):
                  return {"success": False, "error": "Failed to get upload link or batch ID"}
              
              upload_results = []
              for i, (file_path, upload_url) in enumerate(zip(file_paths, file_urls)):
                  try:
                      with open(file_path, 'rb') as f:
                          file_content = f.read()
                          
                          upload_response = await client.put(
                              upload_url,
                              content=file_content,
                              headers={},  
                              timeout=300.0
                          )
                      
                      if upload_response.status_code != 200:
                          upload_results.append({"file": file_names[i], "success": False})
                      else:
                          upload_results.append({"file": file_names[i], "success": True})
                  except Exception as e:
                      upload_results.append({"file": file_names[i], "success": False, "error": str(e)})
              
              if not any(result["success"] for result in upload_results):
                  return {"success": False, "error": "All files failed to upload", "upload_results": upload_results}
              
              task_status = await check_task_status(client, batch_id)
              
              if not task_status.get("success"):
                  return task_status
              
              downloaded_files = await download_batch_results(client, task_status.get("extract_results", []))
              
              return {
                  "success": True, 
                  "downloaded_files": downloaded_files,
                  "batch_id": batch_id,
                  "upload_results": upload_results,
                  "total_files": len(file_paths),
                  "processed_files": len(downloaded_files)
              }
                  
          except Exception as e:
              return {"success": False, "error": str(e)}

• src/pdf2md/server.py:394-394 (registration)

  The @mcp.tool() decorator registers the convert_pdf_file function as an MCP tool.

  @mcp.tool()  

• src/pdf2md/server.py:277-309 (helper)

  Helper function to parse input file_path string into a list of paths, handling quotes, spaces, commas, newlines.

  def parse_path_string(path_string):
      """
      Parse file path string separated by spaces, commas, or newlines
      
      Args:
          path_string: File path string
          
      Returns:
          list: List of file paths
      """
      if isinstance(path_string, str):
          if (path_string.startswith('"') and path_string.endswith('"')) or \
             (path_string.startswith("'") and path_string.endswith("'")):
              path_string = path_string[1:-1]
      
      paths = []
      for part in path_string.split():
          if ',' in part:
              paths.extend(part.split(','))
          elif '\n' in part:
              paths.extend(part.split('\n'))
          else:
              paths.append(part)
      
      cleaned_paths = []
      for path in paths:
          if (path.startswith('"') and path.endswith('"')) or \
             (path.startswith("'") and path.endswith("'")):
              cleaned_paths.append(path[1:-1])
          else:
              cleaned_paths.append(path)
      
      return cleaned_paths

• src/pdf2md/server.py:73-504 (helper)

  Supporting helper functions for task status polling, batch result downloading, ZIP extraction, and status printing, shared with convert_pdf_url but used by convert_pdf_file.

  async def check_task_status(client, batch_id, max_retries=60, sleep_seconds=5):
      """
      Check batch task status
      
      Args:
          client: HTTP client
          batch_id: Batch ID
          max_retries: Maximum number of retries
          sleep_seconds: Seconds between retries
          
      Returns:
          dict: Dictionary containing task status information, or error message if failed
      """
      retry_count = 0
      
      while retry_count < max_retries:
          retry_count += 1
          
          try:
              status_response = await client.get(
                  f"{MINERU_BATCH_RESULTS_API}/{batch_id}",
                  headers=HEADERS,
                  timeout=60.0  
              )
              
              if status_response.status_code != 200:
                  retry_count += 1
                  if retry_count < max_retries:
                      await asyncio.sleep(sleep_seconds)
                  continue
              
              try:
                  status_data = status_response.json()
              except Exception as e:
                  retry_count += 1
                  if retry_count < max_retries:
                      await asyncio.sleep(sleep_seconds)
                  continue
              
              task_data = status_data.get("data", {})
              extract_results = task_data.get("extract_result", [])
              
              all_done, any_done = print_task_status(extract_results)
              
              if all_done:
                  return {
                      "success": True,
                      "extract_results": extract_results,
                      "task_data": task_data,
                      "status_data": status_data
                  }
              
              await asyncio.sleep(sleep_seconds)
              
          except Exception as e:
              retry_count += 1
              if retry_count < max_retries:
                  await asyncio.sleep(sleep_seconds)
      
      return {
          "success": False,
          "error": "Polling timeout, unable to get final results"
      }
  
  async def download_batch_results(client, extract_results):
      """
      Download batch task results
      
      Args:
          client: HTTP client
          extract_results: List of task results
          
      Returns:
          list: List of downloaded file information
      """
      downloaded_files = []
      
      for i, result in enumerate(extract_results):
          if result.get("state") == "done":
              try:
                  file_name = result.get("file_name", f"file_{i+1}")
                  zip_url = result.get("full_zip_url", "")
                  
                  if not zip_url:
                      continue
                  
                  downloaded_file = await download_zip_file(client, zip_url, file_name)
                  if downloaded_file:
                      downloaded_files.append(downloaded_file)
              except Exception as e:
                  pass
      
      return downloaded_files
  
  async def download_zip_file(client, zip_url, file_name, prefix="md", max_retries=3):
      """
      Download and save ZIP file, then automatically unzip
      
      Args:
          client: HTTP client
          zip_url: ZIP file URL
          file_name: File name
          prefix: File prefix
          max_retries: Maximum number of retries
          
      Returns:
          dict: Dictionary containing file name and unzip directory, or None if failed
      """
      retry_count = 0
      
      while retry_count < max_retries:
          try:
              zip_response = await client.get(zip_url, follow_redirects=True, timeout=120.0)
              
              if zip_response.status_code == 200:
                  current_date = time.strftime("%Y%m%d")
                  
                  base_name = os.path.splitext(file_name)[0]
                  # Only remove control characters and chars that are invalid in filenames
                  safe_name = re.sub(r'[<>:"/\\|?*\x00-\x1f]', '', base_name).strip()
                  # Replace spaces with underscores
                  safe_name = re.sub(r'\s+', '_', safe_name)
                  
                  if safe_name.isdigit() or re.match(r'^\d+\.\d+$', safe_name):
                      safe_name = f"paper_{safe_name}"
                      
                  zip_filename = f"{prefix}_{safe_name}_{current_date}.zip"
                  
                  download_dir = Path(OUTPUT_DIR)
                  if not download_dir.exists():
                      try:
                          download_dir.mkdir(parents=True, exist_ok=True)
                      except Exception as e:
                          print(f"Error creating directory: {e}")
                          return None
                  
                  save_path = download_dir / zip_filename
                  
                  with open(save_path, "wb") as f:
                      f.write(zip_response.content)
                  
                  extract_dir = download_dir / safe_name
                  if not extract_dir.exists():
                      extract_dir.mkdir(parents=True)
                  
                  import zipfile
                  try:
                      with zipfile.ZipFile(save_path, 'r') as zip_ref:
                          zip_ref.extractall(extract_dir)
                      os.remove(save_path)
                      
                      return {
                          "file_name": file_name,
                          "extract_dir": str(extract_dir)
                      }
                  except Exception as e:
                      pass
              else:
                  retry_count += 1
                  if retry_count < max_retries:
                      await asyncio.sleep(2)
                  continue
          except Exception as e:
              retry_count += 1
              if retry_count < max_retries:
                  await asyncio.sleep(2)
              continue
      
      return None
  
  def parse_url_string(url_string):
      """
      Parse URL string separated by spaces, commas, or newlines
      
      Args:
          url_string: URL string
          
      Returns:
          list: List of URLs
      """
      if isinstance(url_string, str):
          if (url_string.startswith('"') and url_string.endswith('"')) or \
             (url_string.startswith("'") and url_string.endswith("'")):
              url_string = url_string[1:-1]
      
      urls = []
      for part in url_string.split():
          if ',' in part:
              urls.extend(part.split(','))
          elif '\n' in part:
              urls.extend(part.split('\n'))
          else:
              urls.append(part)
      
      cleaned_urls = []
      for url in urls:
          if (url.startswith('"') and url.endswith('"')) or \
             (url.startswith("'") and url.endswith("'")):
              cleaned_urls.append(url[1:-1])
          else:
              cleaned_urls.append(url)
      
      return cleaned_urls
  
  def parse_path_string(path_string):
      """
      Parse file path string separated by spaces, commas, or newlines
      
      Args:
          path_string: File path string
          
      Returns:
          list: List of file paths
      """
      if isinstance(path_string, str):
          if (path_string.startswith('"') and path_string.endswith('"')) or \
             (path_string.startswith("'") and path_string.endswith("'")):
              path_string = path_string[1:-1]
      
      paths = []
      for part in path_string.split():
          if ',' in part:
              paths.extend(part.split(','))
          elif '\n' in part:
              paths.extend(part.split('\n'))
          else:
              paths.append(part)
      
      cleaned_paths = []
      for path in paths:
          if (path.startswith('"') and path.endswith('"')) or \
             (path.startswith("'") and path.endswith("'")):
              cleaned_paths.append(path[1:-1])
          else:
              cleaned_paths.append(path)
      
      return cleaned_paths
  
  # Create MCP server
  mcp = FastMCP("PDF to Markdown Conversion Service")
  
  @mcp.tool()  
  async def convert_pdf_url(url: str, enable_ocr: bool = True) -> Dict[str, Any]:
      """
      Convert PDF URL to Markdown, supports single URL or URL list
      
      Args:
          url: PDF file URL or URL list, can be separated by spaces, commas, or newlines
          enable_ocr: Whether to enable OCR (default: True)
  
      Returns:
          dict: Conversion result information
      """
      if not MINERU_API_KEY:
          return {"success": False, "error": "Missing API key, please set environment variable MINERU_API_KEY"}
      
      if isinstance(url, str):
          urls = parse_url_string(url)
      else:
          urls = [url]  
      
      async with httpx.AsyncClient(timeout=300.0) as client:
          try:
              files = []
              for i, url_item in enumerate(urls):
                  files.append({
                      "url": url_item, 
                      "is_ocr": enable_ocr, 
                      "data_id": f"url_convert_{i+1}_{int(time.time())}"
                  })
              
              batch_data = {
                  "enable_formula": True,
                  "language": "auto",
                  "layout_model": "doclayout_yolo",
                  "enable_table": True,
                  "files": files
              }
              
              response = await client.post(
                  MINERU_BATCH_API,
                  headers=HEADERS,
                  json=batch_data,
                  timeout=300.0
              )
              
              if response.status_code != 200:
                  return {"success": False, "error": f"Request failed: {response.status_code}"}
              
              try:
                  status_data = response.json()
                  
                  if status_data.get("code") != 0 and status_data.get("code") != 200:
                      error_msg = status_data.get("msg", "Unknown error")
                      return {"success": False, "error": f"API returned error: {error_msg}"}
                      
                  batch_id = status_data.get("data", {}).get("batch_id", "")
                  if not batch_id:
                      return {"success": False, "error": "Failed to get batch ID"}
                  
                  task_status = await check_task_status(client, batch_id)
                  
                  if not task_status.get("success"):
                      return task_status
                  
                  downloaded_files = await download_batch_results(client, task_status.get("extract_results", []))
                  
                  return {
                      "success": True, 
                      "downloaded_files": downloaded_files,
                      "batch_id": batch_id,
                      "total_urls": len(urls),
                      "processed_urls": len(downloaded_files)
                  }
                  
              except json.JSONDecodeError as e:
                  return {"success": False, "error": f"Failed to parse JSON: {e}"}
                  
          except Exception as e:
              return {"success": False, "error": str(e)}
  
  @mcp.tool()  
  async def convert_pdf_file(file_path: str, enable_ocr: bool = True) -> Dict[str, Any]:
      """
      Convert local PDF file to Markdown, supports single file or file list
      
      Args:
          file_path: PDF file local path or path list, can be separated by spaces, commas, or newlines
          enable_ocr: Whether to enable OCR (default: True)
  
      Returns:
          dict: Conversion result information
      """
      if not MINERU_API_KEY:
          return {"success": False, "error": "Missing API key, please set environment variable MINERU_API_KEY"}
      
      if isinstance(file_path, str):
          file_paths = parse_path_string(file_path)
      else:
          file_paths = [file_path]  
      
      for path in file_paths:
          if not os.path.exists(path):
              return {"success": False, "error": f"File does not exist: {path}"}
          else:
              if not path.lower().endswith('.pdf'):
                  return {"success": False, "error": f"File is not in PDF format: {path}"}
      
      async with httpx.AsyncClient(timeout=300.0) as client:
          try:
              file_names = [os.path.basename(path) for path in file_paths]
              
              files_data = []
              for i, name in enumerate(file_names):
                  files_data.append({
                      "name": name,
                      "is_ocr": enable_ocr,
                      "data_id": f"file_convert_{i+1}_{int(time.time())}"
                  })
              
              file_url_data = {
                  "enable_formula": True,
                  "language": "auto",
                  "layout_model": "doclayout_yolo",
                  "enable_table": True,
                  "files": files_data
              }
              
              file_url_response = await client.post(
                  MINERU_FILE_URLS_API,
                  headers=HEADERS,
                  json=file_url_data,
                  timeout=60.0
              )
              
              if file_url_response.status_code != 200:
                  return {"success": False, "error": f"Failed to get upload link: {file_url_response.status_code}"}
              
              file_url_result = file_url_response.json()
              
              if file_url_result.get("code") != 0 and file_url_result.get("code") != 200:
                  error_msg = file_url_result.get("msg", "Unknown error")
                  return {"success": False, "error": f"Failed to get upload link: {error_msg}"}
              
              batch_id = file_url_result.get("data", {}).get("batch_id", "")
              file_urls = file_url_result.get("data", {}).get("file_urls", [])
              
              if not batch_id or not file_urls or len(file_urls) != len(file_paths):
                  return {"success": False, "error": "Failed to get upload link or batch ID"}
              
              upload_results = []
              for i, (file_path, upload_url) in enumerate(zip(file_paths, file_urls)):
                  try:
                      with open(file_path, 'rb') as f:
                          file_content = f.read()
                          
                          upload_response = await client.put(
                              upload_url,
                              content=file_content,
                              headers={},  
                              timeout=300.0
                          )
                      
                      if upload_response.status_code != 200:
                          upload_results.append({"file": file_names[i], "success": False})
                      else:
                          upload_results.append({"file": file_names[i], "success": True})
                  except Exception as e:
                      upload_results.append({"file": file_names[i], "success": False, "error": str(e)})
              
              if not any(result["success"] for result in upload_results):
                  return {"success": False, "error": "All files failed to upload", "upload_results": upload_results}
              
              task_status = await check_task_status(client, batch_id)
              
              if not task_status.get("success"):
                  return task_status
              
              downloaded_files = await download_batch_results(client, task_status.get("extract_results", []))
              
              return {
                  "success": True, 
                  "downloaded_files": downloaded_files,
                  "batch_id": batch_id,
                  "upload_results": upload_results,
                  "total_files": len(file_paths),
                  "processed_files": len(downloaded_files)
              }
                  
          except Exception as e:
              return {"success": False, "error": str(e)}