Treasure Data MCP Server

Instructions

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

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)}"
          )