Workflow examples

These complete scripts use the Workflow API from the installed cgcsdk package. Create the files in your own project; access to CGC source or documentation repositories is not required.

Setup

python -m venv venv

Activate the environment and install the SDK:

# PowerShell
venv\Scripts\Activate.ps1
pip install cgcsdk

# Linux/macOS
source venv/bin/activate
pip install cgcsdk

Confirm the selected CGC context:

cgc context list

Each example below creates real CGC resources. Jobs finish automatically. Persistent volumes are not deleted automatically.

One Self-Contained CSV Job

One job creates a CSV in the container, processes three rows and checks the total of 4600 cents. No input file, GPU or PVC is required. Results are printed in logs; temporary files are removed.

job_workflow.py

job_workflow.py

#!/usr/bin/env python3
"""Run one CSV-processing job; temporary files are created inside its container.

Usage:
    python job_workflow.py --dry-run
    python job_workflow.py

No input file, PVC, GPU or extra packages are required. Results are printed in
job logs; temporary CSV files are removed before the job exits. A successful
job is deleted after its logs are retrieved. Failed jobs are left for diagnosis
and remain subject to the backend's finished-job retention policy.
"""

import argparse
import json

from cgc.sdk import App, Image, ShellCommand
from cgc.sdk.exceptions import SDKException


PROCESSING_CODE = r'''
import csv
import json
import tempfile
from pathlib import Path

with tempfile.TemporaryDirectory(prefix="cgc-csv-") as workdir:
    input_path = Path(workdir) / "input.csv"
    output_path = Path(workdir) / "processed.csv"
    rows = [
        {"product": "alpha", "quantity": 2, "unit_price_cents": 1000},
        {"product": "beta", "quantity": 3, "unit_price_cents": 700},
        {"product": "gamma", "quantity": 1, "unit_price_cents": 500},
    ]
    with input_path.open("w", newline="", encoding="utf-8") as stream:
        writer = csv.DictWriter(stream, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)

    with input_path.open(newline="", encoding="utf-8") as source:
        processed = [
            {
                "product": row["product"],
                "total_cents": int(row["quantity"]) * int(row["unit_price_cents"]),
            }
            for row in csv.DictReader(source)
        ]
    with output_path.open("w", newline="", encoding="utf-8") as stream:
        writer = csv.DictWriter(stream, fieldnames=["product", "total_cents"])
        writer.writeheader()
        writer.writerows(processed)

    with output_path.open(newline="", encoding="utf-8") as stream:
        saved = list(csv.DictReader(stream))
    expected = [("alpha", 2000), ("beta", 2100), ("gamma", 500)]
    actual = [(row["product"], int(row["total_cents"])) for row in saved]
    if actual != expected:
        raise RuntimeError("Processed CSV failed verification: " + repr(actual))
    summary = {"rows": len(saved), "total_cents": sum(total for _, total in actual)}

print("PROCESSING_RESULT " + json.dumps(summary), flush=True)
print("PROCESSING_OK", flush=True)
'''


app = App("csv-job-example", image=Image.from_registry("python:3.12-slim"))


@app.function(name="csv-process", cpu=1, memory=1, gpu=0)
def process_data():
    return ShellCommand(executable=("python", "-u", "-c", PROCESSING_CODE)).render()


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--dry-run", action="store_true", help="Print the plan without contacting CGC")
    parser.add_argument("--timeout", type=int, default=600, help="Job wait timeout in seconds")
    parser.add_argument("--poll-seconds", type=int, default=5)
    args = parser.parse_args(argv)
    if args.timeout <= 0 or args.poll_seconds <= 0:
        parser.error("--timeout and --poll-seconds must be positive")

    if args.dry_run:
        print(json.dumps(process_data.submit(dry_run=True), indent=2))
        return 0

    print("Creating one CSV job in the active CGC context; no GPU or PVC.")
    run = process_data.submit()
    print("Job:", run.name)
    try:
        status = run.wait(timeout_seconds=args.timeout, poll_seconds=args.poll_seconds)
        logs = run.logs(timestamps=True)
        text = logs if isinstance(logs, str) else json.dumps(logs)
        print(text)
        if status != "Succeeded" or "PROCESSING_OK" not in text:
            raise SDKException(-1, f"Job '{run.name}' did not pass verification. Status: {status}.")
    except Exception:
        print("Job left for diagnosis:", run.name)
        raise

    run.delete()
    print("CSV processing verified; successful job deleted:", run.name)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python job_workflow.py --timeout 600

After verifying the result, the script deletes its own job. Failure or timeout leaves it for diagnosis. Use a volume to persist files.

File-processing job

This example uses a public countries CSV by default and accepts optional arguments:

Argument	Meaning
--input-url	Public URL of the text or CSV file whose lines will be counted. Defaults to a small public countries CSV.
--volume	Optional CGC volume name used to persist the line-count file.
--volume-size	Volume capacity in GB when a missing volume is created. Default: 20.
--storage-class	Storage class for a new volume; otherwise uses the configured default.
--mount-path	Container path where the volume is mounted. Default: /data.
--output	Optional result file path. With --volume, it must be inside --mount-path.
--delete-job	Delete only a successful job after its logs have been retrieved.
--timeout-seconds	Maximum number of seconds to wait for the job. Default: 300.

It creates a short-lived job using one CPU core and 2 GB of memory.

View process_file_workflow.py

process_file_workflow.py

#!/usr/bin/env python3
"""Download a text/CSV file in a CGC workflow job and count its lines."""

import argparse
import json
import shlex
import time
from pathlib import PurePosixPath

from cgc.sdk import App, Image, Volume


DEFAULT_INPUT_URL = "https://raw.githubusercontent.com/cs109/2014_data/master/countries.csv"


def wait_for_job(run, timeout_seconds: int, poll_seconds: int = 5):
    deadline = time.time() + timeout_seconds
    last_status = None

    while time.time() < deadline:
        status = run.status()
        if status != last_status:
            print("Job status:", status)
            last_status = status
        if status in {"Succeeded", "Failed"}:
            return status
        time.sleep(poll_seconds)

    return "Timeout"


def print_job_logs(run):
    response = run.logs(timestamps=True)
    logs = response.get("details", {}).get("logs", {})

    if not logs:
        print("No logs returned.")
        return

    print("Job logs:")
    for pod_name, containers in logs.items():
        for container_name, text in containers.items():
            print(f"[{pod_name}/{container_name}]")
            print(text.rstrip() or "<empty>")


def create_process_file_task(volume=None):
    app = App(
        name="file-processing-workflow",
        image=Image.from_registry("python:3.11-slim"),
        volumes=[volume] if volume else [],
    )


    @app.function(
        name="count-file-lines",
        cpu=1,
        memory=2,
        ttl_seconds_after_finished=3600,
    )
    def process_file(input_url: str, output_path: str = ""):
        code = f"""
from pathlib import Path
from urllib.request import urlopen

input_url = {json.dumps(input_url)}
output_path = {json.dumps(output_path)}

with urlopen(input_url, timeout=60) as response:
    text = response.read().decode("utf-8")

line_count = len(text.splitlines())
print(f"Downloaded: {{input_url}}")
print(f"Line count: {{line_count}}")

if output_path:
    path = Path(output_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(str(line_count), encoding="utf-8")
    print(f"Saved line count to: {{path}}")
"""
        return f"python -c {shlex.quote(code)}"

    return process_file


def wait_for_volume(volume, timeout_seconds=300):
    deadline = time.monotonic() + timeout_seconds
    while time.monotonic() < deadline:
        if volume.ready():
            return
        time.sleep(5)
    raise RuntimeError(f"Volume '{volume.name}' did not become ready")


def parse_args():
    parser = argparse.ArgumentParser(
        description="Count lines in a remote text or CSV file using a CGC job."
    )
    parser.add_argument(
        "--input-url",
        default=DEFAULT_INPUT_URL,
        help="Public URL of the text or CSV file processed by the CGC job.",
    )
    parser.add_argument(
        "--volume",
        help="CGC volume name used to persist the result file.",
    )
    parser.add_argument(
        "--volume-size",
        type=int,
        default=20,
        help="Volume capacity in GB when a missing volume is created.",
    )
    parser.add_argument(
        "--storage-class",
        help="Storage class for a new volume; omit to use the configured default.",
    )
    parser.add_argument(
        "--mount-path",
        default="/data",
        help="Path where the volume is mounted inside the job container.",
    )
    parser.add_argument(
        "--output",
        help="Result file path. With a volume, it must be inside --mount-path.",
    )
    parser.add_argument(
        "--delete-job",
        action="store_true",
        help="Delete this job after successful completion and log retrieval.",
    )
    parser.add_argument(
        "--timeout-seconds",
        type=int,
        default=300,
        help="Maximum number of seconds to wait for the job.",
    )
    return parser.parse_args()


def main():
    args = parse_args()
    if args.timeout_seconds <= 0:
        raise ValueError("--timeout-seconds must be positive")
    if args.volume:
        mount = PurePosixPath(args.mount_path)
        if not mount.is_absolute() or ".." in mount.parts:
            raise ValueError("--mount-path must be an absolute container path without '..'")
    volume = None
    output_path = args.output or ""

    if args.volume:
        volume = Volume.from_name(
            name=args.volume,
            mount_path=args.mount_path,
            size=args.volume_size,
            storage_class=args.storage_class,
            create_if_missing=True,
        )
        if not output_path:
            output_path = str(PurePosixPath(args.mount_path) / "line-count.txt")
        if ".." in PurePosixPath(output_path).parts or not PurePosixPath(output_path).is_relative_to(
            PurePosixPath(args.mount_path)
        ):
            raise ValueError("--output must be inside --mount-path")
        volume.ensure()
        wait_for_volume(volume, args.timeout_seconds)
    elif output_path:
        print(
            "Warning: no volume was selected. The output file will disappear "
            "when the job container is removed."
        )

    process_file = create_process_file_task(volume)
    run = process_file.submit(args.input_url, output_path)
    print("Created job:", run.name)

    status = wait_for_job(run, timeout_seconds=args.timeout_seconds)

    print("Final status:", status)

    if status == "Timeout":
        print("The job did not finish before the timeout.")
        print("Check platform events with:")
        print(f"cgc events {run.name}")
    else:
        print_job_logs(run)

    if status != "Succeeded":
        raise RuntimeError(f"Job '{run.name}' ended with status {status}; left for diagnosis")

    if args.delete_job:
        run.delete()
        print("Deleted job:", run.name)


if __name__ == "__main__":
    main()

Run it:

python process_file_workflow.py

Persist the result on a CGC volume:

python process_file_workflow.py \
  --volume file-processing-data \
  --storage-class ssd-rwx-ec \
  --output /data/line-count.txt

--storage-class must name a class available in your environment; supply it when no default is configured.

Pass --input-url only when you want to use a different public text or CSV file.

PyTorch image-processing job

This workflow uses two files:

• process_image.py is uploaded to the image-processing-data volume.
• run_image_workflow.py creates jobs that run the uploaded script.

Processing script

View process_image.py

process_image.py

import argparse

import numpy as np
import torch
from PIL import Image


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--input",
        required=True,
        help="Path to the source JPEG or PNG inside the job container.",
    )
    parser.add_argument(
        "--output",
        required=True,
        help="Path where the processed image will be saved.",
    )
    args = parser.parse_args()

    image = Image.open(args.input).convert("RGB")
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    pixels = torch.from_numpy(np.array(image)).to(device)
    processed = 255 - pixels

    output = Image.fromarray(processed.cpu().numpy().astype(np.uint8))
    output.save(args.output)

    print(f"Processed image on {device} and saved it to {args.output}")


if __name__ == "__main__":
    main()

Start Filebrowser first if it is not already running:

cgc compute filebrowser create

The workflow script creates or reuses the image-processing-data volume, mounts it to Filebrowser, asks you to upload process_image.py and the source images, then submits the job. The workflow mounts the same volume at /data.

Workflow script

Argument	Meaning
--input	Input path under /data. Repeat for multiple images. Defaults to /data/input.jpg.
--output	Output path under /data. Repeat for multiple images. Defaults to /data/output.jpg.

When provided, the number and order of --input and --output values must match.

View run_image_workflow.py

run_image_workflow.py

import argparse
import shlex
import time
from pathlib import PurePosixPath

from cgc.sdk import App, Image, Volume, volume as volume_api
from cgc.sdk.exceptions import SDKException


VOLUME_NAME = "image-processing-data"
MOUNT_PATH = "/data"
FILEBROWSER_NAME = "filebrowser"


def is_mount_details(value):
    return (
        isinstance(value, dict)
        and value.get("pvc_name") == VOLUME_NAME
        and value.get("target_template_name") == FILEBROWSER_NAME
        and "mount_path" in value
    )


def get_mount_details_from_error(error):
    for value in error.args:
        if is_mount_details(value):
            return value
        if isinstance(value, dict) and is_mount_details(value.get("details")):
            return value["details"]
    return None


def prepare_volume():
    data = Volume.from_name(
        name=VOLUME_NAME,
        mount_path=MOUNT_PATH,
        size=2,
        storage_class="ssd-rwx-ec",
        create_if_missing=True,
    )
    data.ensure()
    deadline = time.monotonic() + 300
    while not data.ready():
        if time.monotonic() >= deadline:
            raise RuntimeError("Volume did not become ready: " + VOLUME_NAME)
        time.sleep(5)

    try:
        response = volume_api.volume_mount(
            name=VOLUME_NAME,
            target=FILEBROWSER_NAME,
        )
    except SDKException as error:
        details = get_mount_details_from_error(error)
        if details is None and error.reason == "PVC_MOUNT_ALREADY_MOUNTED":
            print("Volume is already mounted to Filebrowser.")
        elif details is not None:
            print(
                "Volume is mounted to Filebrowser. "
                f"Filebrowser mount path: {details['mount_path']}"
            )
        else:
            raise
    else:
        details = response.get("details", {})
        if response["code"] != 200 and not is_mount_details(details):
            raise RuntimeError(f"Could not mount volume to Filebrowser: {response}")
        if "mount_path" in details:
            print("Filebrowser mount path:", details["mount_path"])

    return data


def create_process_image_task(data):
    app = App(
        name="image-processing-workflow",
        image=Image.from_registry("pytorch/pytorch:2.7.1-cuda12.8-cudnn9-runtime"),
        volumes=[data],
    )

    @app.function(
        name="process-image",
        cpu=4,
        memory=16,
        gpu=1,
        gpu_type="nvidia-rtx-a5000",
    )
    def process_image(input_path, output_path):
        return (
            "test -f /data/process_image.py"
            f" && test -f {shlex.quote(input_path)}"
            " && python -m pip install --quiet Pillow"
            " && python /data/process_image.py"
            f" --input {shlex.quote(input_path)}"
            f" --output {shlex.quote(output_path)}"
        )

    return process_image

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--input",
        action="append",
        help="Input image path inside the job container. Repeat for multiple images.",
    )
    parser.add_argument(
        "--output",
        action="append",
        help="Output image path inside the job container. Repeat for multiple images.",
    )
    args = parser.parse_args()
    input_paths = args.input or ["/data/input.jpg"]
    output_paths = args.output or ["/data/output.jpg"]

    if len(input_paths) != len(output_paths):
        parser.error("Provide the same number of --input and --output values.")

    for path in input_paths + output_paths:
        candidate = PurePosixPath(path)
        if ".." in candidate.parts or not candidate.is_relative_to(MOUNT_PATH):
            parser.error("Input and output paths must be inside " + MOUNT_PATH)

    data = prepare_volume()
    process_image = create_process_image_task(data)

    print("Volume is ready and mounted to Filebrowser:", VOLUME_NAME)
    print("Open Filebrowser, open the mounted volume, and upload to its root:")
    print("- process_image.py")
    for input_path in input_paths:
        print(f"- {input_path.removeprefix(MOUNT_PATH + '/')}")
    print()
    print("Use this command to find the Filebrowser URL and app token:")
    print("cgc compute list -d")
    print("Sign in with username 'admin' and the displayed app token.")
    input("Press Enter after all files are uploaded...")

    runs = process_image.for_each(input_paths, output_paths)
    failed = []

    for run in runs:
        status = run.wait(timeout_seconds=600)
        print("Job name:", run.name)
        print("Final status:", status)
        print(run.logs(timestamps=True))

        if status != "Succeeded":
            failed.append(run.name)
        else:
            run.delete()
            print("Deleted successful job:", run.name)

    if failed:
        raise RuntimeError(f"Failed jobs: {', '.join(failed)}")


if __name__ == "__main__":
    main()

Run one image:

python run_image_workflow.py

This expects input.jpg in the mounted volume and writes output.jpg.

Run multiple images:

python run_image_workflow.py \
  --input /data/photo-a.jpg \
  --output /data/photo-a-processed.jpg \
  --input /data/photo-b.jpg \
  --output /data/photo-b-processed.jpg

Long-running HTTP service

The script declares text-analysis-service with @app.inference(). It returns a structured ShellCommand, deploys the resource and its api port, waits for HTTP readiness, sends a request to /analyze, and optionally deletes the resource. The application accepts both /health and /api/health because an ingress URL may already include the port name as its path prefix.

View text_inference.py

text_inference.py

import argparse

from cgc.sdk import App, Image, ShellCommand


RESOURCE_NAME = "text-analysis-service"
PORT = 8000
PORT_NAME = "api"

app = App(
    name="http-service-workflow",
    image=Image.from_registry("python:3.11-slim"),
)


@app.inference(
    entity="custom",
    name=RESOURCE_NAME,
    port=PORT,
    port_name=PORT_NAME,
    cpu=1,
    memory=2,
)
def text_analysis_service():
    code = r"""
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json


class Handler(BaseHTTPRequestHandler):
    def _send_json(self, status_code, payload):
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status_code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path in {"/health", "/api/health"}:
            self._send_json(200, {"status": "ok"})
            return
        self._send_json(404, {"error": "not found", "path": self.path})

    def do_POST(self):
        if self.path not in {"/analyze", "/api/analyze"}:
            self._send_json(404, {"error": "not found", "path": self.path})
            return

        length = int(self.headers.get("Content-Length", "0"))
        payload = json.loads(self.rfile.read(length) or b"{}")
        text = payload.get("text", "")
        words = text.split()
        self._send_json(
            200,
            {
                "input": text,
                "characters": len(text),
                "words": len(words),
                "uppercase": text.upper(),
                "first_five_words": words[:5],
            },
        )

    def log_message(self, format, *args):
        print(format % args, flush=True)


ThreadingHTTPServer(("0.0.0.0", 8000), Handler).serve_forever()
"""
    return ShellCommand(executable=("python", "-c", code))


def parse_args():
    parser = argparse.ArgumentParser(
        description="Deploy and call a long-running HTTP service on CGC."
    )
    parser.add_argument(
        "--text",
        default="CGC SDK workflows can run long-running HTTP services.",
        help="Text sent to the /analyze endpoint.",
    )
    parser.add_argument(
        "--wait-timeout",
        type=int,
        default=600,
        help="How long to wait for the resource and HTTP endpoint, in seconds.",
    )
    parser.add_argument(
        "--delete",
        action="store_true",
        help="Delete the CGC resource at the end of the script.",
    )
    return parser.parse_args()


def main():
    args = parse_args()
    endpoint = None
    if args.wait_timeout <= 0:
        raise ValueError("--wait-timeout must be positive")

    try:
        endpoint = text_analysis_service.deploy(
            wait=True,
            reuse_existing=False,
            resource_timeout_seconds=args.wait_timeout,
        )
        print("Endpoint:", endpoint.url)

        health = endpoint.wait_http(
            "/health",
            timeout_seconds=args.wait_timeout,
        ).json()
        print("Health:", health)

        payload = {"text": args.text}
        print("Sending text for processing:", payload)
        result = endpoint.post_json("/analyze", payload, timeout=30)
        print("Processed response:", result)
    except Exception:
        if endpoint is not None:
            print("Inspect resource logs with endpoint.logs(timestamps=True):", endpoint.name)
        raise
    finally:
        if args.delete and endpoint is not None:
            endpoint.delete()
            print(f"Deleted resource: {RESOURCE_NAME}")
        elif endpoint is not None:
            print(
                f"Resource left running: {RESOURCE_NAME}. "
                "Rerun with --delete when you want to remove it."
            )


if __name__ == "__main__":
    main()

Run it:

python text_inference.py --delete

Roboflow dataset download

Set ROBOFLOW_API_KEY before running this example. The key is not accepted as a command-line argument and should not be written into the script.

Argument	Meaning
--workspace-id	Roboflow workspace identifier.
--project-id	Roboflow project identifier.
--version	Numeric dataset version.
--format	Export format; defaults to yolov11.

The script creates or reuses the object-detection-data volume and preserves it after the download.

View roboflow_dataset.py

roboflow_dataset.py

import argparse
import os
import time

from cgc.sdk import Image, Volume
from cgc.sdk.recipes.datasets import RoboflowDataset
from cgc.sdk.recipes.yolo import YoloConfig, YoloWorkflow


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--workspace-id", required=True)
    parser.add_argument("--project-id", required=True)
    parser.add_argument("--version", required=True, type=int)
    parser.add_argument("--format", default="yolov11")
    args = parser.parse_args()

    try:
        api_key = os.environ["ROBOFLOW_API_KEY"]
    except KeyError as error:
        raise RuntimeError(
            "Set the ROBOFLOW_API_KEY environment variable before running."
        ) from error

    data = Volume.from_name(
        name="object-detection-data",
        mount_path="/root/data",
        size=50,
        storage_class="ssd-rwx-ec",
        create_if_missing=True,
    )
    data.ensure()
    deadline = time.monotonic() + 300
    while not data.ready():
        if time.monotonic() >= deadline:
            raise RuntimeError("Dataset volume did not become ready")
        time.sleep(5)

    dataset = RoboflowDataset(
        workspace_id=args.workspace_id,
        project_id=args.project_id,
        version=args.version,
        format=args.format,
        api_key=api_key,
    )

    workflow = YoloWorkflow(
        name="yolo-training",
        image=Image.from_registry("ultralytics/ultralytics:latest"),
        volume=data,
        config=YoloConfig(),
    )

    run = workflow.download(dataset).submit()
    status = run.wait(timeout_seconds=1800)
    print("Download status:", status)
    print(run.logs(timestamps=True))

    if status != "Succeeded":
        raise RuntimeError(f"Dataset download ended with status: {status}; job left for diagnosis: {run.name}")
    run.delete()
    print("Deleted successful download job:", run.name)


if __name__ == "__main__":
    main()

Run it:

python roboflow_dataset.py \
  --workspace-id your-workspace \
  --project-id your-project \
  --version 1

Pretrained model inference

The YOLO and RT-DETR scripts use a public image URL, print detections, and save the annotated result locally. Delete the endpoint when you no longer need it.

Script	Model	Output file
yolo_inference.py	yolo11n.pt	yolo_prediction.jpg
rtdetr_inference.py	rtdetr-l.pt	rtdetr_prediction.jpg

Both request one nvidia-rtx-a5000 GPU, four CPU cores, and 16 GB of memory.

View yolo_inference.py

yolo_inference.py

import base64
from pathlib import Path

from cgc.sdk import Image
from cgc.sdk.recipes.yolo import YoloConfig, YoloWorkflow


SAMPLE_IMAGE_URL = "https://ultralytics.com/images/bus.jpg"
OUTPUT_IMAGE_PATH = Path("yolo_prediction.jpg")
RESOURCE_NAME = "sdk-yolo-inference-demo"


def create_workflow():
    return YoloWorkflow(
        name=RESOURCE_NAME,
        image=Image.from_registry("ultralytics/ultralytics:latest"),
        config=YoloConfig(
            model_name="yolo11n.pt",
            inference_cpu=4,
            inference_memory=16,
            inference_gpu=1,
            inference_gpu_type="nvidia-rtx-a5000",
        ),
    )


def save_prediction_image(prediction: dict, path: Path = OUTPUT_IMAGE_PATH):
    image_bytes = base64.b64decode(prediction["annotated_image_base64"])
    path.write_bytes(image_bytes)
    return path


if __name__ == "__main__":
    workflow = create_workflow()
    print("Creating or reusing CGC resource and waiting for it to start...")
    endpoint = workflow.deploy_pretrained_inference(wait=True, reuse_existing=True)
    print("Endpoint:", endpoint.url)

    print("Waiting for inference service to become ready...")
    endpoint.wait_http("/health")

    print("Sending prediction request...")
    prediction = endpoint.post_json(
        "/predict",
        {"image_url": SAMPLE_IMAGE_URL},
        timeout=120,
    )

    print("Model:", prediction["model"])
    print("Detections:")
    for detection in prediction["detections"]:
        print(
            "- {class_name}: {confidence:.2f}".format(
                class_name=detection["class_name"],
                confidence=detection["confidence"],
            )
        )
    saved_path = save_prediction_image(prediction)
    print("Saved annotated image:", saved_path.resolve())

View rtdetr_inference.py

rtdetr_inference.py

import base64
from pathlib import Path

from cgc.sdk import Image
from cgc.sdk.recipes.rtdetr import RTDetrConfig, RTDetrWorkflow


SAMPLE_IMAGE_URL = "https://ultralytics.com/images/bus.jpg"
OUTPUT_IMAGE_PATH = Path("rtdetr_prediction.jpg")
RESOURCE_NAME = "sdk-rtdetr-inference-demo"


def create_workflow():
    return RTDetrWorkflow(
        name=RESOURCE_NAME,
        image=Image.from_registry("ultralytics/ultralytics:latest"),
        config=RTDetrConfig(
            model_name="rtdetr-l.pt",
            inference_cpu=4,
            inference_memory=16,
            inference_gpu=1,
            inference_gpu_type="nvidia-rtx-a5000",
        ),
    )


def save_prediction_image(prediction: dict, path: Path = OUTPUT_IMAGE_PATH):
    image_bytes = base64.b64decode(prediction["annotated_image_base64"])
    path.write_bytes(image_bytes)
    return path


if __name__ == "__main__":
    workflow = create_workflow()
    print("Creating or reusing CGC resource and waiting for it to start...")
    endpoint = workflow.deploy_pretrained_inference(wait=True, reuse_existing=True)
    print("Endpoint:", endpoint.url)

    print("Waiting for inference service to become ready...")
    endpoint.wait_http("/health")

    print("Sending prediction request...")
    prediction = endpoint.post_json(
        "/predict",
        {"image_url": SAMPLE_IMAGE_URL},
        timeout=120,
    )

    print("Model:", prediction["model"])
    print("Detections:")
    for detection in prediction["detections"]:
        print(
            "- {class_name}: {confidence:.2f}".format(
                class_name=detection["class_name"],
                confidence=detection["confidence"],
            )
        )
    saved_path = save_prediction_image(prediction)
    print("Saved annotated image:", saved_path.resolve())

Run either script:

python yolo_inference.py
python rtdetr_inference.py

For detailed explanations of every Workflow API argument, start with the Workflow overview.