diff --git a/README.md b/README.md index 5be5973..aabea84 100644 --- a/README.md +++ b/README.md @@ -177,7 +177,7 @@ Conventions: ## Current library -This snapshot contains **123 skills** across five categories. +This snapshot contains **124 skills** across five categories. Use `find agent-skills -name SKILL.md | sort` for the source of truth. @@ -201,9 +201,10 @@ Operational skills for repeatable Codex work: - `write-like-meng-on-x` - calibrate concise X drafts against an authored voice corpus. - `x-bookmark-quote-posts` - turn recent X bookmarks into source-backed quote-post drafts. -### Media (2) +### Media (3) -Image sourcing skills: +Image sourcing and generation skills: +- `atlas-image-generation` - generate original images through a live, schema-validated Atlas Cloud workflow. - `aura-asset-images` - use Aura Assets for stock-style design and marketing imagery. - `unsplash-asset-images` - pick high-quality Unsplash assets by use case, crop, and ratio. diff --git a/agent-skills/media/atlas-image-generation/SKILL.md b/agent-skills/media/atlas-image-generation/SKILL.md new file mode 100644 index 0000000..f4d8c93 --- /dev/null +++ b/agent-skills/media/atlas-image-generation/SKILL.md @@ -0,0 +1,130 @@ +--- +name: atlas-image-generation +description: Use when the user wants to generate and download an image through Atlas Cloud, including live image-model discovery, schema-validated request options, safe one-time submission, bounded result polling, and local artifact verification. +--- + +# Atlas Image Generation + +Generate an image through Atlas Cloud only when the user wants a new artifact. +Use the existing `aura-asset-images` or `unsplash-asset-images` skills when a +licensed stock image already solves the job. + +## Guardrails + +- Read `ATLASCLOUD_API_KEY` from the environment. Never print or persist it. +- Fetch the live model catalog before every generation. Do not rely on a + remembered model ID. +- Fetch the selected model's schema and send only fields it declares. +- Submit the generation POST exactly once. A timeout can still mean the paid + task was created, so never retry an ambiguous POST. +- Retry only catalog, schema, result, and output GET requests with a finite + bound. +- Download returned URLs promptly because provider outputs can expire. +- Stop before submission when the credential, model, schema, or user approval + is missing. + +## Workflow + +1. **Define the artifact.** Confirm the subject, composition, visual style, + aspect ratio, required text, brand constraints, and destination. +2. **Prefer reuse when appropriate.** Use a stock-image skill for generic + photography. Generate only when the brief needs an original composition or + visual treatment. +3. **Discover live models.** Run the bundled helper with `--list-models`. Pick + an exact image model that is visible in the current catalog. +4. **Inspect the request before spending.** Run `--dry-run` with the chosen + model, prompt, and options. The helper fetches the live schema and rejects + unsupported fields. +5. **Confirm cost-bearing work.** Show the non-secret request summary and ask + for confirmation when the user has not already approved generation. +6. **Submit once.** Run the same command without `--dry-run`. Do not rerun it if + submission times out or returns an unknown outcome. +7. **Poll and download.** Let the helper poll the schema-declared result path + with a finite limit, then download the first completed image. +8. **Verify the artifact.** Confirm the file exists, is non-empty, and is an + image before reporting success. Inspect it visually when composition or text + accuracy matters. + +## Helper + +Resolve the skill directory in the active agent environment, then list the +currently available image models: + +```bash +python3 "/scripts/generate_image.py" --list-models +``` + +Preview a schema-validated request without submitting it: + +```bash +python3 "/scripts/generate_image.py" \ + --model "" \ + --prompt "Editorial product scene, balanced composition, no text" \ + --param 'size="1024x1024"' \ + --output outputs/atlas-images/product-scene.png \ + --dry-run --json +``` + +After approval, remove `--dry-run` and keep the remaining arguments unchanged. +The helper supports: + +- `--prompt`, `--prompt-file`, or stdin +- `--param key=` for model-specific schema fields +- `--params-json '{...}'` for several model-specific fields +- `--poll-interval` and `--max-polls` for bounded polling +- `--output` for the downloaded image path +- `--json` for machine-readable output + +Resolve `` to this skill's installed directory before running the +command. Do not assume the user's project directory contains the bundled script. + +## Prompt Recipe + +Build the prompt in this order: + +```text +subject + action/state + environment + composition + lighting + material/style ++ camera/render treatment + intended use + exclusions +``` + +Keep requirements concrete: + +- Name the focal subject and its position. +- Specify negative space when the image will carry overlaid text. +- State `no text` when generated lettering is unwanted. +- Describe palette and material instead of using vague taste words. +- Separate must-have constraints from optional atmosphere. +- Avoid requesting logos, public figures, or copyrighted characters unless the + user has the rights and the provider permits the request. + +## Result States + +- **completed:** Return the request ID, verified local path, model ID, and final + prompt. +- **failed:** Return the provider status and sanitized error. Do not invent an + artifact. +- **polling timeout:** Return the request ID and unknown final status. Do not + create a replacement task. +- **ambiguous submission:** State that the POST outcome is unknown and stop. +- **blocked before submission:** Explain the missing key, schema, model, or + approval without making a paid request. + +## Anti-Patterns + +- Do not hardcode a model catalog into this skill. +- Do not copy parameters from another model's examples. +- Do not retry a POST because the client timed out. +- Do not expose the API key in command output, logs, commits, or URLs. +- Do not claim success from a request ID alone. +- Do not report a downloaded file without checking its bytes and media type. +- Do not generate a new image when a suitable existing asset is cheaper, + faster, and legally clearer. + +## Acceptance Checks + +- The selected ID came from the current live catalog. +- Every submitted option exists in the current model schema. +- Exactly one generation POST was attempted. +- Result polling stopped at a terminal state or the configured bound. +- The delivered local file is non-empty and has an image content type. +- The final response names any unresolved visual, rights, or expiry risk. diff --git a/agent-skills/media/atlas-image-generation/scripts/generate_image.py b/agent-skills/media/atlas-image-generation/scripts/generate_image.py new file mode 100755 index 0000000..1b968c2 --- /dev/null +++ b/agent-skills/media/atlas-image-generation/scripts/generate_image.py @@ -0,0 +1,399 @@ +#!/usr/bin/env python3 +"""Generate one Atlas Cloud image from a live model schema.""" + +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path +import sys +import time +from typing import Any, Iterable +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + + +DEFAULT_API_BASE = "https://api.atlascloud.ai" +TRANSIENT_HTTP = {429, 500, 502, 503, 504} +SUCCESS_STATUSES = {"completed", "succeeded", "success"} +FAILURE_STATUSES = {"failed", "canceled", "cancelled"} + + +class AtlasError(RuntimeError): + """A sanitized error that is safe to show to the user.""" + + +def walk_objects(value: Any) -> Iterable[dict[str, Any]]: + if isinstance(value, dict): + yield value + for child in value.values(): + yield from walk_objects(child) + elif isinstance(value, list): + for child in value: + yield from walk_objects(child) + + +def model_id(item: dict[str, Any]) -> str | None: + value = item.get("name") or item.get("id") or item.get("model") + return value if isinstance(value, str) and value else None + + +def image_models(catalog: Any) -> list[dict[str, Any]]: + found: dict[str, dict[str, Any]] = {} + for item in walk_objects(catalog): + current_id = model_id(item) + if not current_id: + continue + if str(item.get("type", "")).lower() != "image": + continue + if item.get("display_console") is not True: + continue + found[current_id] = item + return [found[key] for key in sorted(found)] + + +def choose_model(catalog: Any, requested_id: str) -> dict[str, Any]: + for item in image_models(catalog): + if model_id(item) == requested_id: + return item + raise AtlasError(f"Image model is not visible in the live catalog: {requested_id}") + + +def schema_url(item: dict[str, Any]) -> str: + value = item.get("schema") or item.get("schema_url") or item.get("input_schema") + if not isinstance(value, str) or not value: + raise AtlasError("Selected model does not expose a schema URL.") + return value + + +def safe_http_detail(error: HTTPError) -> str: + try: + raw = error.read().decode("utf-8", errors="replace") + except Exception: + return "" + try: + payload = json.loads(raw) + except json.JSONDecodeError: + return raw[:500] + if isinstance(payload, dict): + message = payload.get("message") or payload.get("error") or payload.get("detail") + if isinstance(message, str): + return message[:500] + if isinstance(message, dict): + nested = message.get("message") + if isinstance(nested, str): + return nested[:500] + return "API request failed" + + +def request_json( + method: str, + url: str, + *, + api_key: str | None = None, + payload: dict[str, Any] | None = None, + get_attempts: int = 4, + timeout: int = 60, +) -> Any: + method = method.upper() + attempts = get_attempts if method == "GET" else 1 + headers = {"Accept": "application/json", "User-Agent": "atlas-image-generation-skill/1.0"} + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + data = None + if payload is not None: + headers["Content-Type"] = "application/json" + data = json.dumps(payload).encode("utf-8") + + for attempt in range(attempts): + request = Request(url, data=data, method=method, headers=headers) + try: + with urlopen(request, timeout=timeout) as response: + return json.loads(response.read().decode("utf-8")) + except HTTPError as error: + retryable = method == "GET" and error.code in TRANSIENT_HTTP + if not retryable or attempt + 1 == attempts: + detail = safe_http_detail(error) + suffix = f": {detail}" if detail else "" + raise AtlasError(f"{method} {url} returned HTTP {error.code}{suffix}") from None + except (URLError, TimeoutError) as error: + if method != "GET" or attempt + 1 == attempts: + if method == "POST": + raise AtlasError("Generation POST outcome is ambiguous; it was not retried.") from None + reason = getattr(error, "reason", error) + raise AtlasError(f"GET {url} failed after bounded retries: {reason}") from None + time.sleep(2**attempt) + raise AtlasError(f"{method} {url} failed.") + + +def download_image(url: str, output: Path, *, get_attempts: int = 4) -> tuple[str, int]: + for attempt in range(get_attempts): + request = Request(url, headers={"User-Agent": "atlas-image-generation-skill/1.0"}) + try: + with urlopen(request, timeout=90) as response: + media_type = response.headers.get_content_type() + content = response.read() + if not media_type.startswith("image/"): + raise AtlasError(f"Output URL returned non-image content type: {media_type}") + if not content: + raise AtlasError("Output URL returned an empty file.") + output.parent.mkdir(parents=True, exist_ok=True) + output.write_bytes(content) + return media_type, len(content) + except HTTPError as error: + if error.code not in TRANSIENT_HTTP or attempt + 1 == get_attempts: + raise AtlasError(f"Image download returned HTTP {error.code}.") from None + except (URLError, TimeoutError) as error: + if attempt + 1 == get_attempts: + reason = getattr(error, "reason", error) + raise AtlasError(f"Image download failed after bounded retries: {reason}") from None + time.sleep(2**attempt) + raise AtlasError("Image download failed.") + + +def input_schema(schema: dict[str, Any]) -> dict[str, Any]: + try: + value = schema["components"]["schemas"]["Input"] + except (KeyError, TypeError): + raise AtlasError("Model schema does not define components.schemas.Input.") from None + if not isinstance(value, dict): + raise AtlasError("Model Input schema is not an object.") + return value + + +def validate_payload(payload: dict[str, Any], schema: dict[str, Any]) -> None: + spec = input_schema(schema) + required = set(spec.get("required") or []) + properties = spec.get("properties") or {} + allowed = set(properties) | required + missing = sorted(required - set(payload)) + unsupported = sorted(set(payload) - allowed) + if missing: + raise AtlasError(f"Request is missing required schema fields: {', '.join(missing)}") + if unsupported: + raise AtlasError(f"Request includes unsupported schema fields: {', '.join(unsupported)}") + for key, value in payload.items(): + field = properties.get(key) + if not isinstance(field, dict): + continue + enum = field.get("enum") + if isinstance(enum, list) and value not in enum: + choices = ", ".join(map(str, enum)) + raise AtlasError(f"{key} must be one of: {choices}") + + +def endpoint(schema: dict[str, Any], method: str, marker: str) -> str: + for path, methods in (schema.get("paths") or {}).items(): + if isinstance(methods, dict) and method.lower() in methods and marker in path: + return path + raise AtlasError(f"Model schema does not declare a {method.upper()} {marker} endpoint.") + + +def nested_id(value: Any) -> str | None: + for item in walk_objects(value): + current = item.get("id") + if isinstance(current, str) and current: + return current + return None + + +def status_record(value: Any) -> dict[str, Any]: + for item in walk_objects(value): + if isinstance(item.get("status"), str): + return item + return value if isinstance(value, dict) else {} + + +def output_url(value: Any) -> str | None: + for item in walk_objects(value): + outputs = item.get("outputs") + if isinstance(outputs, list): + for output in outputs: + if isinstance(output, str) and output.startswith(("https://", "http://")): + return output + for key in ("output", "url"): + candidate = item.get(key) + if isinstance(candidate, str) and candidate.startswith(("https://", "http://")): + return candidate + return None + + +def read_prompt(args: argparse.Namespace) -> str: + if args.prompt: + value = args.prompt + elif args.prompt_file: + value = Path(args.prompt_file).read_text(encoding="utf-8") + elif not sys.stdin.isatty(): + value = sys.stdin.read() + else: + raise AtlasError("Provide --prompt, --prompt-file, or prompt text on stdin.") + value = value.strip() + if not value: + raise AtlasError("Prompt is empty.") + return value + + +def parse_json_object(value: str, label: str) -> dict[str, Any]: + try: + parsed = json.loads(value) + except json.JSONDecodeError as error: + raise AtlasError(f"{label} must be valid JSON: {error}") from error + if not isinstance(parsed, dict): + raise AtlasError(f"{label} must decode to a JSON object.") + return parsed + + +def parse_params(values: list[str], params_json: str | None) -> dict[str, Any]: + result = parse_json_object(params_json, "--params-json") if params_json else {} + for item in values: + if "=" not in item: + raise AtlasError("--param must use key= syntax.") + key, raw = item.split("=", 1) + key = key.strip() + if not key: + raise AtlasError("--param key cannot be empty.") + try: + result[key] = json.loads(raw) + except json.JSONDecodeError as error: + raise AtlasError(f"--param {key} has invalid JSON: {error}") from error + return result + + +def build_payload(model: str, prompt: str, options: dict[str, Any]) -> dict[str, Any]: + reserved = sorted({"model", "prompt"} & set(options)) + if reserved: + raise AtlasError(f"Model options cannot override reserved fields: {', '.join(reserved)}") + return {"model": model, "prompt": prompt, **options} + + +def default_output() -> Path: + stamp = time.strftime("%Y%m%d-%H%M%S") + return Path("outputs") / "atlas-images" / f"atlas-image-{stamp}.png" + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Generate one Atlas Cloud image from a live model schema.") + parser.add_argument("--list-models", action="store_true", help="List visible live image models and exit.") + parser.add_argument("--model", help="Exact model ID returned by --list-models.") + parser.add_argument("--prompt", help="Image prompt.") + parser.add_argument("--prompt-file", help="UTF-8 prompt file.") + parser.add_argument("--param", action="append", default=[], help="Model option as key=.") + parser.add_argument("--params-json", help="JSON object of model-specific options.") + parser.add_argument("--output", help="Downloaded image path.") + parser.add_argument("--poll-interval", type=float, default=4.0, help="Seconds between result GETs.") + parser.add_argument("--max-polls", type=int, default=30, help="Maximum result GET attempts.") + parser.add_argument("--dry-run", action="store_true", help="Validate and print the request without POSTing.") + parser.add_argument("--json", action="store_true", help="Print machine-readable JSON.") + parser.add_argument("--api-base", default=os.getenv("ATLASCLOUD_MEDIA_API_BASE", DEFAULT_API_BASE)) + return parser + + +def emit(result: dict[str, Any], as_json: bool) -> None: + if as_json: + print(json.dumps(result, indent=2, sort_keys=True)) + return + for key, value in result.items(): + print(f"{key}: {value}") + + +def main() -> int: + args = build_parser().parse_args() + if args.max_polls < 1: + raise AtlasError("--max-polls must be at least 1.") + if args.poll_interval < 0: + raise AtlasError("--poll-interval cannot be negative.") + base = args.api_base.rstrip("/") + catalog = request_json("GET", f"{base}/api/v1/models") + visible_models = image_models(catalog) + + if args.list_models: + rows = [{"id": model_id(item), "schema": schema_url(item)} for item in visible_models] + if args.json: + print(json.dumps(rows, indent=2, sort_keys=True)) + else: + for row in rows: + print(f"{row['id']}\t{row['schema']}") + return 0 + + if not args.model: + raise AtlasError("Select an exact live image model with --model.") + prompt = read_prompt(args) + selected = choose_model(catalog, args.model) + schema = request_json("GET", schema_url(selected)) + options = parse_params(args.param, args.params_json) + payload = build_payload(args.model, prompt, options) + validate_payload(payload, schema) + post_path = endpoint(schema, "POST", "generateImage") + result_path = endpoint(schema, "GET", "{request_id}") + output = Path(args.output) if args.output else default_output() + + if args.dry_run: + emit( + { + "status": "dry-run", + "model": args.model, + "output": str(output), + "payload": payload, + "post_path": post_path, + "result_path": result_path, + }, + args.json, + ) + return 0 + + api_key = os.getenv("ATLASCLOUD_API_KEY") or os.getenv("ATLAS_CLOUD_API_KEY") + if not api_key: + raise AtlasError("Missing ATLASCLOUD_API_KEY in the environment.") + + submitted = request_json("POST", f"{base}{post_path}", api_key=api_key, payload=payload, timeout=90) + request_id = nested_id(submitted) + if not request_id: + raise AtlasError("Generation POST returned no request ID; it was not retried.") + + last_status = "unknown" + result: Any = None + for attempt in range(args.max_polls): + result = request_json( + "GET", + f"{base}{result_path.replace('{request_id}', request_id)}", + api_key=api_key, + ) + record = status_record(result) + last_status = str(record.get("status", "unknown")).lower() + if last_status in SUCCESS_STATUSES | FAILURE_STATUSES: + break + if attempt + 1 < args.max_polls: + time.sleep(args.poll_interval) + else: + raise AtlasError(f"Polling limit reached; request {request_id} has unknown final status.") + + if last_status not in SUCCESS_STATUSES: + record = status_record(result) + message = record.get("error") or record.get("message") or "provider reported failure" + raise AtlasError(f"Generation {request_id} ended with {last_status}: {message}") + + remote_url = output_url(result) + if not remote_url: + raise AtlasError(f"Generation {request_id} completed without an image URL.") + media_type, size_bytes = download_image(remote_url, output) + emit( + { + "status": "completed", + "request_id": request_id, + "model": args.model, + "output": str(output), + "media_type": media_type, + "bytes": size_bytes, + }, + args.json, + ) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except AtlasError as error: + print(f"error: {error}", file=sys.stderr) + raise SystemExit(1) diff --git a/agent-skills/media/atlas-image-generation/tests/test_generate_image.py b/agent-skills/media/atlas-image-generation/tests/test_generate_image.py new file mode 100644 index 0000000..294cfaa --- /dev/null +++ b/agent-skills/media/atlas-image-generation/tests/test_generate_image.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import importlib.util +import io +import json +from pathlib import Path +import sys +import unittest +from unittest.mock import patch +from urllib.error import HTTPError, URLError + + +SCRIPT = Path(__file__).parents[1] / "scripts" / "generate_image.py" +SPEC = importlib.util.spec_from_file_location("generate_image", SCRIPT) +assert SPEC and SPEC.loader +generate_image = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = generate_image +SPEC.loader.exec_module(generate_image) + + +class FakeHeaders: + def get_content_type(self) -> str: + return "application/json" + + +class FakeResponse: + def __init__(self, payload: object): + self.payload = payload + self.headers = FakeHeaders() + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, traceback): + return False + + def read(self) -> bytes: + return json.dumps(self.payload).encode() + + +class GenerateImageTests(unittest.TestCase): + def test_parser_ignores_chat_api_base_environment_variable(self): + with patch.dict( + generate_image.os.environ, + {"ATLASCLOUD_API_BASE": "https://chat.example/v1"}, + clear=True, + ): + parser = generate_image.build_parser() + self.assertEqual(parser.get_default("api_base"), generate_image.DEFAULT_API_BASE) + + def test_image_models_returns_only_visible_image_entries(self): + catalog = { + "data": [ + {"name": "visible/image", "type": "Image", "display_console": True, "schema": "https://schema"}, + {"name": "hidden/image", "type": "Image", "display_console": False, "schema": "https://schema"}, + {"name": "visible/video", "type": "Video", "display_console": True, "schema": "https://schema"}, + ] + } + self.assertEqual([generate_image.model_id(item) for item in generate_image.image_models(catalog)], ["visible/image"]) + + def test_validate_payload_uses_required_and_properties(self): + schema = { + "components": { + "schemas": { + "Input": { + "required": ["model", "prompt"], + "properties": {"prompt": {"type": "string"}, "size": {"enum": ["square"]}}, + } + } + } + } + generate_image.validate_payload({"model": "visible/image", "prompt": "test", "size": "square"}, schema) + with self.assertRaisesRegex(generate_image.AtlasError, "unsupported"): + generate_image.validate_payload({"model": "visible/image", "prompt": "test", "seed": 1}, schema) + with self.assertRaisesRegex(generate_image.AtlasError, "must be one of"): + generate_image.validate_payload({"model": "visible/image", "prompt": "test", "size": "wide"}, schema) + + def test_model_options_cannot_replace_model_or_prompt(self): + options = generate_image.parse_params([], '{"model":"other","size":"square"}') + with self.assertRaisesRegex(generate_image.AtlasError, "reserved fields: model"): + generate_image.build_payload("visible/image", "test", options) + + def test_post_network_failure_is_not_retried(self): + with patch.object(generate_image, "urlopen", side_effect=URLError("timeout")) as mocked: + with self.assertRaisesRegex(generate_image.AtlasError, "not retried"): + generate_image.request_json("POST", "https://api/generate", payload={"prompt": "test"}) + self.assertEqual(mocked.call_count, 1) + + def test_get_transient_failure_retries_with_bound(self): + error = HTTPError("https://api/result", 503, "busy", {}, io.BytesIO(b'{"message":"busy"}')) + with patch.object(generate_image, "urlopen", side_effect=[error, FakeResponse({"status": "completed"})]) as mocked: + with patch.object(generate_image.time, "sleep"): + result = generate_image.request_json("GET", "https://api/result", get_attempts=3) + self.assertEqual(result["status"], "completed") + self.assertEqual(mocked.call_count, 2) + + def test_endpoint_and_output_discovery_follow_schema_and_result(self): + schema = { + "paths": { + "/api/v1/model/generateImage": {"post": {}}, + "/api/v1/model/result/{request_id}": {"get": {}}, + } + } + self.assertEqual(generate_image.endpoint(schema, "POST", "generateImage"), "/api/v1/model/generateImage") + self.assertEqual( + generate_image.endpoint(schema, "GET", "{request_id}"), + "/api/v1/model/result/{request_id}", + ) + self.assertEqual(generate_image.output_url({"data": {"outputs": ["https://cdn/image.png"]}}), "https://cdn/image.png") + + +if __name__ == "__main__": + unittest.main()