From 54670eab1a33eae58409690993aecdb9a0523071 Mon Sep 17 00:00:00 2001 From: binyangzhu000-sudo <224954946+binyangzhu000-sudo@users.noreply.github.com> Date: Tue, 18 Aug 2026 13:04:13 +0800 Subject: [PATCH] feat(skills): add Atlas Cloud image generation Signed-off-by: binyangzhu000-sudo <224954946+binyangzhu000-sudo@users.noreply.github.com> --- docs/README.skills.md | 1 + skills/atlas-cloud-image-generation/SKILL.md | 57 ++++ .../scripts/generate_image.py | 258 ++++++++++++++++++ .../tests/test_generate_image.py | 127 +++++++++ 4 files changed, 443 insertions(+) create mode 100644 skills/atlas-cloud-image-generation/SKILL.md create mode 100644 skills/atlas-cloud-image-generation/scripts/generate_image.py create mode 100644 skills/atlas-cloud-image-generation/tests/test_generate_image.py diff --git a/docs/README.skills.md b/docs/README.skills.md index aa492a833..ea18b8d61 100644 --- a/docs/README.skills.md +++ b/docs/README.skills.md @@ -59,6 +59,7 @@ See [CONTRIBUTING.md](../CONTRIBUTING.md#adding-skills) for guidelines on how to | [arize-trace](../skills/arize-trace/SKILL.md)
`gh skills install github/awesome-copilot arize-trace` | Downloads, exports, and inspects existing Arize traces and spans to understand what an LLM app is doing or debug runtime issues. Covers exporting traces by ID, spans by ID, sessions by ID, and root-cause investigation using the ax CLI. Use when the user wants to look at existing trace data, see what their LLM app is doing, export traces, download spans, investigate errors, or analyze behavior regressions. | `references/ax-profiles.md`
`references/ax-setup.md` | | [aspire](../skills/aspire/SKILL.md)
`gh skills install github/awesome-copilot aspire` | Aspire skill covering the Aspire CLI, AppHost orchestration, service discovery, integrations, MCP server, VS Code extension, Dev Containers, GitHub Codespaces, templates, dashboard, and deployment. Use when the user asks to create, run, debug, configure, deploy, or troubleshoot an Aspire distributed application. | `references/architecture.md`
`references/cli-reference.md`
`references/dashboard.md`
`references/deployment.md`
`references/integrations-catalog.md`
`references/mcp-server.md`
`references/polyglot-apis.md`
`references/testing.md`
`references/troubleshooting.md` | | [aspnet-minimal-api-openapi](../skills/aspnet-minimal-api-openapi/SKILL.md)
`gh skills install github/awesome-copilot aspnet-minimal-api-openapi` | Create ASP.NET Minimal API endpoints with proper OpenAPI documentation | None | +| [atlas-cloud-image-generation](../skills/atlas-cloud-image-generation/SKILL.md)
`gh skills install github/awesome-copilot atlas-cloud-image-generation` | Generate or edit images through the Atlas Cloud asynchronous image API. Use for prompt-to-image, single-image edits, and compositions with up to three local reference images; supports bounded polling and secure local downloads. | `scripts/generate_image.py`
`tests` | | [audit-integrity](../skills/audit-integrity/SKILL.md)
`gh skills install github/awesome-copilot audit-integrity` | Shared audit integrity framework for all AppSec agents — enforces output quality, intellectual honesty, and continuous improvement through anti-rationalization guards, self-critique loops, retry protocols, non-negotiable behaviors, self-reflection quality gates (1-10 scoring, ≥8 threshold), and a self-learning system with lesson/memory governance for security analysis agents. | `references/anti-rationalization-guard.md`
`references/clarification-protocol.md`
`references/non-negotiable-behaviors.md`
`references/retry-protocol.md`
`references/self-critique-loop.md`
`references/self-learning-system.md`
`references/self-reflection-quality-gate.md` | | [automate-this](../skills/automate-this/SKILL.md)
`gh skills install github/awesome-copilot automate-this` | Analyze a screen recording of a manual process and produce targeted, working automation scripts. Extracts frames and audio narration from video files, reconstructs the step-by-step workflow, and proposes automation at multiple complexity levels using tools already installed on the user machine. | None | | [autoresearch](../skills/autoresearch/SKILL.md)
`gh skills install github/awesome-copilot autoresearch` | Autonomous iterative experimentation loop for any programming task. Guides the user through defining goals, measurable metrics, and scope constraints, then runs an autonomous loop of code changes, testing, measuring, and keeping/discarding results. Inspired by Karpathy's autoresearch. USE FOR: autonomous improvement, iterative optimization, experiment loop, auto research, performance tuning, automated experimentation, hill climbing, try things automatically, optimize code, run experiments, autonomous coding loop. DO NOT USE FOR: one-shot tasks, simple bug fixes, code review, or tasks without a measurable metric. | None | diff --git a/skills/atlas-cloud-image-generation/SKILL.md b/skills/atlas-cloud-image-generation/SKILL.md new file mode 100644 index 000000000..27aa0c3d8 --- /dev/null +++ b/skills/atlas-cloud-image-generation/SKILL.md @@ -0,0 +1,57 @@ +--- +name: atlas-cloud-image-generation +description: 'Generate or edit images through the Atlas Cloud asynchronous image API. Use for prompt-to-image, single-image edits, and compositions with up to three local reference images; supports bounded polling and secure local downloads.' +metadata: + requires: + bins: + - python3 + env: + - ATLASCLOUD_API_KEY + primaryEnv: ATLASCLOUD_API_KEY +--- + +# Atlas Cloud Image Generation + +Generate or edit images with Atlas Cloud's native asynchronous API. The bundled script submits each generation once, polls the returned prediction with a bounded GET loop, and downloads completed images without forwarding the API key. + +## Generate an image + +```bash +python3 {baseDir}/scripts/generate_image.py \ + --prompt "A clean isometric game item icon on a solid background" \ + --size 1024x1024 \ + --filename item.png +``` + +The default generation model is `qwen-image-3.0/text-to-image`. + +## Edit or compose images + +Pass one to three local references. The script automatically selects `qwen-image-3.0/edit` when references are present. + +```bash +python3 {baseDir}/scripts/generate_image.py \ + --prompt "Keep the character identity and change the outfit to a raincoat" \ + --input-image character.png \ + --filename raincoat.png +``` + +Repeat `--input-image` for multi-image composition. Each reference must be PNG, JPEG, GIF, or WebP and no larger than 10 MiB. + +## Options + +- `--size WIDTHxHEIGHT` supports 512–2048 for generation and 512–1440 for edits. Omit it to let the model choose. +- `--count` requests 1–4 outputs. Multiple outputs use numbered filenames. +- `--negative-prompt` describes content to avoid. +- `--seed` accepts 0–2147483647 for reproducible generation. +- `--no-prompt-extend` disables prompt rewriting. +- `--model` overrides the mode-specific default when another model uses the same request schema. +- `--dry-run` prints the request payload without requiring an API key or sending a request. + +## Configuration + +Set `ATLASCLOUD_API_KEY` in the process environment. `ATLASCLOUD_BASE_URL` optionally overrides the default `https://api.atlascloud.ai/api/v1` endpoint for compatible deployments. + +Generated output URLs are temporary, so the script downloads them immediately. Downloads must be credential-free HTTPS, are limited to 64 MiB, do not follow redirects, and never receive the Atlas authorization header. + +If a request times out after submission, do not submit it again automatically because that can create a second billable generation. Use the prediction ID shown on stderr with `GET /model/prediction/{id}` to inspect the original task. diff --git a/skills/atlas-cloud-image-generation/scripts/generate_image.py b/skills/atlas-cloud-image-generation/scripts/generate_image.py new file mode 100644 index 000000000..927ad6397 --- /dev/null +++ b/skills/atlas-cloud-image-generation/scripts/generate_image.py @@ -0,0 +1,258 @@ +#!/usr/bin/env python3 +"""Generate or edit images through Atlas Cloud's asynchronous image API.""" + +from __future__ import annotations + +import argparse +import base64 +import ipaddress +import json +import mimetypes +import os +import re +import sys +import time +from pathlib import Path +from typing import Any +from urllib import error, parse, request + + +DEFAULT_BASE_URL = "https://api.atlascloud.ai/api/v1" +DEFAULT_GENERATION_MODEL = "qwen-image-3.0/text-to-image" +DEFAULT_EDIT_MODEL = "qwen-image-3.0/edit" +USER_AGENT = "awesome-copilot-atlas-image/1.0" +MAX_REFERENCE_BYTES = 10 * 1024 * 1024 +MAX_DOWNLOAD_BYTES = 64 * 1024 * 1024 +SIZE_PATTERN = re.compile(r"^(\d+)[x*](\d+)$") +SUPPORTED_MIME = { + "image/png": ".png", + "image/jpeg": ".jpg", + "image/gif": ".gif", + "image/webp": ".webp", +} + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--prompt", required=True, help="Prompt or editing instruction.") + parser.add_argument("--filename", required=True, help="Output filename or path.") + parser.add_argument("--input-image", action="append", default=[], help="Local reference image (repeatable, max 3).") + parser.add_argument("--model", help="Atlas model ID. Defaults according to whether references are present.") + parser.add_argument("--size", help="Output size as WIDTHxHEIGHT. Omit for automatic sizing.") + parser.add_argument("--count", type=int, default=1, help="Number of outputs (1-4).") + parser.add_argument("--negative-prompt", help="Content to avoid.") + parser.add_argument("--seed", type=int, help="Generation seed (0-2147483647).") + parser.add_argument("--no-prompt-extend", action="store_true", help="Disable automatic prompt rewriting.") + parser.add_argument("--poll-interval", type=float, default=3.0, help="Seconds between GET polls.") + parser.add_argument("--max-polls", type=int, default=100, help="Maximum prediction GET requests.") + parser.add_argument("--timeout", type=float, default=120.0, help="Per-request timeout in seconds.") + parser.add_argument("--dry-run", action="store_true", help="Print the request without sending it.") + return parser.parse_args() + + +def fail(message: str) -> None: + raise SystemExit(message) + + +def image_data_url(raw_path: str) -> str: + path = Path(raw_path) + if not path.is_file(): + fail(f"Input image not found: {path}") + size = path.stat().st_size + if size > MAX_REFERENCE_BYTES: + fail(f"Input image exceeds 10 MiB: {path}") + mime = mimetypes.guess_type(path.name)[0] + if mime == "image/jpg": + mime = "image/jpeg" + if mime not in SUPPORTED_MIME: + fail(f"Unsupported input image type: {path.suffix or 'unknown'}") + encoded = base64.b64encode(path.read_bytes()).decode("ascii") + return f"data:{mime};base64,{encoded}" + + +def normalize_size(raw_size: str | None, editing: bool) -> str | None: + if not raw_size: + return None + match = SIZE_PATTERN.fullmatch(raw_size) + if not match: + fail("--size must use WIDTHxHEIGHT, for example 1024x1024.") + width, height = (int(value) for value in match.groups()) + maximum = 1440 if editing else 2048 + if not (512 <= width <= maximum and 512 <= height <= maximum): + fail(f"--size dimensions must each be between 512 and {maximum} for this mode.") + return f"{width}*{height}" + + +def build_payload(args: argparse.Namespace) -> dict[str, Any]: + if not args.prompt.strip(): + fail("--prompt cannot be empty.") + if len(args.input_image) > 3: + fail("At most three --input-image values are supported.") + if not 1 <= args.count <= 4: + fail("--count must be between 1 and 4.") + if args.seed is not None and not 0 <= args.seed <= 2147483647: + fail("--seed must be between 0 and 2147483647.") + + editing = bool(args.input_image) + payload: dict[str, Any] = { + "model": args.model or (DEFAULT_EDIT_MODEL if editing else DEFAULT_GENERATION_MODEL), + "prompt": args.prompt, + "n": args.count, + "prompt_extend": not args.no_prompt_extend, + } + size = normalize_size(args.size, editing) + if size: + payload["size"] = size + if args.negative_prompt: + payload["negative_prompt"] = args.negative_prompt + if args.seed is not None: + payload["seed"] = args.seed + if editing: + payload["reference_image_urls"] = [image_data_url(path) for path in args.input_image] + return payload + + +def unwrap_response(payload: Any) -> dict[str, Any]: + if not isinstance(payload, dict): + fail("Atlas returned a non-object response.") + if payload.get("code") not in (None, 0, 200): + fail(f"Atlas API error: {payload.get('message') or payload.get('msg') or payload.get('code')}") + data = payload.get("data", payload) + if not isinstance(data, dict): + fail("Atlas returned an invalid data object.") + return data + + +def json_request(url: str, api_key: str, method: str, body: bytes | None, timeout: float) -> dict[str, Any]: + headers = {"Authorization": f"Bearer {api_key}", "User-Agent": USER_AGENT} + if body is not None: + headers["Content-Type"] = "application/json" + req = request.Request(url, data=body, method=method, headers=headers) + try: + with request.urlopen(req, timeout=timeout) as response: + return unwrap_response(json.loads(response.read().decode("utf-8"))) + except error.HTTPError as exc: + details = exc.read().decode("utf-8", errors="replace") + fail(f"Atlas request failed: HTTP {exc.code}\n{details}") + except error.URLError as exc: + fail(f"Atlas request failed: {exc.reason}") + + +def run_prediction(payload: dict[str, Any], api_key: str, base_url: str, args: argparse.Namespace) -> dict[str, Any]: + submit_url = f"{base_url}/model/generateImage" + prediction = json_request(submit_url, api_key, "POST", json.dumps(payload).encode("utf-8"), args.timeout) + prediction_id = prediction.get("id") + if not prediction_id: + fail("Atlas submission did not return a prediction ID.") + print(f"Prediction ID: {prediction_id}", file=sys.stderr) + + if args.max_polls < 1 or args.poll_interval < 0: + fail("--max-polls must be positive and --poll-interval cannot be negative.") + poll_url = f"{base_url}/model/prediction/{parse.quote(str(prediction_id), safe='')}" + for attempt in range(args.max_polls): + prediction = json_request(poll_url, api_key, "GET", None, args.timeout) + status = str(prediction.get("status", "")).lower() + if status in {"completed", "succeeded"}: + outputs = prediction.get("outputs") + if not isinstance(outputs, list) or not outputs: + fail("Atlas prediction completed without output URLs.") + return prediction + if status in {"failed", "canceled", "cancelled"}: + fail(f"Atlas prediction {status}: {prediction.get('error') or 'no details'}") + if attempt + 1 < args.max_polls: + time.sleep(args.poll_interval) + fail(f"Prediction {prediction_id} did not complete after {args.max_polls} polls. Do not resubmit automatically.") + + +class NoRedirectHandler(request.HTTPRedirectHandler): + def redirect_request(self, req: Any, fp: Any, code: int, msg: str, headers: Any, newurl: str) -> None: + return None + + +def validate_output_url(raw_url: str) -> str: + parsed = parse.urlparse(raw_url) + if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password: + fail("Atlas output URL must be credential-free HTTPS.") + try: + port = parsed.port + except ValueError: + fail("Atlas output URL contains an invalid port.") + if port not in (None, 443): + fail("Atlas output URL must use the default HTTPS port.") + hostname = parsed.hostname.lower() + if hostname in {"localhost", "localhost.localdomain"} or hostname.endswith(".localhost"): + fail("Atlas output URL cannot target localhost.") + try: + address = ipaddress.ip_address(hostname) + except ValueError: + pass + else: + if not address.is_global: + fail("Atlas output URL cannot target a non-public address.") + return raw_url + + +def detect_image(raw: bytes) -> tuple[str, str]: + if raw.startswith(b"\x89PNG\r\n\x1a\n"): + return "image/png", ".png" + if raw.startswith(b"\xff\xd8\xff"): + return "image/jpeg", ".jpg" + if raw.startswith((b"GIF87a", b"GIF89a")): + return "image/gif", ".gif" + if len(raw) >= 12 and raw[:4] == b"RIFF" and raw[8:12] == b"WEBP": + return "image/webp", ".webp" + fail("Downloaded output is not a recognized PNG, JPEG, GIF, or WebP image.") + + +def download_output(raw_url: str, timeout: float) -> tuple[bytes, str]: + url = validate_output_url(raw_url) + opener = request.build_opener(NoRedirectHandler()) + try: + req = request.Request(url, method="GET", headers={"User-Agent": USER_AGENT}) + with opener.open(req, timeout=timeout) as response: + declared = int(response.headers.get("Content-Length", "0") or "0") + if declared > MAX_DOWNLOAD_BYTES: + fail("Atlas output exceeds the 64 MiB download limit.") + raw = response.read(MAX_DOWNLOAD_BYTES + 1) + except error.HTTPError as exc: + fail(f"Atlas output download failed: HTTP {exc.code}") + except error.URLError as exc: + fail(f"Atlas output download failed: {exc.reason}") + if len(raw) > MAX_DOWNLOAD_BYTES: + fail("Atlas output exceeds the 64 MiB download limit.") + return raw, detect_image(raw)[1] + + +def output_path(filename: str, index: int, total: int, suffix: str) -> Path: + base = Path(filename) + stem = base.stem if base.suffix else base.name + name = f"{stem}-{index + 1}{suffix}" if total > 1 else f"{stem}{suffix}" + return base.with_name(name) + + +def main() -> int: + args = parse_args() + payload = build_payload(args) + base_url = os.environ.get("ATLASCLOUD_BASE_URL", DEFAULT_BASE_URL).rstrip("/") + if args.dry_run: + print(json.dumps({"url": f"{base_url}/model/generateImage", "request": payload}, indent=2)) + return 0 + + api_key = os.environ.get("ATLASCLOUD_API_KEY") + if not api_key: + fail("ATLASCLOUD_API_KEY is not set in the environment.") + prediction = run_prediction(payload, api_key, base_url, args) + outputs = prediction["outputs"] + for index, raw_url in enumerate(outputs): + raw, suffix = download_output(str(raw_url), args.timeout) + path = output_path(args.filename, index, len(outputs), suffix) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(raw) + resolved = path.resolve() + print(f"Saved image to: {resolved}") + print(f"MEDIA: {resolved}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/atlas-cloud-image-generation/tests/test_generate_image.py b/skills/atlas-cloud-image-generation/tests/test_generate_image.py new file mode 100644 index 000000000..189b53fa0 --- /dev/null +++ b/skills/atlas-cloud-image-generation/tests/test_generate_image.py @@ -0,0 +1,127 @@ +import argparse +import importlib.util +import json +import tempfile +import unittest +from pathlib import Path +from unittest import mock + + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "generate_image.py" + + +def load_module(): + spec = importlib.util.spec_from_file_location("atlas_generate_image", SCRIPT) + if spec is None or spec.loader is None: + raise RuntimeError("Unable to load generate_image.py") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class FakeResponse: + def __init__(self, payload): + self.payload = payload + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def read(self, *_args): + return json.dumps(self.payload).encode("utf-8") + + +class FakeImageResponse: + headers = {} + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def read(self, _limit): + return b"\x89PNG\r\n\x1a\nimage" + + +class AtlasImageTest(unittest.TestCase): + def setUp(self): + self.module = load_module() + + def args(self, **overrides): + values = { + "prompt": "A game item", + "filename": "item.png", + "input_image": [], + "model": None, + "size": "1024x1024", + "count": 1, + "negative_prompt": None, + "seed": None, + "no_prompt_extend": False, + "poll_interval": 0, + "max_polls": 2, + "timeout": 5, + "dry_run": False, + } + values.update(overrides) + return argparse.Namespace(**values) + + def test_generation_payload_matches_schema(self): + payload = self.module.build_payload(self.args(count=2, seed=7)) + self.assertEqual( + payload, + { + "model": "qwen-image-3.0/text-to-image", + "prompt": "A game item", + "n": 2, + "prompt_extend": True, + "size": "1024*1024", + "seed": 7, + }, + ) + + def test_edit_payload_embeds_local_reference(self): + with tempfile.TemporaryDirectory() as directory: + image = Path(directory) / "reference.png" + image.write_bytes(b"\x89PNG\r\n\x1a\nreference") + payload = self.module.build_payload(self.args(input_image=[str(image)])) + self.assertEqual(payload["model"], "qwen-image-3.0/edit") + self.assertEqual(payload["reference_image_urls"], ["data:image/png;base64,iVBORw0KGgpyZWZlcmVuY2U="]) + + def test_prediction_submits_once_then_only_polls_with_get(self): + submit = FakeResponse({"code": 200, "data": {"id": "prediction-1", "status": "created"}}) + processing = FakeResponse({"code": 200, "data": {"id": "prediction-1", "status": "processing"}}) + complete = FakeResponse({"code": 200, "data": {"id": "prediction-1", "status": "completed", "outputs": ["https://cdn.example/image.png"]}}) + with mock.patch.object(self.module.request, "urlopen", side_effect=[submit, processing, complete]) as urlopen: + result = self.module.run_prediction( + {"model": "qwen-image-3.0/text-to-image", "prompt": "test"}, + "test-key", + "https://api.atlascloud.ai/api/v1", + self.args(), + ) + self.assertEqual(result["status"], "completed") + self.assertEqual([call.args[0].method for call in urlopen.call_args_list], ["POST", "GET", "GET"]) + self.assertTrue(all(call.args[0].get_header("Authorization") == "Bearer test-key" for call in urlopen.call_args_list)) + + def test_output_url_rejects_credentials_and_private_ip(self): + for url in ("https://user:pass@example.com/image.png", "https://127.0.0.1/image.png", "http://example.com/image.png"): + with self.subTest(url=url), self.assertRaises(SystemExit): + self.module.validate_output_url(url) + + def test_output_download_does_not_forward_authorization(self): + opener = mock.Mock() + opener.open.return_value = FakeImageResponse() + with mock.patch.object(self.module.request, "build_opener", return_value=opener): + raw, suffix = self.module.download_output("https://cdn.example/image.png", 5) + request_object = opener.open.call_args.args[0] + self.assertIsNone(request_object.get_header("Authorization")) + self.assertEqual(request_object.get_header("User-agent"), "awesome-copilot-atlas-image/1.0") + self.assertEqual(raw, b"\x89PNG\r\n\x1a\nimage") + self.assertEqual(suffix, ".png") + + +if __name__ == "__main__": + unittest.main()