feat(CLI): add tg batches commands for the batch API

src/together/lib/cli/__init__.py

@@ -31,6 +31,7 @@
     EVALS_HELP_EXAMPLES,
     FILES_HELP_EXAMPLES,
     MODELS_HELP_EXAMPLES,
+    BATCHES_HELP_EXAMPLES,
     JIG_LOGS_HELP_EXAMPLES,
     JIG_PUSH_HELP_EXAMPLES,
     ENDPOINTS_HELP_EXAMPLES,
@@ -47,9 +48,11 @@
     FILES_UPLOAD_HELP_EXAMPLES,
     BETA_CLUSTERS_HELP_EXAMPLES,
     MODELS_UPLOAD_HELP_EXAMPLES,
+    BATCHES_SUBMIT_HELP_EXAMPLES,
     BETA_ENDPOINTS_HELP_EXAMPLES,
     JIG_JOB_STATUS_HELP_EXAMPLES,
     JIG_SECRETS_SET_HELP_EXAMPLES,
+    BATCHES_DOWNLOAD_HELP_EXAMPLES,
     ENDPOINTS_CREATE_HELP_EXAMPLES,
     ENDPOINTS_UPDATE_HELP_EXAMPLES,
     BETA_ENDPOINTS_AB_HELP_EXAMPLES,
@@ -540,6 +543,23 @@ async def run_command() -> None:
 evals_app.command((f"{_CLI}.evals.retrieve:retrieve"), alias="get", help="Get eval job details")
 evals_app.command((f"{_CLI}.evals.status:status"), help="Get an eval job's status")
 
+## Batches API commands
+batches_app = app.command(App(name="batches", help="Submit and manage batch jobs", help_epilogue=BATCHES_HELP_EXAMPLES))
+batches_app.command(
+    (f"{_CLI}.batches.submit:submit"),
+    help="Submit a new batch job",
+    help_epilogue=BATCHES_SUBMIT_HELP_EXAMPLES,
+    sort_key=1,
+)
+batches_app.command((f"{_CLI}.batches.list:list"), alias="ls", help="List batch jobs")
+batches_app.command((f"{_CLI}.batches.retrieve:retrieve"), alias="get", help="Get batch job details")
+batches_app.command(
+    (f"{_CLI}.batches.download:download"),
+    help="Download batch job output (and error) files",
+    help_epilogue=BATCHES_DOWNLOAD_HELP_EXAMPLES,
+)
+batches_app.command((f"{_CLI}.batches.cancel:cancel"), help="Cancel a batch job")
+
 ## Telemetry API commands
 telemetry_app = app.command(App(name="telemetry", help="Configure CLI telemetry"))
 telemetry_app.command((f"{_CLI}.telemetry.status:status"), help="Show telemetry status")

src/together/lib/cli/api/batches/_utils.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from typing import Literal
+from typing_extensions import TypeAlias
+
+from together.lib.utils.tools import format_datetime
+from together.types.batch_job import BatchJob
+from together.lib.cli.utils._console import console
+
+BatchApiType: TypeAlias = Literal["chat.completions", "audio.transcriptions", "audio.translations"]
+BatchEndpoint: TypeAlias = Literal["/v1/chat/completions", "/v1/audio/transcriptions", "/v1/audio/translations"]
+
+API_TO_ENDPOINT: dict[BatchApiType, BatchEndpoint] = {
+    "chat.completions": "/v1/chat/completions",
+    "audio.transcriptions": "/v1/audio/transcriptions",
+    "audio.translations": "/v1/audio/translations",
+}
+
+ENDPOINT_TO_API: dict[str, BatchApiType] = {endpoint: api for api, endpoint in API_TO_ENDPOINT.items()}
+
+STATUS_COLORS = {
+    "VALIDATING": "yellow",
+    "IN_PROGRESS": "yellow",
+    "COMPLETED": "green",
+    "FAILED": "red",
+    "EXPIRED": "red",
+    "CANCELLED": "red",
+}
+
+_INCOMPLETE_STATUSES = frozenset({"VALIDATING", "IN_PROGRESS"})
+_DOWNLOADABLE_STATUSES = frozenset({"COMPLETED", "FAILED", "EXPIRED", "CANCELLED"})
+_PROGRESS_BAR_WIDTH = 20
+
+
+def format_endpoint(endpoint: str | None) -> str:
+    if not endpoint:
+        return ""
+    return ENDPOINT_TO_API.get(endpoint, endpoint)
+
+
+def format_status(status: str | None) -> str:
+    if not status:
+        return ""
+    color = STATUS_COLORS.get(status, "white")
+    return f"[bold {color}]{status.capitalize()}[/bold {color}]"
+
+
+def format_progress(progress: float) -> str:
+    pct = max(0.0, min(100.0, progress))
+    filled = round((pct / 100.0) * _PROGRESS_BAR_WIDTH)
+    bar = "█" * filled + "░" * (_PROGRESS_BAR_WIDTH - filled)
+    return f"[yellow]{bar}[/yellow] [bold]{pct:g}%[/bold]"
+
+
+def print_batch_detail(job: BatchJob) -> None:
+    """Print a curated human-readable view of a batch job."""
+    console.print("Batch job details:")
+    if job.created_at:
+        console.print(f" - Created at {format_datetime(job.created_at)}")
+    if job.status == "COMPLETED" and job.completed_at:
+        console.print(f" - Completed at {format_datetime(job.completed_at)}")
+
+    api = format_endpoint(job.endpoint)
+    if api:
+        console.print(f" - {api}")
+
+    if job.x_model_id:
+        console.print(f" - {job.x_model_id}")
+
+    if job.output_file_id:
+        console.print(f" - Output file ID {job.output_file_id}")
+
+    if job.error_file_id:
+        console.print(f" - [red]An error occurred[/red]")
+        console.print(f" - Error file ID {job.error_file_id}")
+
+    if job.error:
+        console.print(f" - [red]An error occurred[/red]")
+        console.print(job.error)
+
+    if job.status in _INCOMPLETE_STATUSES and job.progress is not None:
+        console.print(f"{format_progress(job.progress)} {format_status(job.status)}")
+
+    if job.status in _DOWNLOADABLE_STATUSES and job.id and (job.output_file_id or job.error_file_id):
+        console.print("\nDownload results with:")
+        console.print(f"[dim]-[/dim] [primary]tg batches download {job.id} --output ./out[/primary]")

src/together/lib/cli/api/batches/cancel.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from typing import Annotated
+
+from cyclopts import Parameter
+
+from together._utils._json import openapi_dumps
+from together.lib.cli.utils.config import CLIConfigParameter
+from together.lib.cli.utils._console import console
+from together.lib.cli.components.loader import show_loading_status
+from together.lib.cli.components.model_dump import print_model_dump
+
+
+async def cancel(
+    batch_id: Annotated[str, Parameter(help="The ID of the batch job to cancel")],
+    *,
+    config: CLIConfigParameter,
+) -> None:
+    """Cancel a batch job."""
+    response = await show_loading_status("Cancelling batch job...", config.client.batches.cancel(batch_id))
+    if config.json:
+        console.print_json(openapi_dumps(response).decode("utf-8"))
+        return
+
+    console.print("[green]√[/green] Cancelled batch job")
+    print_model_dump(response, show_nulls=False)

src/together/lib/cli/api/batches/download.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+import sys
+import base64
+from typing import Any, Optional, Annotated
+from pathlib import Path
+
+from cyclopts import Parameter, validators
+
+from together._utils._json import openapi_dumps
+from together.lib.cli.utils.config import CLIConfigParameter
+from together.lib.cli.utils._console import console
+from together.lib.cli.components.loader import show_loading_status
+from together.lib.cli.api.files.retrieve_content import download_file_content
+
+_TERMINAL_STATUSES = frozenset({"COMPLETED", "FAILED", "EXPIRED", "CANCELLED"})
+
+
+def _is_directory_output(path: Path) -> bool:
+    return path.is_dir() or path.suffix == ""
+
+
+def _error_output_path(output: Path) -> Path:
+    """Where to write the error file when *output* is a concrete file path."""
+    suffix = output.suffix or ".jsonl"
+    return output.with_name(f"{output.stem}.errors{suffix}")
+
+
+async def download(
+    id: Annotated[str, Parameter(help="The ID of the batch job")],
+    output: Annotated[
+        Optional[Path],
+        Parameter(
+            name=["--output", "-o"],
+            help="File or directory to save batch result files to; omit to print output to stdout",
+            validator=validators.Path(file_okay=True, dir_okay=True),
+        ),
+    ] = None,
+    *,
+    config: CLIConfigParameter,
+) -> None:
+    """Download output (and error) files for a batch job."""
+    job = await show_loading_status("Retrieving batch job...", config.client.batches.retrieve(id))
+    status = job.status or ""
+
+    if status not in _TERMINAL_STATUSES:
+        console.print(
+            f"[red]Batch job is not ready to download yet[/red] "
+            f"(status: {status or 'unknown'}). "
+            f"Check progress with [primary]tg batches get {id}[/primary]."
+        )
+        sys.exit(1)
+
+    if not job.output_file_id and not job.error_file_id:
+        console.print(f"[red]Batch job has no output or error files to download[/red] (status: {status}).")
+        sys.exit(1)
+
+    if output is not None:
+        saved: list[dict[str, str]] = []
+        directory_output = _is_directory_output(output)
+        error_file_id = job.error_file_id
+
+        if job.output_file_id:
+            out_path = await download_file_content(
+                config.client,
+                job.output_file_id,
+                output=output,
+                loading_message="Downloading batch output...",
+            )
+            assert isinstance(out_path, Path)
+            saved.append({"kind": "output", "id": job.output_file_id, "path": str(out_path)})
+        elif error_file_id and not directory_output:
+            # No output file — write the error file to the exact path the user asked for.
+            err_path = await download_file_content(
+                config.client,
+                error_file_id,
+                output=output,
+                loading_message="Downloading batch errors...",
+            )
+            assert isinstance(err_path, Path)
+            saved.append({"kind": "error", "id": error_file_id, "path": str(err_path)})
+            error_file_id = None
+
+        if error_file_id:
+            err_dest = output if directory_output else _error_output_path(output)
+            err_path = await download_file_content(
+                config.client,
+                error_file_id,
+                output=err_dest,
+                loading_message="Downloading batch errors...",
+            )
+            assert isinstance(err_path, Path)
+            saved.append({"kind": "error", "id": error_file_id, "path": str(err_path)})
+
+        if config.json:
+            console.print_json(openapi_dumps({"batch_id": id, "files": saved}).decode("utf-8"))
+            return
+
+        for item in saved:
+            label = "Output" if item["kind"] == "output" else "Errors"
+            console.print(f"[green]√[/green] {label} saved to [blue]{item['path']}[/blue]")
+        return
+
+    if not job.output_file_id:
+        console.print(
+            "[red]Batch job has no output file[/red]. "
+            "Use [primary]--output[/primary] to download the error file instead."
+        )
+        sys.exit(1)
+
+    raw = await download_file_content(
+        config.client,
+        job.output_file_id,
+        stdout=True,
+        loading_message="Downloading batch output...",
+    )
+    assert isinstance(raw, bytes)
+
+    if config.json:
+        try:
+            payload: dict[str, Any] = {
+                "batch_id": id,
+                "output_file_id": job.output_file_id,
+                "content": raw.decode("utf-8"),
+            }
+        except UnicodeDecodeError:
+            payload = {
+                "batch_id": id,
+                "output_file_id": job.output_file_id,
+                "content_base64": base64.b64encode(raw).decode("ascii"),
+            }
+        if job.error_file_id:
+            payload["error_file_id"] = job.error_file_id
+        console.print_json(openapi_dumps(payload).decode("utf-8"))
+        return
+
+    console.print(raw.decode("utf-8"))
+    if job.error_file_id:
+        console.print(f"\n[dim]Error file also available: tg batches download {id} --output ./out[/dim]")

src/together/lib/cli/api/batches/list.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from together._utils._json import openapi_dumps
+from together.lib.utils.tools import format_datetime
+from together.lib.cli.utils.config import CLIConfigParameter
+from together.lib.cli.utils._console import console
+from together.lib.cli.components.list import ListTable
+from together.lib.cli.components.loader import show_loading_status
+from together.lib.cli.api.batches._utils import STATUS_COLORS, format_endpoint
+from together.lib.cli.utils._mock_pagination import AfterParameter, mock_pagination
+
+
+async def list(
+    after: AfterParameter = None,
+    *,
+    config: CLIConfigParameter,
+) -> None:
+    """List batch jobs."""
+    response = await show_loading_status("Loading batch jobs...", config.client.batches.list())
+    jobs = response or []
+
+    epoch_start = datetime.fromtimestamp(0, tz=timezone.utc)
+    jobs.sort(key=lambda x: x.created_at or epoch_start, reverse=True)
+
+    jobs_to_display, next_cursor = mock_pagination(jobs, cursor_field="id", cursor=after)
+
+    if config.json:
+        console.print_json(openapi_dumps(jobs_to_display).decode("utf-8"))
+        return
+
+    table = ListTable(
+        empty_message="You don't have any batch jobs yet. To submit your first batch run:\n  [dim]-[/dim] [primary]tg batches submit[/primary]"
+    )
+    table.add_primary_column("ID")
+    table.add_column("API")
+    table.add_column("Model")
+    table.add_column("Status")
+    table.add_column("Created At")
+
+    for job in jobs_to_display:
+        status = str(job.status) if job.status is not None else ""
+        status_color = STATUS_COLORS.get(status, "white")
+        if job.status == "IN_PROGRESS" and job.progress is not None:
+            status = f"{status}: {job.progress:g}%"
+
+        table.add_row(
+            job.id or "",
+            format_endpoint(job.endpoint),
+            job.x_model_id or "",
+            f"[{status_color}]{status}[/{status_color}]",
+            format_datetime(job.created_at) if job.created_at else "",
+        )
+    console.print(table)
+    if next_cursor:
+        console.print("\n[blue dim]To display the next page, run:[/blue dim]")
+        console.print(f"  [dim]-[/dim] [white]tg batches list --after {next_cursor}[/white]")

src/together/lib/cli/api/batches/retrieve.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from typing import Annotated
+
+from cyclopts import Parameter
+
+from together._utils._json import openapi_dumps
+from together.lib.cli.utils.config import CLIConfigParameter
+from together.lib.cli.utils._console import console
+from together.lib.cli.components.loader import show_loading_status
+from together.lib.cli.api.batches._utils import print_batch_detail
+
+
+async def retrieve(
+    batch_id: Annotated[str, Parameter(help="The ID of the batch job")],
+    *,
+    config: CLIConfigParameter,
+) -> None:
+    """Get details of a batch job."""
+    response = await show_loading_status("Retrieving batch job...", config.client.batches.retrieve(batch_id))
+    if config.json:
+        console.print_json(openapi_dumps(response).decode("utf-8"))
+        return
+
+    print_batch_detail(response)