Treasure Data MCP Server

Instructions

Implementation Reference

• td_mcp_server/mcp_impl.py:390-441 (handler)

  The primary handler function for the 'td_download_project_archive' tool. It is registered via the @mcp.tool() decorator. Handles input validation, temporary file creation, project existence check, calls the API client to download the archive, and returns structured results or errors.
  @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)}" )
• td_mcp_server/api.py:419-462 (helper)

  Supporting helper method in TreasureDataClient class that performs the core HTTP request to download the project archive as a tar.gz file from the Treasure Data API and saves it locally.
  def download_project_archive(self, project_id: str, output_path: str) -> bool: """ Download a project's archive as a tar.gz file. This method downloads the complete archive of a workflow project, including all SQL queries, Digdag workflow files, Python scripts, and other resources. The archive is saved as a tar.gz file at the specified output path. Args: project_id: The ID of the workflow project to download output_path: The file path where the archive will be saved Returns: True if the download was successful, False otherwise Raises: requests.HTTPError: If the API returns an error response (except 404) IOError: If there's an issue writing to the output file """ url = f"{self.workflow_base_url}/projects/{project_id}/archive" try: response = requests.get(url, headers=self.headers, stream=True) # Handle 404 specifically before raising other status codes if response.status_code == 404: return False # Raise for other error status codes response.raise_for_status() with open(output_path, "wb") as f: for chunk in response.iter_content(chunk_size=8192): f.write(chunk) return True except (OSError, requests.RequestException) as e: # Re-raise if it's not a 404 error if not isinstance(e, requests.HTTPError) or ( hasattr(e, "response") and e.response.status_code != 404 ): raise return False