Add list_buckets tool to enumerate buckets at endpoints

Allows users to ask the bot what buckets are available at a given
backend, making it easier to discover data sources before transfers.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: fluidnumerics-joe

src/xfer/slackbot/claude_agent.py

@@ -22,6 +22,7 @@
     get_source_stats,
     get_transfer_progress,
     get_transfer_progress_by_job,
+    list_buckets,
     submit_transfer,
 )
 
@@ -108,6 +109,28 @@
             "required": [],
         },
     },
+    {
+        "name": "list_buckets",
+        "description": """List buckets available at a specific backend/endpoint.
+
+Use this when users want to:
+- See what buckets exist at a storage endpoint
+- Browse available data sources before starting a transfer
+- Discover bucket names they can use in transfer paths
+
+This runs 'rclone lsd' to list top-level directories (buckets) at the remote.
+Note: The backend must be in the allowed list, and credentials must have ListBuckets permission.""",
+        "input_schema": {
+            "type": "object",
+            "properties": {
+                "backend": {
+                    "type": "string",
+                    "description": "The backend name to list buckets for (e.g., 's3src', 'gcs')",
+                },
+            },
+            "required": ["backend"],
+        },
+    },
     {
         "name": "cancel_job",
         "description": """Cancel a running transfer job.
@@ -248,6 +271,7 @@
 - Submit new data transfer jobs (source -> destination)
 - Check status of running/completed transfers
 - List available storage backends
+- List buckets available at each backend
 - Cancel jobs if requested
 - Request access to new backends on behalf of users
 - Scan source paths to get file statistics and transfer estimates
@@ -353,6 +377,16 @@ def execute_tool(
             backends = get_allowed_backends(self.config)
             return json.dumps({"allowed_backends": backends})
 
+        elif tool_name == "list_buckets":
+            backend = tool_input["backend"]
+            result = list_buckets(backend, self.config)
+            return json.dumps({
+                "backend": result.backend,
+                "buckets": result.buckets,
+                "bucket_count": len(result.buckets),
+                "error": result.error,
+            })
+
         elif tool_name == "cancel_job":
             success, message = cancel_job(
                 tool_input["job_id"],

src/xfer/slackbot/slurm_tools.py

@@ -807,6 +807,15 @@ class PathCheckResult:
     details: Optional[str] = None
 
 
+@dataclass
+class BucketListResult:
+    """Result of listing buckets at an endpoint."""
+
+    backend: str
+    buckets: list[str]
+    error: Optional[str] = None
+
+
 def check_path_exists(path: str, config: BotConfig) -> PathCheckResult:
     """
     Check if a bucket/path exists at a remote endpoint.
@@ -893,6 +902,97 @@ def check_path_exists(path: str, config: BotConfig) -> PathCheckResult:
             )
 
 
+def list_buckets(backend: str, config: BotConfig) -> BucketListResult:
+    """
+    List buckets/top-level directories at a remote endpoint.
+
+    Uses rclone lsd to list directories at the root of the remote.
+
+    Args:
+        backend: The backend name (e.g., "s3src" or "gcs").
+                 Can include a trailing colon (e.g., "s3src:").
+
+    Returns:
+        BucketListResult with list of bucket names or an error.
+    """
+    # Normalize backend name - remove trailing colon if present
+    backend = backend.rstrip(":")
+
+    # Validate backend first
+    valid, msg = validate_backend(backend, config)
+    if not valid:
+        return BucketListResult(
+            backend=backend,
+            buckets=[],
+            error=msg,
+        )
+
+    # Build rclone lsd command to list buckets
+    # lsd lists directories, at root level these are buckets
+    rclone_cmd = [
+        "rclone",
+        "lsd",
+        f"{backend}:",
+        "--config",
+        config.rclone.container_conf_path,
+    ]
+
+    # Build srun command with container
+    mounts = f"{config.rclone.config_path}:{config.rclone.container_conf_path}:ro"
+
+    srun_cmd = [
+        "srun",
+        "-n",
+        "1",
+        "-c",
+        "2",
+        "--container-image",
+        config.rclone.image,
+        "--container-mounts",
+        mounts,
+        "--no-container-remap-root",
+    ] + rclone_cmd
+
+    try:
+        result = run_cmd(srun_cmd, capture=True, check=True)
+        # Parse rclone lsd output - format is:
+        # "          -1 2024-01-15 10:30:00        -1 bucket-name"
+        # We want the last column (bucket name)
+        buckets = []
+        for line in result.stdout.strip().splitlines():
+            parts = line.split()
+            if parts:
+                # Bucket name is the last field
+                buckets.append(parts[-1])
+
+        return BucketListResult(
+            backend=backend,
+            buckets=sorted(buckets),
+        )
+    except subprocess.CalledProcessError as e:
+        error_output = e.stderr or str(e)
+
+        # Check for common error patterns
+        if "AccessDenied" in error_output or "access denied" in error_output.lower():
+            return BucketListResult(
+                backend=backend,
+                buckets=[],
+                error="Access denied - credentials may not have ListBuckets permission",
+            )
+        elif "InvalidAccessKeyId" in error_output:
+            return BucketListResult(
+                backend=backend,
+                buckets=[],
+                error="Invalid access key - check credentials configuration",
+            )
+        else:
+            return BucketListResult(
+                backend=backend,
+                buckets=[],
+                error=f"Failed to list buckets: {error_output[:200]}",
+            )
+
+
 @dataclass
 class JobLogs:
     """Logs and analysis data for a job."""