mcp-pdf2md

Instructions

Implementation Reference

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

  The main execution logic for the 'convert_pdf_file' MCP tool. Decorated with @mcp.tool() for automatic registration and schema inference from signature/docstring. Handles local PDF(s): validation, upload to MinerU API via batch file-urls, polls task status, downloads and unzips Markdown results.
  @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:395-405 (schema)

  Function signature and docstring defining input schema (file_path: str, enable_ocr: bool) and output (Dict[str, Any]) for the tool, used by FastMCP for JSON schema generation.
  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 """
• src/pdf2md/server.py:394-394 (registration)

  FastMCP decorator that registers 'convert_pdf_file' as an MCP tool, inferring schema from function signature and docstring.
  @mcp.tool()
• src/pdf2md/server.py:277-308 (helper)

  Utility function used by convert_pdf_file to parse input file_path string into a list of individual paths.
  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)