diff --git a/crates/mcp/README.md b/crates/mcp/README.md
index d6b0e34c8..ed5c52f27 100644
--- a/crates/mcp/README.md
+++ b/crates/mcp/README.md
@@ -231,6 +231,20 @@ Retrieve an object from S3 with two operation modes: read content directly or do
 - `local_path` (string, optional): Local file path (required when mode is "download")
 - `max_content_size` (number, optional): Maximum content size in bytes for read mode (default: 1MB)
 
+### `create_bucket`
+
+Create a new S3 bucket with the specified name.
+
+**Parameters:**
+
+- `bucket_name` (string): Source S3 bucket.
+
+### `delete_bucket`
+
+Delete the specified S3 bucket. If the bucket is not empty, the deletion will fail. You should delete all objects and objects inside them before calling this method.**WARNING: This operation will permanently delete the bucket and all objects within it!**
+
+- `bucket_name` (string): Source S3 bucket.
+
 ## Architecture
 
 The MCP server is built with a modular architecture:
diff --git a/crates/mcp/src/s3_client.rs b/crates/mcp/src/s3_client.rs
index c35dc66b7..047a9aab3 100644
--- a/crates/mcp/src/s3_client.rs
+++ b/crates/mcp/src/s3_client.rs
@@ -151,6 +151,36 @@ impl S3Client {
         Ok(Self { client })
     }
 
+    pub async fn create_bucket(&self, bucket_name: &str) -> Result<BucketInfo> {
+        info!("Creating S3 bucket: {}", bucket_name);
+
+        self.client
+            .create_bucket()
+            .bucket(bucket_name)
+            .send()
+            .await
+            .context(format!("Failed to create S3 bucket: {bucket_name}"))?;
+
+        info!("Bucket '{}' created successfully", bucket_name);
+        Ok(BucketInfo {
+            name: bucket_name.to_string(),
+            creation_date: None, // Creation date not returned by create_bucket
+        })
+    }
+
+    pub async fn delete_bucket(&self, bucket_name: &str) -> Result<()> {
+        info!("Deleting S3 bucket: {}", bucket_name);
+        self.client
+            .delete_bucket()
+            .bucket(bucket_name)
+            .send()
+            .await
+            .context(format!("Failed to delete S3 bucket: {bucket_name}"))?;
+
+        info!("Bucket '{}' deleted successfully", bucket_name);
+        Ok(())
+    }
+
     pub async fn list_buckets(&self) -> Result<Vec<BucketInfo>> {
         debug!("Listing S3 buckets");
 
diff --git a/crates/mcp/src/server.rs b/crates/mcp/src/server.rs
index 02bfec066..da69922d6 100644
--- a/crates/mcp/src/server.rs
+++ b/crates/mcp/src/server.rs
@@ -54,6 +54,18 @@ pub struct UploadFileRequest {
     pub cache_control: Option<String>,
 }
 
+#[derive(Serialize, Deserialize, JsonSchema)]
+pub struct CreateBucketReqeust {
+    #[schemars(description = "Name of the S3 bucket to create")]
+    pub bucket_name: String,
+}
+
+#[derive(Serialize, Deserialize, JsonSchema)]
+pub struct DeleteBucketReqeust {
+    #[schemars(description = "Name of the S3 bucket to delete")]
+    pub bucket_name: String,
+}
+
 #[derive(Serialize, Deserialize, JsonSchema)]
 pub struct GetObjectRequest {
     #[schemars(description = "Name of the S3 bucket")]
@@ -110,6 +122,53 @@ impl RustfsMcpServer {
         })
     }
 
+    #[tool(description = "Create a new S3 bucket with the specified name")]
+    pub async fn create_bucket(&self, Parameters(req): Parameters<CreateBucketReqeust>) -> String {
+        info!("Executing create_bucket tool for bucket: {}", req.bucket_name);
+
+        match self.s3_client.create_bucket(&req.bucket_name).await {
+            Ok(_) => {
+                format!("Successfully created bucket: {}", req.bucket_name)
+            }
+            Err(e) => {
+                format!("Failed to create bucket '{}': {:?}", req.bucket_name, e)
+            }
+        }
+    }
+
+    #[tool(description = "Delete an existing S3 bucket with the specified name")]
+    pub async fn delete_bucket(&self, Parameters(req): Parameters<DeleteBucketReqeust>) -> String {
+        info!("Executing delete_bucket tool for bucket: {}", req.bucket_name);
+
+        // check if bucket is empty, if not, can not delete bucket directly.
+        let object_result = match self
+            .s3_client
+            .list_objects_v2(&req.bucket_name, ListObjectsOptions::default())
+            .await
+        {
+            Ok(result) => result,
+            Err(e) => {
+                error!("Failed to list objects in bucket '{}': {:?}", req.bucket_name, e);
+                return format!("Failed to list objects in bucket '{}': {:?}", req.bucket_name, e);
+            }
+        };
+
+        if !object_result.objects.is_empty() {
+            error!("Bucket '{}' is not empty", req.bucket_name);
+            return format!("Failed to delete bucket '{}': bucket is not empty", req.bucket_name);
+        }
+
+        // delete the bucket.
+        match self.s3_client.delete_bucket(&req.bucket_name).await {
+            Ok(_) => {
+                format!("Successfully deleted bucket: {}", req.bucket_name)
+            }
+            Err(e) => {
+                format!("Failed to delete bucket '{}': {:?}", req.bucket_name, e)
+            }
+        }
+    }
+
     #[tool(description = "List all S3 buckets accessible with the configured credentials")]
     pub async fn list_buckets(&self) -> String {
         info!("Executing list_buckets tool");
@@ -667,4 +726,20 @@ mod tests {
         assert_eq!(read_mode_deser, GetObjectMode::Read);
         assert_eq!(download_mode_deser, GetObjectMode::Download);
     }
+
+    #[test]
+    fn test_bucket_creation() {
+        let request = CreateBucketReqeust {
+            bucket_name: "test-bucket".to_string(),
+        };
+        assert_eq!(request.bucket_name, "test-bucket");
+    }
+
+    #[test]
+    fn test_bucket_deletion() {
+        let request = DeleteBucketReqeust {
+            bucket_name: "test-bucket".to_string(),
+        };
+        assert_eq!(request.bucket_name, "test-bucket");
+    }
 }