Treasure Data MCP Server

Instructions

Implementation Reference

• td_mcp_server/mcp_impl.py:444-510 (handler)

  Handler function implementing the td_list_project_files tool. Decorated with @mcp.tool() for automatic registration in the MCP server. Validates input, lists files from tar.gz archive securely, categorizes by type, and returns structured file list.
  @mcp.tool() async def td_list_project_files(archive_path: str) -> dict[str, Any]: """List all files contained in a project archive. This tool extracts and lists the content of a previously downloaded project archive, showing all files and directories within the project. Args: archive_path: The path to the downloaded project archive (.tar.gz file) """ # Input validation - prevent path traversal if not _validate_archive_path(archive_path): return _format_error_response("Invalid archive path") try: if not os.path.exists(archive_path): return _format_error_response("Archive file not found") file_list = [] with tarfile.open(archive_path, "r:gz") as tar: for member in tar.getmembers(): # Security check for each member if not _safe_extract_member(member, "/tmp/validation"): continue # Skip unsafe members file_info = { "name": member.name, "type": "directory" if member.isdir() else "file", "size": member.size, } # Add extension information for files if not member.isdir(): ext = Path(member.name).suffix.lower() file_info["extension"] = ext # Identify file types based on extension if ext == ".dig": file_info["file_type"] = "Digdag workflow" elif ext == ".sql": file_info["file_type"] = "SQL query" elif ext == ".py": file_info["file_type"] = "Python script" elif ext in [".yml", ".yaml"]: file_info["file_type"] = "YAML configuration" else: file_info["file_type"] = "Other" file_list.append(file_info) # Sort files: directories first, then by name file_list.sort(key=lambda x: (0 if x["type"] == "directory" else 1, x["name"])) return { "success": True, "archive_path": archive_path, "file_count": len(file_list), "files": file_list, } except (OSError, tarfile.ReadError) as e: return _format_error_response(f"Failed to list project files: {str(e)}") except Exception as e: return _format_error_response( f"Unexpected error while listing project files: {str(e)}" )
• td_mcp_server/mcp_impl.py:61-82 (helper)

  Helper function to validate the archive_path parameter, ensuring it's a safe .tar.gz path in a temporary directory.
  def _validate_archive_path(archive_path: str) -> bool: """Validate archive path to ensure it's in allowed temporary directories.""" if not archive_path: return False # Normalize the path to prevent tricks normalized_path = os.path.normpath(archive_path) # Allow paths in temp directories or test paths temp_prefix = tempfile.gettempdir() allowed_prefixes = [temp_prefix, "/tmp"] if not any(normalized_path.startswith(prefix) for prefix in allowed_prefixes): return False # Prevent path traversal if ".." in normalized_path: return False if not archive_path.endswith(".tar.gz"): return False return True
• td_mcp_server/mcp_impl.py:85-107 (helper)

  Helper function for secure tar member validation to prevent path traversal and large file extraction.
  def _safe_extract_member(member, extract_path: str) -> bool: """Safely extract a tar member, preventing path traversal and other attacks.""" # Normalize the member name member_path = os.path.normpath(member.name) # Prevent absolute paths if member_path.startswith("/") or member_path.startswith("\\"): return False # Prevent path traversal if ".." in member_path: return False # Check final extracted path final_path = os.path.join(extract_path, member_path) if not final_path.startswith(extract_path): return False # Check file size (prevent zip bombs) if hasattr(member, "size") and member.size > MAX_FILE_SIZE: return False return True
• td_mcp_server/mcp_impl.py:110-112 (helper)

  Utility helper to format consistent error responses used throughout the tool.
  def _format_error_response(error_msg: str) -> dict[str, str]: """Format error response without exposing sensitive information.""" return {"error": error_msg}