download_files
Download multiple files from URLs to your local filesystem with configurable timeout and file size limits. Specify output directory and manage batch downloads efficiently.
Instructions
Download multiple files from URLs and save to local filesystem.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| max_size_mb | No | Maximum file size in MB (default: 500) | |
| output_dir | No | Directory to save downloaded files | |
| timeout | No | Download timeout in seconds | |
| urls | Yes | List of URLs to download |
Implementation Reference
- src/mcp_url_downloader/server.py:433-477 (handler)The main handler function for the 'download_files' MCP tool. It is registered using the @mcp.tool decorator and implements batch file downloading from URLs with concurrency control, security checks, and structured response.@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)
- Pydantic schema for the tool's output: DownloadResponse model containing aggregated download results.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")
- Pydantic schema for individual download results used in the tool's response.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")
- Core helper function that handles downloading a single file, including URL validation, SSRF protection, size limits, MIME type checks, and file I/O.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), )
- src/mcp_url_downloader/server.py:433-433 (registration)The @mcp.tool decorator registers the download_files function as an MCP tool with FastMCP.@mcp.tool(description="Download multiple files from URLs and save to local filesystem.")