Add Batch Processing CLI commands

Author: stellasphere

CLI-COMMANDS.md

@@ -47,6 +47,27 @@ roboflow download my-workspace/my-project/3 -f coco   # alias
 roboflow infer photo.jpg -m my-project/3
 ```
 
+### Batch Process Asset Library images
+
+```bash
+# Exact reviewed selection (CPU is the product default):
+roboflow batch create --workflow inspect-defects --image-ids img_1,img_2
+
+# Or every current match for a structured RoboQL filter:
+roboflow batch create --workflow inspect-defects --query "tag:night-shift"
+
+# Monitor and control the durable job:
+roboflow batch status <job-id>
+roboflow batch list
+roboflow batch abort <job-id>
+roboflow batch restart <job-id>
+```
+
+The create response includes `taskId`, `jobId`, and `requestId`. A job continues if the terminal
+closes. If a create request has an ambiguous network result, retry with the same `--request-id` to
+avoid a duplicate. Local folders must first be uploaded into Roboflow; Batch Processing never sends
+local file contents through an Agent or CLI job-configuration request.
+
 ### Train, monitor, cancel, stop
 
 ```bash

roboflow/adapters/rfapi.py

@@ -1376,6 +1376,125 @@ def get_video_job_status(api_key, job_id):
     return response.json()
 
 
+# ---------------------------------------------------------------------------
+# Batch Processing (Asset Library orchestration)
+# ---------------------------------------------------------------------------
+
+
+def _batch_processing_url(workspace_url, suffix=""):
+    return f"{API_URL}/batch-processing/v1/external/{workspace_url}/asset-library/jobs{suffix}"
+
+
+def _batch_processing_headers(api_key):
+    # Keep credentials out of URLs, proxy logs, and shell history. validateToken supports Bearer.
+    return {"Authorization": f"Bearer {api_key}"}
+
+
+def _raise_for_batch_processing_response(response):
+    message = response.text
+    try:
+        body = response.json()
+        if isinstance(body, dict):
+            error = body.get("error")
+            if isinstance(error, dict):
+                message = error.get("message") or error.get("hint") or message
+            elif error:
+                message = str(error)
+            else:
+                message = body.get("message") or message
+    except (TypeError, ValueError):
+        pass
+    raise RoboflowError(message, status_code=response.status_code)
+
+
+def create_asset_library_batch_job(
+    api_key,
+    workspace_url,
+    *,
+    workflow_id,
+    idempotency_key,
+    image_ids=None,
+    query=None,
+    machine_type="cpu",
+    display_name=None,
+):
+    """Queue a published Workflow over an exact Asset Library selection."""
+    payload = {
+        "workflowId": workflow_id,
+        "idempotencyKey": idempotency_key,
+        "machineType": machine_type,
+    }
+    if image_ids is not None:
+        payload["imageIds"] = image_ids
+    if query is not None:
+        payload["query"] = query
+    if display_name:
+        payload["displayName"] = display_name
+    response = requests.post(
+        _batch_processing_url(workspace_url),
+        headers=_batch_processing_headers(api_key),
+        json=payload,
+    )
+    if response.status_code != 202:
+        _raise_for_batch_processing_response(response)
+    return response.json()
+
+
+def list_batch_processing_jobs(api_key, workspace_url, *, page_size=10, next_page_token=None, search=None):
+    """List durable Batch Processing jobs in a workspace."""
+    params = {"pageSize": page_size}
+    if next_page_token:
+        params["nextPageToken"] = next_page_token
+    if search:
+        params["search"] = search
+    response = requests.get(
+        _batch_processing_url(workspace_url),
+        headers=_batch_processing_headers(api_key),
+        params=params,
+    )
+    if response.status_code != 200:
+        _raise_for_batch_processing_response(response)
+    return response.json()
+
+
+def get_batch_processing_job(api_key, workspace_url, job_id):
+    """Get current metadata for one Batch Processing job."""
+    encoded = quote(job_id, safe="")
+    response = requests.get(
+        _batch_processing_url(workspace_url, f"/{encoded}"),
+        headers=_batch_processing_headers(api_key),
+    )
+    if response.status_code != 200:
+        _raise_for_batch_processing_response(response)
+    return response.json()
+
+
+def abort_batch_processing_job(api_key, workspace_url, job_id):
+    """Abort one Batch Processing job."""
+    encoded = quote(job_id, safe="")
+    response = requests.post(
+        _batch_processing_url(workspace_url, f"/{encoded}/abort"),
+        headers=_batch_processing_headers(api_key),
+        json={},
+    )
+    if response.status_code != 200:
+        _raise_for_batch_processing_response(response)
+    return response.json()
+
+
+def restart_batch_processing_job(api_key, workspace_url, job_id):
+    """Restart one Batch Processing job with its existing configuration."""
+    encoded = quote(job_id, safe="")
+    response = requests.post(
+        _batch_processing_url(workspace_url, f"/{encoded}/restart"),
+        headers=_batch_processing_headers(api_key),
+        json={},
+    )
+    if response.status_code != 200:
+        _raise_for_batch_processing_response(response)
+    return response.json()
+
+
 # ---------------------------------------------------------------------------
 # Phase 2: Universe search
 # ---------------------------------------------------------------------------

roboflow/cli/__init__.py

@@ -213,7 +213,7 @@ def _walk(group: Any, prefix: str = "") -> None:
 app.add_typer(api_key_app, name="api-key")
 app.add_typer(asynctasks_app, name="asynctasks")
 app.add_typer(auth_app, name="auth")
-app.add_typer(batch_app, name="batch", hidden=True)  # All stubs — hidden until implemented
+app.add_typer(batch_app, name="batch")
 app.add_typer(completion_app, name="completion")
 app.add_typer(deployment_app, name="deployment")
 app.add_typer(device_app, name="device")