td_download_project_archive
Download a Treasure Data workflow project archive containing SQL queries, Digdag files, Python scripts, and resources. Retrieve the complete project bundle for backup or migration purposes.
Instructions
Download a project's archive (tar.gz) and return information about the download.
This tool downloads the complete archive of a Treasure Data workflow project,
which contains all SQL queries, Digdag (.dig) files, Python scripts, and other
resources. The file is temporarily stored on the server.
Args:
project_id: The ID of the workflow project to downloadInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| project_id | Yes |
Implementation Reference
- td_mcp_server/mcp_impl.py:390-442 (handler)The @mcp.tool()-decorated handler function that implements the td_download_project_archive tool. It validates input, creates a TreasureDataClient, downloads the project archive to a secure temporary location, and returns metadata including the archive path.
@mcp.tool() async def td_download_project_archive(project_id: str) -> dict[str, Any]: """Download a project's archive (tar.gz) and return information about the download. This tool downloads the complete archive of a Treasure Data workflow project, which contains all SQL queries, Digdag (.dig) files, Python scripts, and other resources. The file is temporarily stored on the server. Args: project_id: The ID of the workflow project to download """ # Input validation - prevent path traversal if not _validate_project_id(project_id): return _format_error_response("Invalid project ID format") client = _create_client(include_workflow=True) if isinstance(client, dict): return client try: # Create temporary directory with secure permissions temp_dir = tempfile.mkdtemp(prefix="td_project_") os.chmod(temp_dir, TEMP_DIR_PERMISSIONS) # Use sanitized project_id for filename safe_filename = re.sub(r"[^a-zA-Z0-9_-]", "_", project_id) output_path = os.path.join(temp_dir, f"project_{safe_filename}.tar.gz") # Check that project exists before attempting download project = client.get_project(project_id) if not project: return {"error": f"Project with ID '{project_id}' not found"} # Download the archive success = client.download_project_archive(project_id, output_path) if not success: return {"error": f"Failed to download archive for project '{project_id}'"} return { "success": True, "project_id": project_id, "project_name": project.name, "archive_path": output_path, "temp_dir": temp_dir, "message": f"Successfully downloaded archive for project '{project.name}'", } except (ValueError, requests.RequestException, OSError) as e: return _format_error_response(f"Failed to download project archive: {str(e)}") except Exception as e: return _format_error_response( f"Unexpected error while downloading project archive: {str(e)}" )