url-download-mcp

Instructions

Implementation Reference

• src/mcp_url_downloader/server.py:433-477 (handler)

  Main handler function for the 'download_files' MCP tool. Decorated with @mcp.tool for automatic registration. Handles multiple URL downloads with concurrency limits, security validations (SSRF protection, path traversal), file size checks, sanitization, and returns structured results.
  @mcp.tool(description="Download multiple files from URLs and save to local filesystem.") async def download_files( urls: Annotated[list[str], Field(description="List of URLs to download")], output_dir: Annotated[ str | None, Field(description="Directory to save downloaded files") ] = None, timeout: Annotated[int, Field(description="Download timeout in seconds", ge=1, le=300)] = 60, max_size_mb: Annotated[ int, Field(description="Maximum file size in MB (default: 500)", ge=1, le=5000) ] = MAX_FILE_SIZE_MB, ) -> DownloadResponse: """Download files from URLs and save to the local filesystem. Args: urls: List of URLs to download output_dir: Directory to save the files (defaults to ~/Downloads/mcp_downloads) timeout: Download timeout in seconds (1-300) max_size_mb: Maximum file size in MB (1-5000) Returns: DownloadResponse with results for each file """ if output_dir is None: output_dir = str(DEFAULT_DOWNLOAD_DIR) # Limit number of URLs per request if len(urls) > MAX_URLS_PER_REQUEST: raise ValueError(f"Maximum {MAX_URLS_PER_REQUEST} URLs per request") # Use semaphore to limit concurrent downloads semaphore = asyncio.Semaphore(MAX_CONCURRENT_DOWNLOADS) async def download_with_limit(url: str) -> DownloadResult: async with semaphore: return await _download_single_file_internal(url, output_dir, None, timeout, max_size_mb) # Download all files with concurrency limit tasks = [download_with_limit(url) for url in urls] results = await asyncio.gather(*tasks, return_exceptions=False) success_count = sum(1 for r in results if r.success) failed_count = len(results) - success_count return DownloadResponse(results=results, success_count=success_count, failed_count=failed_count)
• src/mcp_url_downloader/server.py:94-111 (schema)

  Pydantic models defining the output schema for download_files tool: DownloadResult for individual files and DownloadResponse aggregating multiple results. Input schema defined via Annotated parameters in the handler.
  class DownloadResult(BaseModel): """Download result model with file information""" file_path: str = Field(..., description="Full path where the file was saved") file_name: str = Field(..., description="Name of the downloaded file") file_size: int = Field(..., description="Size of the downloaded file in bytes") content_type: str | None = Field(None, description="MIME type of the downloaded file") success: bool = Field(..., description="Whether the download was successful") error: str | None = Field(None, description="Error message if download failed") class DownloadResponse(BaseModel): """Response model for download operations""" results: list[DownloadResult] = Field(..., description="List of download results") success_count: int = Field(..., description="Number of successful downloads") failed_count: int = Field(..., description="Number of failed downloads")
• src/mcp_url_downloader/server.py:91-92 (registration)

  Initialization of the FastMCP server instance. Tool functions decorated with @mcp.tool are automatically registered here.
  mcp = FastMCP("download-server", instructions=DESCRIPTION)
• src/mcp_url_downloader/server.py:300-431 (helper)

  Core helper function performing single file download with HTTP client, size limits, MIME validation, filename sanitization, SSRF protection, and error handling. Used by both download_files and download_single_file tools.
  async def _download_single_file_internal( url: str, output_dir: str, filename: str | None, timeout: int, max_size_mb: int, ) -> DownloadResult: """Internal async function to download a single file. Args: url: URL to download from output_dir: Directory to save file filename: Optional custom filename timeout: Download timeout in seconds max_size_mb: Maximum file size in MB Returns: DownloadResult with download information """ file_path = None try: # Validate URL for SSRF _validate_url_safe(url) # Validate and resolve output directory output_path = _validate_output_dir(output_dir) output_path.mkdir(parents=True, exist_ok=True) # Determine filename if not filename: filename = _extract_filename_from_url(url) else: filename = _sanitize_filename(filename) # Get unique filepath to avoid collisions file_path = _get_unique_filepath(output_path / filename) final_filename = file_path.name max_size_bytes = max_size_mb * 1024 * 1024 # Headers for better compatibility headers = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36" ), "Accept": "*/*", "Accept-Language": "en-US,en;q=0.9", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive", } async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client: # First, do a HEAD request to check size try: head_response = await client.head(url, headers=headers) content_length = head_response.headers.get("Content-Length") if content_length: size = int(content_length) if size > max_size_bytes: size_mb = size / (1024 * 1024) raise ValueError( f"File size ({size_mb:.2f} MB) exceeds " f"maximum allowed size ({max_size_mb} MB)" ) except httpx.HTTPStatusError: # HEAD request not supported, continue with GET pass # Download the file async with client.stream("GET", url, headers=headers) as response: response.raise_for_status() content_type = response.headers.get("Content-Type", "").split(";")[0] downloaded = 0 # Validate MIME type if present if content_type and content_type not in ALLOWED_CONTENT_TYPES: raise ValueError(f"File type not allowed: {content_type}") # Write to file with open(file_path, "wb") as f: async for chunk in response.aiter_bytes(chunk_size=8192): downloaded += len(chunk) # Check size during download if downloaded > max_size_bytes: # Delete partial file if file_path.exists(): file_path.unlink() size_mb = downloaded / (1024 * 1024) raise ValueError( f"File exceeded size limit during download " f"({size_mb:.2f} MB > {max_size_mb} MB)" ) f.write(chunk) # Verify file was created if not file_path.exists(): raise ValueError("File was not created") actual_size = file_path.stat().st_size return DownloadResult( file_path=str(file_path), file_name=final_filename, file_size=actual_size, content_type=content_type, success=True, error=None, ) except Exception as e: # Clean up partial file if exists if file_path and file_path.exists(): try: file_path.unlink() except Exception: pass # Best effort cleanup return DownloadResult( file_path="", file_name=filename or "", file_size=0, content_type=None, success=False, error=_sanitize_error(e), )