← Files AstriaARCHIVED FILE
bin/astria
104 KB · Sep 30, 2026 · 23:16 UTC
#!/usr/bin/env python3
"""astria — command-line client for the Astria API.
Credentials resolve in this order:
1. Environment variables (ASTRIA_API_KEY or ASTRIA_AUTH_TOKEN, ASTRIA_BASE_URL,
WORKSPACE_ID, GEMINI_TUNE_ID, SEEDREAM_TUNE_ID, GPT_IMAGE_2_TUNE_ID)
2. ~/.astria/config.json (written by `astria login`)
Profiles work like the AWS CLI: `ASTRIA_PROFILE=localhost` (or
`astria --profile localhost ...`) reads ~/.astria/config.localhost.json — a
separate API key, base URL and workspace. The Astria web agent injects the
environment variables instead, so it never needs a config file.
Only depends on Python 3 stdlib + `curl` (used for the actual HTTP).
"""
import argparse
import ast
import base64
import concurrent.futures
import hashlib
import json
import mimetypes
import os
import platform
import re
import shutil
import sqlite3
import subprocess
import sys
import threading
import time
import webbrowser
from pathlib import Path
from urllib.parse import urlencode, urlparse
VERSION = "1.19.0"
def _platform_token():
system = platform.system() or sys.platform
machine = platform.machine()
release = platform.release()
return "; ".join(part for part in [system, release, machine] if part)
def _client_token():
"""Identify the host environment for the User-Agent comment — an explicit
ASTRIA_CLIENT override, else the generic AI_AGENT marker, else Claude Code."""
val = (os.environ.get("ASTRIA_CLIENT")
or os.environ.get("AI_AGENT")
or ("claude-code" if os.environ.get("CLAUDECODE") else None))
if not val:
return None
return "".join(c for c in val if c.isalnum() or c in " ._/-")[:80]
def _user_agent():
comment = "; ".join(part for part in [
_platform_token(),
f"Python/{platform.python_version()}",
_client_token(),
] if part)
return f"astria-cli/{VERSION} ({comment})"
USER_AGENT = _user_agent()
DEFAULT_BASE_URL = "https://api.astria.ai"
UPGRADE_URL = "https://raw.githubusercontent.com/astriaai/cli/main/astria"
CONFIG_DIR = Path(os.environ.get("ASTRIA_CONFIG_DIR") or (Path.home() / ".astria"))
# CONFIG_PATH / MODELS_CACHE are profile-specific — finalised by apply_profile().
ACTIVE_PROFILE = "default"
CONFIG_PATH = CONFIG_DIR / "config.json"
MODELS_CACHE = CONFIG_DIR / "models.json"
# Offline fallback for the platform model tunes. The live ids come from the
# /cli/models endpoint (cached in MODELS_CACHE, refreshed daily) — these values
# are only used when that endpoint has never been reachable.
MODEL_TUNES = {
"gemini": "3618064",
"seedream": "3691308",
"gpt-image-2": "4665564",
}
MODEL_ENV = {
"gemini": "GEMINI_TUNE_ID",
"seedream": "SEEDREAM_TUNE_ID",
"gpt-image-2": "GPT_IMAGE_2_TUNE_ID",
}
MODELS_TTL_DEFAULT = 86400
# --------------------------------------------------------------------------
# errors / output
# --------------------------------------------------------------------------
class AstriaError(Exception):
pass
def die(msg, code=1):
print(f"astria: {msg}", file=sys.stderr)
sys.exit(code)
def emit(obj):
"""Print a result as JSON on stdout — skills parse this."""
json.dump(obj, sys.stdout, ensure_ascii=False, indent=2)
sys.stdout.write("\n")
# --------------------------------------------------------------------------
# config + credential resolution
# --------------------------------------------------------------------------
def resolve_profile(cli_value):
"""Profile precedence: --profile flag > ASTRIA_PROFILE env > 'default'."""
profile = cli_value or os.environ.get("ASTRIA_PROFILE") or "default"
if not profile.replace("-", "").replace("_", "").replace(".", "").isalnum():
die(f"invalid profile name {profile!r} — use letters, digits, '-', '_', '.'")
return profile
def argv_profile():
"""Peek at --profile in argv before the parser exists, so build_parser()
can read the right profile's models cache for the --help enum text."""
argv = sys.argv[1:]
for i, token in enumerate(argv):
if token == "--profile":
return argv[i + 1] if i + 1 < len(argv) else None
if token.startswith("--profile="):
return token.split("=", 1)[1]
return None
def normalize_meta_argv(argv):
"""Support command-style help/version aliases before argparse requires a verb."""
command_index = 0
while command_index < len(argv):
token = argv[command_index]
if token == "--profile":
command_index += 2
continue
if token.startswith("--profile="):
command_index += 1
continue
break
if command_index >= len(argv):
return argv
command = argv[command_index]
if command == "help":
return argv[:command_index] + argv[command_index + 1:] + ["--help"]
if command == "version" and command_index == len(argv) - 1:
return argv[:command_index] + ["--version"]
return argv
def apply_profile(profile):
"""Point CONFIG_PATH / MODELS_CACHE at this profile's files. The 'default'
profile keeps the original config.json / models.json names; a named profile
'X' uses config.X.json / models.X.json — like the AWS CLI's profiles."""
global ACTIVE_PROFILE, CONFIG_PATH, MODELS_CACHE
ACTIVE_PROFILE = profile
suffix = "" if profile == "default" else f".{profile}"
CONFIG_PATH = CONFIG_DIR / f"config{suffix}.json"
MODELS_CACHE = CONFIG_DIR / f"models{suffix}.json"
def load_config():
if CONFIG_PATH.is_file():
try:
return json.loads(CONFIG_PATH.read_text())
except (OSError, ValueError):
return {}
return {}
def save_config(cfg):
CONFIG_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
CONFIG_PATH.write_text(json.dumps(cfg, indent=2) + "\n")
CONFIG_PATH.chmod(0o600)
def api_key(cfg):
key = os.environ.get("ASTRIA_API_KEY") or os.environ.get("ASTRIA_AUTH_TOKEN") or cfg.get("api_key")
if not key:
die("not authenticated — run `astria login` (or set ASTRIA_API_KEY)")
return key
def base_url(cfg):
return (os.environ.get("ASTRIA_BASE_URL") or cfg.get("base_url") or DEFAULT_BASE_URL).rstrip("/")
def login_hint(cfg, override_base_url=None):
return f"{(override_base_url or base_url(cfg)).rstrip('/')}/users/edit/api"
def resolve_workspace(cfg, override):
"""Workspace scope: --workspace flag > env > config. `all` and `personal` pass through."""
ws = override or os.environ.get("ASTRIA_WORKSPACE_ID") or os.environ.get("WORKSPACE_ID") or cfg.get("workspace_id")
return str(ws) if ws not in (None, "", "personal") else None
def resolve_num_images(cfg, override, fallback):
value = override or cfg.get("default_num_images") or fallback
if not str(value).isdigit() or int(value) < 1:
die("default_num_images / --num-images must be a positive integer")
return str(value)
# --- model catalog (/cli/models, cached daily) ---
def load_models_cache():
if MODELS_CACHE.is_file():
try:
return json.loads(MODELS_CACHE.read_text())
except (OSError, ValueError):
return None
return None
def models_cache_stale(record, cfg):
if not record or not record.get("models"):
return True
if record.get("base_url") != base_url(cfg):
return True
return time.time() - record.get("fetched_at", 0) > record.get("ttl", MODELS_TTL_DEFAULT)
def fetch_models(cfg):
"""Fetch /cli/models and write the cache. Returns the record, or None on failure."""
try:
payload = request(cfg, "GET", "/cli/models", auth=False, no_workspace=True)
except AstriaError:
return None
if not isinstance(payload, dict) or not isinstance(payload.get("models"), dict):
return None
record = {
"base_url": base_url(cfg),
"fetched_at": int(time.time()),
"ttl": payload.get("cache_ttl") or MODELS_TTL_DEFAULT,
"default": payload.get("default", "gemini"),
"default_video_model": payload.get("default_video_model"),
"models": payload["models"],
"video_models": payload.get("video_models") or {},
"cli": payload.get("cli") or {},
"skills": payload.get("skills") or {},
}
try:
CONFIG_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
MODELS_CACHE.write_text(json.dumps(record, indent=2) + "\n")
except OSError:
pass
return record
def _version_tuple(v):
try:
return tuple(int(x) for x in str(v).split("."))
except (TypeError, ValueError):
return None
def install_cli_upgrade(url, target, *, force=False):
current = _version_tuple(VERSION)
try:
proc = subprocess.run(["curl", "-fsSL", "-A", USER_AGENT, url], capture_output=True)
except FileNotFoundError:
raise AstriaError("`curl` not found on PATH — install curl")
if proc.returncode != 0:
raise AstriaError(f"could not download upgrade from {url}: {proc.stderr.decode('utf-8', 'replace').strip()}")
try:
source = proc.stdout.decode("utf-8")
ast.parse(source)
except (UnicodeDecodeError, SyntaxError) as exc:
raise AstriaError(f"downloaded upgrade is not a valid astria CLI script: {exc}")
new_version = downloaded_cli_version(source)
downloaded = _version_tuple(new_version)
if current and downloaded and downloaded < current and not force:
raise AstriaError(f"refusing to downgrade astria {VERSION} to {new_version}; pass --force to install anyway")
if current and downloaded and downloaded == current and not force:
return None
if not target.is_file():
raise AstriaError(f"cannot upgrade {target}: executable file not found")
if not os.access(target, os.W_OK):
raise AstriaError(f"cannot write {target}; rerun with permissions to update that file")
tmp = target.with_name(f".{target.name}.upgrade-{os.getpid()}")
try:
mode = target.stat().st_mode & 0o777
tmp.write_bytes(proc.stdout)
tmp.chmod(mode or 0o755)
os.replace(tmp, target)
except OSError as exc:
raise AstriaError(f"could not install upgrade at {target}: {exc}")
finally:
try:
tmp.unlink()
except FileNotFoundError:
pass
return new_version
def auto_upgrade_cli(cli):
if os.environ.get("ASTRIA_NO_AUTO_UPGRADE") or os.environ.get("ASTRIA_UPGRADED"):
return False
upgrade = cli.get("upgrade") or "please upgrade the astria CLI"
target = current_executable_path()
if not target.is_file() or not os.access(target, os.W_OK):
print(f"astria: a newer CLI is available ({cli.get('latest_version')}) — {upgrade}", file=sys.stderr)
return True
try:
new_version = install_cli_upgrade(cli.get("download_url") or UPGRADE_URL, target)
except Exception as exc:
print(f"astria: auto-upgrade failed: {exc}; {upgrade}", file=sys.stderr)
return True
if not new_version:
return True
print(f"astria: upgraded {VERSION} -> {new_version}; restarting", file=sys.stderr)
os.environ["ASTRIA_UPGRADED"] = "1"
try:
os.execv(str(target), [str(target)] + sys.argv[1:])
except OSError as exc:
print(f"astria: restart after auto-upgrade failed: {exc}", file=sys.stderr)
return True
def check_cli_version(record, *, auto_upgrade=False):
"""Enforce the min/latest CLI version from the cached /cli/models `cli` block:
below min_version is a hard stop, below latest_version is an upgrade nudge."""
cli = (record or {}).get("cli") or {}
current = _version_tuple(VERSION)
if not current:
return
upgrade = cli.get("upgrade") or "please upgrade the astria CLI"
minimum = _version_tuple(cli.get("min_version"))
if minimum and current < minimum:
die(f"this astria CLI ({VERSION}) is no longer supported — {upgrade}")
latest = _version_tuple(cli.get("latest_version"))
if latest and current < latest:
if auto_upgrade and auto_upgrade_cli(cli):
return
print(f"astria: a newer CLI is available ({cli.get('latest_version')}) — {upgrade}",
file=sys.stderr)
def installed_skills_version():
root = os.environ.get("CLAUDE_PLUGIN_ROOT")
if not root:
return None
try:
data = json.loads((Path(root) / ".claude-plugin" / "plugin.json").read_text())
except (OSError, json.JSONDecodeError):
return None
return data.get("version")
def skills_upgrade_context(record):
current = installed_skills_version()
if not current:
return None
skills = (record or {}).get("skills") or {}
latest = skills.get("latest_version")
current_tuple = _version_tuple(current)
latest_tuple = _version_tuple(latest)
if not current_tuple or not latest_tuple or current_tuple >= latest_tuple:
return None
upgrade = skills.get("upgrade") or "reinstall the Astria skills plugin from astriaai/skills"
return (
f"The Astria skills plugin is outdated ({current} < {latest}). "
f"If the user asks to do anything with Astria, tell them to {upgrade}."
)
def downloaded_cli_version(source):
match = re.search(r'^VERSION = "([^"]+)"', source, re.MULTILINE)
return match.group(1) if match else None
def current_executable_path(override=None):
if override:
return Path(override).resolve()
candidate = Path(sys.argv[0])
if not candidate.is_absolute() and candidate.parent == Path("."):
found = shutil.which(sys.argv[0])
if found:
candidate = Path(found)
return candidate.resolve()
def model_catalog(cfg):
"""Model name -> tune id catalog: the cached /cli/models response (refreshed
daily), falling back to built-in MODEL_TUNES when the endpoint is unreachable."""
record = load_models_cache()
if models_cache_stale(record, cfg):
record = fetch_models(cfg) or record
if record and record.get("models"):
return record
return {
"default": "gemini",
"models": {name: {"tune_id": tid} for name, tid in MODEL_TUNES.items()},
"builtin": True,
}
def resolve_model(cfg, name):
"""Map a model name (gemini/seedream/...) or a raw tune id to a tune id."""
catalog = None
if name is None:
name = cfg.get("default_model")
if name is None:
catalog = model_catalog(cfg)
name = catalog.get("default", "gemini")
if str(name).isdigit():
return str(name)
key = str(name).lower()
# 1. environment override — the Astria web app injects these
if key in MODEL_ENV and os.environ.get(MODEL_ENV[key]):
return os.environ[MODEL_ENV[key]]
# 2. explicit per-model override in the config file
if cfg.get(f"{key}_tune_id"):
return str(cfg[f"{key}_tune_id"])
# 3. the server catalog (cached daily)
if catalog is None:
catalog = model_catalog(cfg)
entry = catalog.get("models", {}).get(key)
if entry and entry.get("tune_id"):
return str(entry["tune_id"])
# 4. built-in offline fallback
if key in MODEL_TUNES:
return MODEL_TUNES[key]
die(f"unknown model {name!r} — run `astria models` to list available models, or pass a numeric tune id")
# --------------------------------------------------------------------------
# HTTP — delegated to curl (handles redirects and retries)
# --------------------------------------------------------------------------
def request(cfg, method, path, *, query=None, form=None, json_body=None, workspace=None,
no_workspace=False, auth=True, timeout=120):
url = base_url(cfg) + path
if query:
url += ("&" if "?" in url else "?") + urlencode(query, doseq=True)
argv = ["curl", "-sS", "--retry", "2", "--max-time", str(timeout), "-A", USER_AGENT, "-X", method,
"-H", "Accept: application/json"]
if auth:
argv += ["-H", f"Authorization: Bearer {api_key(cfg)}"]
if not no_workspace:
ws = workspace if workspace is not None else resolve_workspace(cfg, None)
if ws:
argv += ["-H", f"X-Workspace-Id: {ws}"]
stdin = None
if json_body is not None:
argv += ["-H", "Content-Type: application/json", "--data-binary", "@-"]
stdin = json.dumps(json_body).encode()
for key, value in (form or []):
argv += ["--form-string", f"{key}={value}"]
argv += ["-w", "\n%{http_code}", url]
try:
proc = subprocess.run(argv, input=stdin, capture_output=True)
except FileNotFoundError:
die("`curl` not found on PATH — install curl")
if proc.returncode != 0:
raise AstriaError(f"network error ({method} {path}): {proc.stderr.decode('utf-8', 'replace').strip()}")
raw = proc.stdout.decode("utf-8", "replace")
body, _, status = raw.rpartition("\n")
try:
status = int(status)
except ValueError:
raise AstriaError(f"unexpected curl output for {method} {path}")
parsed = None
if body.strip():
try:
parsed = json.loads(body)
except ValueError:
parsed = None
if status == 426:
hint = (parsed.get("upgrade") or parsed.get("error")) if isinstance(parsed, dict) else None
die(f"this astria CLI ({VERSION}) is no longer supported — {hint or 'please upgrade to continue'}")
if status >= 400:
msg = None
if isinstance(parsed, dict):
msg = parsed.get("error") or parsed.get("message")
raise AstriaError(f"{method} {path} → HTTP {status}: {msg or body.strip()[:400]}")
return parsed if parsed is not None else body
def field(key, value):
"""A form field. Local attachments are represented by direct-upload signed ids."""
return (key, value)
def remote_url(value):
return str(value).startswith(("http://", "https://"))
def media_field(prefix, value, resolved):
"""Resolve an image/video argument to the right form field: URL vs signed id.
A URL goes in the `_url` variant of the field, with the suffix inserted
inside the param brackets — `prompt[input_image]` -> `prompt[input_image_url]`."""
if remote_url(value):
url_key = f"{prefix[:-1]}_url]" if prefix.endswith("]") else f"{prefix}_url"
return field(url_key, value)
return field(prefix, resolved)
DIRECT_UPLOAD_MAX_BYTES = 100 * 1024 * 1024
DIRECT_UPLOAD_WORKERS = 6
CONTENT_TYPE_OVERRIDES = {
".avif": "image/avif",
".heic": "image/heic",
".heif": "image/heif",
".mov": "video/quicktime",
".webp": "image/webp",
}
DIRECT_UPLOAD_CONTENT_TYPE_PREFIXES = ("image/", "video/", "audio/")
def file_content_type(path):
return (CONTENT_TYPE_OVERRIDES.get(path.suffix.lower())
or mimetypes.guess_type(path.name)[0]
or "application/octet-stream")
def file_md5(path):
digest = hashlib.md5()
with path.open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(chunk)
return base64.b64encode(digest.digest()).decode("ascii")
def direct_upload(cfg, value):
"""Upload one local file straight to Active Storage and return its signed id."""
return direct_uploads(cfg, [value])[0]
def prepare_direct_upload(value):
path = Path(value)
if not path.is_file():
die(f"file not found: {value}")
size = path.stat().st_size
if size > DIRECT_UPLOAD_MAX_BYTES:
die(f"file must be at most 100 MB: {value}")
content_type = file_content_type(path)
if content_type != "text/plain" and not content_type.startswith(DIRECT_UPLOAD_CONTENT_TYPE_PREFIXES):
die(f"unsupported upload content type {content_type}: {value}")
return path, size, content_type
def perform_direct_upload_record(cfg, prepared):
path, size, content_type = prepared
blob = request(cfg, "POST", "/api/direct_uploads", json_body={"blob": {
"filename": path.name,
"byte_size": size,
"checksum": file_md5(path),
"content_type": content_type,
}}, no_workspace=True)
upload = blob["direct_upload"]
argv = ["curl", "-sS", "--retry", "2", "--max-time", "600", "-X", "PUT"]
for key, header_value in upload["headers"].items():
argv += ["-H", f"{key}: {header_value}"]
argv += ["--data-binary", f"@{path}", "-w", "\n%{http_code}", upload["url"]]
try:
proc = subprocess.run(argv, capture_output=True)
except FileNotFoundError:
die("`curl` not found on PATH — install curl")
raw = proc.stdout.decode("utf-8", "replace")
body, _, status = raw.rpartition("\n")
if proc.returncode != 0 or not status.isdigit() or int(status) >= 400:
detail = proc.stderr.decode("utf-8", "replace").strip() or body.strip()[:400]
raise AstriaError(f"direct upload failed for {path}: {detail or status or 'unexpected response'}")
return blob
def perform_direct_upload(cfg, prepared):
return perform_direct_upload_record(cfg, prepared)["signed_id"]
def parallel_direct_upload(cfg, values, upload):
prepared = [prepare_direct_upload(value) for value in values]
if not prepared:
return []
with concurrent.futures.ThreadPoolExecutor(
max_workers=min(DIRECT_UPLOAD_WORKERS, len(prepared))) as executor:
return list(executor.map(lambda item: upload(cfg, item), prepared))
def direct_uploads(cfg, values):
"""Direct-upload local files in parallel and return signed ids in input order."""
return parallel_direct_upload(cfg, values, perform_direct_upload)
def direct_upload_records(cfg, values):
"""Direct-upload local files in parallel and retain signed ids + public URLs."""
return parallel_direct_upload(cfg, values, perform_direct_upload_record)
def direct_upload_handoff_records(cfg, paths):
"""Upload bounded handoff files; opaque skill assets travel as text/plain."""
prepared = []
for path in paths:
size = path.stat().st_size
content_type = file_content_type(path)
if content_type != "text/plain" and not content_type.startswith(DIRECT_UPLOAD_CONTENT_TYPE_PREFIXES):
content_type = "text/plain"
prepared.append((path, size, content_type))
if not prepared:
return []
with concurrent.futures.ThreadPoolExecutor(
max_workers=min(DIRECT_UPLOAD_WORKERS, len(prepared))) as executor:
return list(executor.map(lambda item: perform_direct_upload_record(cfg, item), prepared))
def resolve_media_values(cfg, values):
"""Replace local paths with signed ids while leaving remote URLs untouched."""
uploads = iter(direct_uploads(cfg, [value for value in values if not remote_url(value)]))
return [value if remote_url(value) else next(uploads) for value in values]
def image_reference_fields(values, resolved_values):
values = values or []
source_kinds = {
"url" if remote_url(value) else "file"
for value in values
}
if len(source_kinds) > 1:
die("--image-reference cannot mix URLs and local files in one request")
fields = []
for value, resolved in zip(values, resolved_values):
if source_kinds == {"url"}:
fields.append(field("prompt[image_reference_urls][]", value))
continue
fields.append(field("prompt[image_references][]", resolved))
return fields
# --------------------------------------------------------------------------
# polling
# --------------------------------------------------------------------------
WAIT_TIMEOUT_SEC = 600
WAIT_INTERVAL_SEC = 5
WAIT_INTERVAL_MIN = 2
WAIT_WORKERS = 6
def prompt_settled(prompt):
"""A prompt is settled once it has images OR a user_error — both are terminal."""
return bool(prompt.get("images")) or bool(prompt.get("user_error"))
def wait_prompts(cfg, ids, *, workspace=None, timeout=WAIT_TIMEOUT_SEC,
interval=WAIT_INTERVAL_SEC, on_settled=None):
"""Poll many prompt ids in parallel until each settles or the wall-clock
timeout fires. Returns {id: prompt_record_or_None}. on_settled(id, record)
fires once per id as soon as it settles, so callers can stream output."""
interval = max(WAIT_INTERVAL_MIN, interval)
deadline = time.monotonic() + timeout
settled = {}
pending = list(ids)
while pending and time.monotonic() < deadline:
round_start = time.monotonic()
queue = list(pending)
lock = threading.Lock()
results = {}
errors = {}
def pump():
while True:
with lock:
if not queue:
return
pid = queue.pop(0)
try:
results[pid] = request(cfg, "GET", f"/prompts/{pid}.json", workspace=workspace)
except AstriaError as exc:
# Transient — retry this id next round, don't mark settled.
errors[pid] = str(exc)
threads = [threading.Thread(target=pump) for _ in range(min(WAIT_WORKERS, len(queue)))]
for t in threads:
t.start()
for t in threads:
t.join()
still_pending = []
for pid in pending:
record = results.get(pid)
if record is not None and prompt_settled(record):
settled[pid] = record
if on_settled:
on_settled(pid, record)
else:
still_pending.append(pid)
pending = still_pending
if not pending:
break
# Sleep the remainder of the interval — but not past the deadline.
elapsed = time.monotonic() - round_start
nap = min(max(0.0, interval - elapsed), max(0.0, deadline - time.monotonic()))
if nap > 0:
time.sleep(nap)
for pid in pending:
settled.setdefault(pid, None)
return settled
def poll_prompt(cfg, tune_id, prompt_id, workspace=None):
"""Single-prompt wait used by `generate --wait` / `video --wait`. Delegates
to wait_prompts so both code paths share the same settle rules."""
results = wait_prompts(cfg, [str(prompt_id)], workspace=workspace,
timeout=WAIT_TIMEOUT_SEC, interval=WAIT_INTERVAL_SEC)
record = results.get(str(prompt_id))
if record is None:
raise AstriaError(f"prompt {prompt_id} did not finish within {WAIT_TIMEOUT_SEC}s")
return record
# --------------------------------------------------------------------------
# commands — auth
# --------------------------------------------------------------------------
def cmd_login(args, cfg):
key = args.api_key
if not key:
if sys.stdin.isatty():
try:
import getpass
key = getpass.getpass(f"Astria API key ({login_hint(cfg, args.base_url)}): ").strip()
except (EOFError, KeyboardInterrupt):
die("aborted")
else:
key = sys.stdin.readline().strip()
if not key:
die("no API key provided")
cfg["api_key"] = key
if args.base_url:
cfg["base_url"] = args.base_url.rstrip("/")
if args.workspace:
cfg["workspace_id"] = args.workspace
save_config(cfg)
try:
user = request(cfg, "GET", "/users", no_workspace=True)
who = user.get("name") or user.get("email") or "your account"
print(f"Logged in as {who} (profile '{ACTIVE_PROFILE}'). Config saved to {CONFIG_PATH}")
except AstriaError as exc:
die(f"saved key but verification failed: {exc}")
def cmd_logout(args, cfg):
if CONFIG_PATH.is_file():
CONFIG_PATH.unlink()
print(f"Removed {CONFIG_PATH}")
else:
print("Not logged in.")
def cmd_whoami(args, cfg):
user = request(cfg, "GET", "/users", no_workspace=True)
emit({k: user.get(k) for k in ("id", "name", "email", "usd_balance_mc", "purchased_plugin_at")})
def cmd_config(args, cfg):
if args.action == "get" or args.action is None:
emit({
"profile": ACTIVE_PROFILE,
"base_url": base_url(cfg),
"workspace_id": resolve_workspace(cfg, None),
"default_model": cfg.get("default_model", "gemini"),
"default_num_images": cfg.get("default_num_images"),
"config_path": str(CONFIG_PATH),
"authenticated": bool(os.environ.get("ASTRIA_API_KEY") or os.environ.get("ASTRIA_AUTH_TOKEN") or cfg.get("api_key")),
})
elif args.action == "set":
if not args.key or args.value is None:
die("usage: astria config set <key> <value>")
aliases = {"workspace": "workspace_id"}
allowed = {"base_url", "workspace_id", "workspace", "default_model", "default_num_images",
"gemini_tune_id", "seedream_tune_id", "gpt-image-2_tune_id"}
if args.key not in allowed:
die(f"unknown config key {args.key!r} — one of: {', '.join(sorted(allowed))}")
key = aliases.get(args.key, args.key)
if key == "default_num_images":
if not str(args.value).isdigit() or int(args.value) < 1:
die("default_num_images must be a positive integer")
cfg[key] = str(int(args.value))
else:
cfg[key] = args.value
save_config(cfg)
print(f"Set {key} = {cfg[key]}")
def cmd_models(args, cfg):
record = fetch_models(cfg) if args.refresh else model_catalog(cfg)
if record is None:
die("could not refresh models from the server")
emit({
"default": record.get("default", "gemini"),
"default_video_model": record.get("default_video_model"),
"source": "built-in (offline)" if record.get("builtin") else "server",
"note": ("this is the curated set of popular models only — any --model "
"also accepts a raw tune id directly. Browse every partner "
"model and its tune id at astria.ai/gallery/tunes?branch=partner-1 "
"(or run `astria tunes list --gallery --branch partner-1`)."),
"models": record.get("models", {}),
"video_models": record.get("video_models", {}),
})
def cmd_upgrade(args, cfg):
record = None if args.url else fetch_models(cfg)
record = record or load_models_cache() or {}
cli = record.get("cli") or {}
url = args.url or cli.get("download_url") or UPGRADE_URL
target = current_executable_path(args.path)
new_version = install_cli_upgrade(url, target, force=args.force)
if not new_version:
print(f"astria {VERSION} is already current at {target}")
return
print(f"Upgraded astria {VERSION} -> {new_version or 'unknown'} at {target}")
# --------------------------------------------------------------------------
# commands — tunes / references
# --------------------------------------------------------------------------
def cmd_tunes_list(args, cfg):
query = []
if args.title:
query.append(("title", args.title))
for name in args.name or []:
query.append(("name[]", name))
if args.limit:
query.append(("limit", args.limit))
if args.offset:
query.append(("offset", args.offset))
if args.model_type:
query.append(("model_type", args.model_type))
if args.branch:
query.append(("branch", args.branch))
path = "/gallery/tunes" if args.gallery else "/tunes"
tunes = request(cfg, "GET", path, query=query, workspace=args.workspace)
emit([{k: t.get(k) for k in ("id", "title", "name", "model_type", "orig_images")} for t in tunes])
def cmd_tunes_get(args, cfg):
emit(request(cfg, "GET", f"/tunes/{args.id}", workspace=args.workspace))
def cmd_tunes_create(args, cfg):
form = [
field("tune[title]", args.title),
field("tune[name]", args.name),
field("tune[model_type]", args.model_type),
field("tune[base_tune_id]", resolve_model(cfg, args.model)),
]
if args.description:
form.append(field("tune[characteristics]", json.dumps({"short_description": args.description})))
for url in args.image_url or []:
form.append(field("tune[image_urls][]", url))
for signed_id in direct_uploads(cfg, args.image or []):
form.append(field("tune[images][]", signed_id))
if not (args.image_url or args.image):
die("provide at least one --image-url or --image")
emit(request(cfg, "POST", "/tunes", form=form, workspace=args.workspace))
def cmd_tunes_update(args, cfg):
form = []
if args.title:
form.append(field("tune[title]", args.title))
if args.name:
form.append(field("tune[name]", args.name))
if args.description:
form.append(field("tune[characteristics]", json.dumps({"short_description": args.description})))
if not form:
die("nothing to update — pass --title, --name or --description")
emit(request(cfg, "PATCH", f"/tunes/{args.id}", form=form, workspace=args.workspace))
# --------------------------------------------------------------------------
# commands — prompts
# --------------------------------------------------------------------------
def cmd_prompts_list(args, cfg):
query = []
for key in ("text", "pack_id", "base_pack_id", "tune_id", "user_id", "orig_prompt_id", "order_id"):
value = getattr(args, key, None)
if value:
query.append((key, value))
for value in args.expand or []:
query.append(("expand[]", value))
for flag in ("liked", "today", "is_video", "is_api"):
if getattr(args, flag, False):
query.append((flag, "true"))
if args.limit:
query.append(("limit", args.limit))
if args.offset:
query.append(("offset", args.offset))
path = f"/tunes/{args.tune_id}/prompts" if args.tune_id else "/prompts"
prompts = request(cfg, "GET", path, query=query, workspace=args.workspace)
keys = ["id", "text", "num_images", "cost_mc", "images", "aspect_ratio", "resolution", "pack_id", "tune_id", "order_id"]
if args.expand:
keys += ["debug_images", "debug_filenames", "debug_content_types"]
emit([{k: p.get(k) for k in keys} for p in prompts])
def cmd_prompts_get(args, cfg):
tune_id = resolve_model(cfg, args.model) if args.model else args.tune
query = [("expand[]", value) for value in args.expand or []]
emit(request(cfg, "GET", f"/tunes/{tune_id}/prompts/{args.id}", query=query, workspace=args.workspace))
def cmd_prompts_update(args, cfg):
tune_id = resolve_model(cfg, args.model) if args.model else args.tune
form = []
if args.pack_id:
form.append(field("prompt[pack_id]", args.pack_id))
if args.base_pack_id:
form.append(field("prompt[base_pack_id]", args.base_pack_id))
if args.workspace_id:
form.append(field("prompt[workspace_id]", args.workspace_id))
if not form:
die("nothing to update — pass --pack-id, --base-pack-id or --workspace-id")
emit(request(cfg, "PATCH", f"/tunes/{tune_id}/prompts/{args.id}", form=form, workspace=args.workspace))
def cmd_prompts_wait(args, cfg):
ids = collect_prompt_ids(args)
if not ids:
die("no prompt ids — pass them as arguments, via --prompts-file, or on stdin")
out_lock = threading.Lock()
counts = {"done": 0, "failed": 0}
def on_settled(pid, record):
status = "done" if record.get("images") else "failed"
counts[status] += 1
line = {
"id": pid,
"images": record.get("images") or [],
"user_error": record.get("user_error"),
"status": status,
"settled_at": int(time.time()),
}
with out_lock:
json.dump(line, sys.stdout, ensure_ascii=False)
sys.stdout.write("\n")
sys.stdout.flush()
results = wait_prompts(cfg, ids, workspace=args.workspace,
timeout=args.timeout, interval=args.interval,
on_settled=on_settled)
timed_out = [pid for pid in ids if results.get(pid) is None]
summary = {
"total": len(ids),
"succeeded": counts["done"],
"failed": counts["failed"],
"timed_out": timed_out,
}
with out_lock:
json.dump(summary, sys.stdout, ensure_ascii=False)
sys.stdout.write("\n")
sys.stdout.flush()
if timed_out:
sys.exit(1)
# --------------------------------------------------------------------------
# commands — generate (image) / video
# --------------------------------------------------------------------------
def parse_reference_specs(values):
specs = []
for raw in values or []:
if "=" not in raw:
die("invalid --reference value: use NAME=PATH_OR_URL, e.g. --reference woman=./model.jpg")
name, image = raw.split("=", 1)
name = name.strip()
image = image.strip()
if not name or not image:
die("invalid --reference value: both NAME and PATH_OR_URL are required")
specs.append((name, image))
return specs
def reference_image_field(value, resolved):
if remote_url(value):
return field("tune[image_urls][]", value)
return field("tune[images][]", resolved)
def reference_title(name, image):
parsed = urlparse(str(image))
label = Path(parsed.path).stem if parsed.scheme else Path(image).stem
return label or name
def create_reference_mentions(args, cfg, tune_id, reference_specs, resolved_images):
mentions = []
for (name, image), resolved in zip(reference_specs, resolved_images):
form = [
field("tune[title]", reference_title(name, image)),
field("tune[name]", name),
field("tune[model_type]", "faceid"),
field("tune[base_tune_id]", tune_id),
reference_image_field(image, resolved),
]
tune = request(cfg, "POST", "/tunes", form=form, workspace=args.workspace)
mentions.append(f"<faceid:{tune['id']}:1> {tune['name']}")
return mentions
def prepend_reference_mentions(text, mentions):
if text is None:
return text
missing = [mention for mention in mentions if mention.split(" ", 1)[0] not in text]
return f"{' '.join(missing)} {text}".strip() if missing else text
def _generate(args, cfg, video):
reference_specs = parse_reference_specs(getattr(args, "references", None))
tune_id = resolve_model(cfg, args.model)
prompt_media = [
("prompt[input_image]", getattr(args, "input_image", None)),
("prompt[mask_image]", getattr(args, "mask_image", None)),
]
prompt_media = [(prefix, value) for prefix, value in prompt_media if value]
image_references = args.image_references or [] if video else []
source_kinds = {"url" if remote_url(value) else "file" for value in image_references}
if len(source_kinds) > 1:
die("--image-reference cannot mix URLs and local files in one request")
video_media = []
if video:
video_media = [
("prompt[video_first_frame]", args.first_frame),
("prompt[video_last_frame]", args.last_frame),
("prompt[input_video]", args.input_video),
("prompt[audio_reference]", args.audio_reference),
]
video_media = [(prefix, value) for prefix, value in video_media if value]
reference_images = [image for _, image in reference_specs]
media_values = (
reference_images
+ [value for _, value in prompt_media]
+ image_references
+ [value for _, value in video_media]
)
resolved_media = resolve_media_values(cfg, media_values)
cursor = 0
resolved_reference_images = resolved_media[cursor:cursor + len(reference_images)]
cursor += len(reference_images)
resolved_prompt_media = resolved_media[cursor:cursor + len(prompt_media)]
cursor += len(prompt_media)
resolved_image_references = resolved_media[cursor:cursor + len(image_references)]
cursor += len(image_references)
resolved_video_media = resolved_media[cursor:cursor + len(video_media)]
reference_mentions = create_reference_mentions(
args, cfg, tune_id, reference_specs, resolved_reference_images,
)
prompt_text = prepend_reference_mentions(args.text, reference_mentions)
form = []
if prompt_text is not None:
form.append(field("prompt[text]", prompt_text))
# Editing a single input image defaults to one result; text-to-image to two.
num_images = resolve_num_images(cfg, args.num_images, 1 if getattr(args, "input_image", None) else 2)
form.append(field("prompt[num_images]", num_images))
if getattr(args, "seed", None):
form.append(field("prompt[seed]", args.seed))
if getattr(args, "film_grain", None) is not None:
form.append(field("prompt[film_grain]",
"true" if args.film_grain else "false"))
if args.aspect_ratio:
form.append(field("prompt[aspect_ratio]", args.aspect_ratio))
if getattr(args, "resolution", None):
form.append(field("prompt[resolution]", args.resolution))
for (prefix, value), resolved in zip(prompt_media, resolved_prompt_media):
form.append(media_field(prefix, value, resolved))
if getattr(args, "pack_id", None):
form.append(field("prompt[pack_id]", args.pack_id))
if getattr(args, "base_pack_id", None):
form.append(field("prompt[base_pack_id]", args.base_pack_id))
if video:
form.append(field("prompt[video_model]", args.video_model))
form.append(field("prompt[video_prompt]",
prepend_reference_mentions(args.video_prompt, reference_mentions)))
form.extend(image_reference_fields(image_references, resolved_image_references))
if args.duration:
form.append(field("prompt[video_duration]", args.duration))
for (prefix, value), resolved in zip(video_media, resolved_video_media):
form.append(media_field(prefix, value, resolved))
if args.generate_audio is not None:
form.append(field("prompt[generate_audio]",
"true" if args.generate_audio else "false"))
prompt = request(cfg, "POST", f"/tunes/{tune_id}/prompts", form=form, workspace=args.workspace)
prompt = prompt[0] if isinstance(prompt, list) else prompt
if args.wait:
prompt = poll_prompt(cfg, tune_id, prompt["id"], workspace=args.workspace)
emit(prompt)
def cmd_generate(args, cfg):
if getattr(args, "video_prompt", None):
die("--video-prompt belongs to `astria video`, not `astria generate`. Use `astria video --video-prompt ...` for video generation.")
if not args.text:
die("--text is required for `astria generate`. For video generation without a first-frame prompt, use `astria video --video-prompt ...`.")
_generate(args, cfg, video=False)
def cmd_video(args, cfg):
if getattr(args, "input_image", None):
die("--input-image is for image generation/editing. For video, use --first-frame, --last-frame, --input-video, --image-reference, or --reference NAME=PATH_OR_URL.")
_generate(args, cfg, video=True)
VARIATE_VIDEO_MODEL = "seedance25_720p"
VARIATE_PROMPT_SCHEMA = {
"type": "object",
"properties": {
"video_prompt": {"type": "string"},
},
"required": ["video_prompt"],
"additionalProperties": False,
}
VARIATE_SYSTEM_MESSAGE = (
"Write one production-ready Seedance 2.5 video-to-video edit prompt. "
"Use the source video description as factual timing, action, staging, camera, lighting, "
"transitions, and audio context. Apply the edit brief without inventing changes it does not request. "
"When replacement reference tokens are supplied, include every token exactly once and state clearly "
"what each replaces. Do not mention the source description, brief, these instructions, or analysis. "
"Return only the requested JSON."
)
def parse_variate_reference_specs(values):
specs = []
for raw in values or []:
name = None
source = raw.strip()
if "=" in source:
candidate, separator, remainder = source.partition("=")
if (separator and candidate.strip() and remainder.strip()
and not remote_url(candidate.strip())
and "/" not in candidate and "\\" not in candidate):
name = candidate.strip()
source = remainder.strip()
if not source:
die("invalid --reference value: PATH_OR_URL cannot be empty")
specs.append({"name": name, "source": source})
return specs
def variate_reference_title(name, source, class_info):
parsed = urlparse(str(source))
filename = Path(parsed.path).name if parsed.scheme else Path(source).name
stem = Path(filename).stem
return (class_info.get("first_name") or class_info.get("short_description")
or stem or name or class_info.get("class_name") or "image")
def create_variate_reference(args, cfg, default_tune_id, spec, upload_record):
source = spec["source"]
class_info = {}
if not spec["name"]:
class_info = request(
cfg,
"POST",
"/images/class_name",
form=[field("file_url", source if remote_url(source) else upload_record["public_url"])],
workspace=args.workspace,
)
name = spec["name"] or class_info.get("class_name") or "image"
characteristics = {
key: value
for key, value in class_info.items()
if key not in ("class_name", "captions")
}
form = [
field("tune[title]", variate_reference_title(name, source, class_info)),
field("tune[name]", name),
]
if characteristics.get("short_description"):
form.append(field("tune[characteristics]", json.dumps(characteristics)))
form += [
field("tune[model_type]", "faceid"),
field("tune[base_tune_id]", default_tune_id),
reference_image_field(source, source if remote_url(source) else upload_record["signed_id"]),
field("tune[image_captions][]", class_info.get("caption", "")),
]
tune = request(cfg, "POST", "/tunes", form=form, workspace=args.workspace)
return {
"id": tune["id"],
"name": tune["name"],
"title": tune.get("title") or variate_reference_title(name, source, class_info),
"source": source,
"created": True,
}
def variate_prompt_messages(description, brief, references):
tokens = [f"<faceid:{reference['id']}:1>" for reference in references]
reference_tokens = "\n".join(tokens) or "(none)"
return [
{"role": "system", "content": VARIATE_SYSTEM_MESSAGE},
{
"role": "user",
"content": "\n\n".join([
f"Source video description:\n{description.strip()}",
f"Edit brief:\n{brief.strip() or '(none)'}",
f"Replacement reference tokens:\n{reference_tokens}",
]),
},
]
def compose_variate_video_prompt(args, cfg, description, references):
response = request(
cfg,
"POST",
"/assist/complete",
json_body={
"messages": variate_prompt_messages(description, args.brief, references),
"response_schema": VARIATE_PROMPT_SCHEMA,
},
workspace=args.workspace,
timeout=30,
)
try:
video_prompt = json.loads(response["text"])["video_prompt"].strip()
except (KeyError, TypeError, json.JSONDecodeError) as exc:
raise AstriaError("the Variate prompt writer returned invalid JSON") from exc
if not video_prompt:
raise AstriaError("the Variate prompt writer returned an empty video prompt")
missing_tokens = [
f"<faceid:{reference['id']}:1>"
for reference in references
if f"<faceid:{reference['id']}:1>" not in video_prompt
]
return (f"Replace objects in the scene with {' '.join(missing_tokens)}. {video_prompt}"
if missing_tokens else video_prompt)
INSPECT_TIMEOUT_SEC = 240
INSPECT_INTERVAL_SEC = 2
def inspect_video(cfg, payload, workspace=None):
"""POST /videos/inspect only enqueues the inspection (download, ffmpeg and
Gemini run on Astria's worker); poll GET /videos/inspect/<id> until it
settles. Servers that still answer synchronously return `description`
straight away and skip the loop."""
record = request(cfg, "POST", "/videos/inspect", json_body=payload,
workspace=workspace, timeout=125)
if not isinstance(record, dict):
raise AstriaError("unexpected response from POST /videos/inspect")
deadline = time.monotonic() + INSPECT_TIMEOUT_SEC
while record.get("status") in ("pending", "processing"):
if time.monotonic() > deadline:
raise AstriaError(f"video inspection {record.get('id')} did not finish within {INSPECT_TIMEOUT_SEC}s")
time.sleep(INSPECT_INTERVAL_SEC)
record = request(cfg, "GET", f"/videos/inspect/{record['id']}",
workspace=workspace, timeout=30)
if not isinstance(record, dict):
raise AstriaError("unexpected response from GET /videos/inspect")
if record.get("status") == "failed":
raise AstriaError(f"video inspection failed: {record.get('error') or 'unknown error'}")
if not str(record.get("description") or "").strip():
raise AstriaError("video inspection returned no description")
return record
def variate_description(args, cfg, source_payload):
if args.description is not None:
description = args.description.strip()
elif args.description_file:
path = Path(args.description_file)
if not path.is_file():
die(f"file not found: {args.description_file}")
description = path.read_text(encoding="utf-8").strip()
else:
inspected = inspect_video(cfg, source_payload, workspace=args.workspace)
description = inspected["description"].strip()
if not description:
die("video description cannot be empty")
return description
def cmd_variate(args, cfg):
if remote_url(args.source) and not args.source.startswith("https://"):
die("video URL must use HTTPS")
reference_specs = parse_variate_reference_specs(args.references)
existing_ids = []
for value in args.tune_ids or []:
if not str(value).isdigit():
die(f"invalid tune id: {value}")
tune_id = int(value)
if tune_id not in existing_ids:
existing_ids.append(tune_id)
if not args.brief.strip() and not reference_specs and not existing_ids:
die("provide --brief, --reference, or --tune-id")
catalog = model_catalog(cfg)
video_models = catalog.get("video_models")
if video_models is not None and VARIATE_VIDEO_MODEL not in video_models:
die(f"{VARIATE_VIDEO_MODEL} is not available for this account")
default_tune_id = resolve_model(cfg, None)
local_values = ([] if remote_url(args.source) else [args.source]) + [
spec["source"] for spec in reference_specs if not remote_url(spec["source"])
]
upload_records = iter(direct_upload_records(cfg, local_values))
source_record = None if remote_url(args.source) else next(upload_records)
reference_uploads = [
None if remote_url(spec["source"]) else next(upload_records)
for spec in reference_specs
]
source_payload = (
{"file_url": args.source}
if remote_url(args.source)
else {"blob_signed_id": source_record["signed_id"]}
)
description = variate_description(args, cfg, source_payload)
references = [
{"id": tune_id, "created": False}
for tune_id in existing_ids
]
references += [
create_variate_reference(args, cfg, default_tune_id, spec, upload_record)
for spec, upload_record in zip(reference_specs, reference_uploads)
]
video_prompt = compose_variate_video_prompt(args, cfg, description, references)
prompt_payload = {
"text": "",
"tune_id": int(default_tune_id),
"video_model": VARIATE_VIDEO_MODEL,
"video_prompt": video_prompt,
"video_audio": True,
"input_video_url" if remote_url(args.source) else "input_video": (
args.source if remote_url(args.source) else source_record["signed_id"]
),
}
if args.base_pack_id:
prompt_payload["base_pack_id"] = args.base_pack_id
if args.order_id:
prompt_payload["order_id"] = args.order_id
response = request(
cfg,
"POST",
"/prompts",
query=[("view", "board")],
json_body={"prompt": prompt_payload},
workspace=args.workspace,
)
prompt = response["prompt"]
if args.wait:
prompt = poll_prompt(cfg, default_tune_id, prompt["id"], workspace=args.workspace)
emit({
"description": description,
"references": references,
"video_prompt": video_prompt,
"prompt": prompt,
})
def cmd_inspect_video(args, cfg):
if str(args.source).startswith(("http://", "https://")):
if not str(args.source).startswith("https://"):
die("video URL must use HTTPS")
payload = {"file_url": args.source}
else:
payload = {"blob_signed_id": direct_upload(cfg, args.source)}
if args.tune_id:
payload["tune_ids"] = [int(tune_id) for tune_id in args.tune_id]
emit(inspect_video(cfg, payload, workspace=args.workspace))
def cmd_inspect(args, cfg):
if remote_url(args.source):
if not args.source.startswith("https://"):
die("image URL must use HTTPS")
image_url = args.source
else:
image_url = direct_upload_records(cfg, [args.source])[0]["public_url"]
form = [field("file_url", image_url)]
form.extend(field("names[]", name) for name in args.name or [])
result = request(
cfg,
"POST",
"/images/describe",
form=form,
workspace=args.workspace,
timeout=125,
)
if not isinstance(result, dict) or not str(result.get("description") or "").strip():
raise AstriaError("image inspection returned no description")
emit(result)
# --------------------------------------------------------------------------
# commands — packs / workspaces
# --------------------------------------------------------------------------
def cmd_packs_list(args, cfg):
query = []
if args.limit:
query.append(("limit", args.limit))
if args.offset:
query.append(("offset", args.offset))
packs = request(cfg, "GET", "/packs", query=query, workspace=args.workspace)
keys = ("id", "title", "slug", "subtitle", "main_class_name", "costs")
emit([{k: p.get(k) for k in keys} for p in packs])
def cmd_packs_get(args, cfg):
emit(request(cfg, "GET", f"/p/{args.slug}.json", workspace=args.workspace))
def cmd_packs_create(args, cfg):
payload = {"pack": {
"title": args.title,
"base_tune_id": int(resolve_model(cfg, args.model)),
"model_type": args.model_type,
}}
if args.subtitle:
payload["pack"]["subtitle"] = args.subtitle
emit(request(cfg, "POST", "/packs", json_body=payload, workspace=args.workspace))
def prompt_attribute_overrides(args):
"""Collect the per-prompt override flags into a prompt_attributes dict, sent
under tune[prompt_attributes][...]. Named flags cover the common knobs;
--attr KEY=VALUE is the escape hatch for any other prompt attribute."""
attrs = {}
if args.num_images:
attrs["num_images"] = args.num_images
if args.aspect_ratio:
attrs["aspect_ratio"] = args.aspect_ratio
if args.resolution:
attrs["resolution"] = args.resolution
if args.inpaint_faces is not None:
attrs["inpaint_faces"] = "true" if args.inpaint_faces else "false"
for raw in (args.attr or []):
if "=" not in raw:
die(f"--attr must be KEY=VALUE (got {raw!r})")
key, value = raw.split("=", 1)
attrs[key.strip()] = value.strip()
return attrs
def _id_list(value):
return [int(v) for v in str(value).split(",") if v.strip()] if value else []
def cmd_packs_run(args, cfg):
"""Run a pack (template) — POST /p/:slug/tunes — firing its template prompts.
Two ways to say *who* the pack runs on:
* --tune-id ID (repeatable) reuses existing tunes. Multi packs require at
least one; the server dispatches tune_ids to its multi handler.
* a training set (--title/--name/--image/--image-url) trains a fresh tune
from photos, then generates (regular packs).
Overrides (--brief, and the prompt_attributes flags) ride along on either path."""
tune_ids = [tid for raw in (args.tune_id or []) for tid in _id_list(raw)]
training = any((args.title, args.name, args.image, args.image_url))
if not tune_ids and not training:
die("nothing to run — pass --tune-id (existing tunes) "
"or a training set (--title/--name/--image)")
form = [field("tune[tune_ids][]", tid) for tid in tune_ids]
if args.title:
form.append(field("tune[title]", args.title))
if args.name:
form.append(field("tune[name]", args.name))
for signed_id in direct_uploads(cfg, args.image or []):
form.append(field("tune[images][]", signed_id))
for url in (args.image_url or []):
form.append(field("tune[image_urls][]", url))
if args.brief:
form.append(field("tune[brief]", args.brief))
for pid in _id_list(args.prompt_ids):
form.append(field("tune[prompt_ids][]", pid))
for key, value in prompt_attribute_overrides(args).items():
form.append(field(f"tune[prompt_attributes][{key}]", value))
emit(request(cfg, "POST", f"/p/{args.slug}/tunes", form=form, workspace=args.workspace))
def cmd_workspaces_list(args, cfg):
workspaces = request(cfg, "GET", "/workspaces.json", no_workspace=True)
emit([{k: w.get(k) for k in ("id", "title", "slug")} for w in workspaces])
def cmd_workspaces_create(args, cfg):
payload = {"workspace": {"title": args.title}}
emit(request(cfg, "POST", "/workspaces.json", json_body=payload, no_workspace=True))
# --------------------------------------------------------------------------
# commands — external agent handoff
# --------------------------------------------------------------------------
HANDOFF_MAX_MARKDOWN_BYTES = 256 * 1024
HANDOFF_MAX_FILE_BYTES = 10 * 1024 * 1024
HANDOFF_MAX_TOTAL_BYTES = 25 * 1024 * 1024
HANDOFF_MAX_FILES = 50
HANDOFF_SECRET_NAMES = {
".env", ".git", ".ssh", ".aws", "credentials.json", "id_rsa", "id_ed25519",
}
def handoff_source():
client = (_client_token() or "").lower()
if "claude" in client:
return "claude-code"
if "chatgpt" in client or "codex" in client:
return "chatgpt"
if os.environ.get("CODEX_HOME"):
return "chatgpt"
if "cursor" in client:
return "cursor"
return "other"
def handoff_sha256(path):
digest = hashlib.sha256()
with path.open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def safe_bundle_segment(value):
cleaned = re.sub(r"[^A-Za-z0-9._ -]", "_", value).strip(" .")
return cleaned or "file"
def validate_handoff_source(path):
if path.is_symlink():
die(f"handoff paths may not be symlinks: {path}")
lowered = {part.lower() for part in path.parts}
if lowered & HANDOFF_SECRET_NAMES or any(part.startswith(".env.") for part in lowered):
die(f"refusing to upload likely credential material: {path}")
def collect_handoff_path(value, bundle_root, kind, require_skill=False):
root = Path(value).expanduser()
if not root.exists():
die(f"handoff path not found: {value}")
validate_handoff_source(root)
if require_skill and (not root.is_dir() or not (root / "SKILL.md").is_file()):
die(f"skill directory must contain SKILL.md: {value}")
paths = [root] if root.is_file() else sorted(path for path in root.rglob("*") if path.is_file())
records = []
for path in paths:
validate_handoff_source(path)
relative = Path(path.name) if root.is_file() else path.relative_to(root)
destination = Path(bundle_root)
if root.is_dir():
destination /= safe_bundle_segment(root.name)
destination /= Path(*(safe_bundle_segment(part) for part in relative.parts))
records.append({
"local_path": path,
"path": destination.as_posix(),
"kind": kind,
})
return records
def cmd_agent_handoff(args, cfg):
workspace = resolve_workspace(cfg, args.workspace)
if not workspace or workspace == "all":
die("a workspace is required — pass --workspace or set WORKSPACE_ID")
handoff_path = Path(args.handoff).expanduser()
if not handoff_path.is_file():
die(f"handoff file not found: {args.handoff}")
validate_handoff_source(handoff_path)
handoff_bytes = handoff_path.read_bytes()
if len(handoff_bytes) > HANDOFF_MAX_MARKDOWN_BYTES:
die("handoff markdown must be at most 256 KB")
try:
handoff_markdown = handoff_bytes.decode("utf-8")
except UnicodeDecodeError:
die("handoff markdown must be UTF-8 text")
records = [
record
for value in (args.attach or [])
for record in collect_handoff_path(value, "files", "attachment")
] + [
record
for value in (args.skill or [])
for record in collect_handoff_path(value, "skills", "skill", require_skill=True)
]
if len(records) > HANDOFF_MAX_FILES:
die(f"a handoff may contain at most {HANDOFF_MAX_FILES} files")
if len({record["path"] for record in records}) != len(records):
die("two handoff files resolve to the same bundle path")
for record in records:
if record["local_path"].stat().st_size > HANDOFF_MAX_FILE_BYTES:
die(f"handoff file must be at most 10 MB: {record['local_path']}")
if sum(record["local_path"].stat().st_size for record in records) > HANDOFF_MAX_TOTAL_BYTES:
die("handoff files must total at most 25 MB")
uploads = direct_upload_handoff_records(cfg, [record["local_path"] for record in records])
files = [{
"path": record["path"],
"kind": record["kind"],
"signed_id": upload["signed_id"],
"sha256": handoff_sha256(record["local_path"]),
"executable": os.access(record["local_path"], os.X_OK),
} for record, upload in zip(records, uploads)]
result = request(cfg, "POST", "/api/agent_handoffs", json_body={"agent_handoff": {
"workspace_id": workspace,
"source": args.source or handoff_source(),
"title": args.title or handoff_path.stem.replace("_", " ").replace("-", " ").title() or "Agent handoff",
"handoff_markdown": handoff_markdown,
"files": files,
}}, workspace=workspace)
emit(result)
if args.open:
webbrowser.open(result["deep_link"])
# --------------------------------------------------------------------------
# commands — workspace landing page
# --------------------------------------------------------------------------
def cmd_landing_get(args, cfg):
ws = resolve_workspace(cfg, args.workspace)
if not ws:
die("a workspace is required — pass --workspace or set WORKSPACE_ID")
workspace = request(cfg, "GET", f"/workspaces/{ws}.json", workspace=ws)
if args.html:
sys.stdout.write((workspace.get("landing_page_html") or "") + "\n")
else:
emit(workspace)
def cmd_landing_set(args, cfg):
ws = resolve_workspace(cfg, args.workspace)
if not ws:
die("a workspace is required — pass --workspace or set WORKSPACE_ID")
# Send as a JSON body, not a multipart field. A multipart `@file` part is
# parsed server-side as an ActionDispatch::Http::UploadedFile, and assigning
# that to the string column stores its literal inspect string instead of the
# HTML. JSON keeps landing_page_html unambiguously a string.
if args.brief:
workspace = {"brief": args.brief}
else:
if not Path(args.html_file).is_file():
die(f"file not found: {args.html_file}")
workspace = {"landing_page_html": Path(args.html_file).read_text(encoding="utf-8")}
emit(request(cfg, "PATCH", f"/workspaces/{ws}.json",
json_body={"workspace": workspace}, workspace=ws))
# --------------------------------------------------------------------------
# commands — cache (local snapshot of tunes/prompts/packs/user)
# --------------------------------------------------------------------------
CACHE_FIELDS = {
"tunes": ("id", "title", "name", "orig_images"),
"prompts": ("id", "text", "num_images", "images", "aspect_ratio", "resolution", "input_image", "input_video", "order_id"),
"packs": ("id", "title", "slug", "main_class_name", "multiplier", "multiplier_api"),
"user": ("usd_balance_mc", "email", "name", "purchased_plugin_at"),
}
CACHE_MAX_AGE = 86400
def cache_dir(cfg, override):
ws = resolve_workspace(cfg, override) or "personal"
slug = "".join(c if c.isalnum() or c in "_.-" else "_" for c in ws)
root = Path(os.environ.get("ASTRIA_CACHE_DIR") or (Path.cwd() / ".cache"))
return root / f"ws_{slug}"
# Indexed columns mirrored from the full record into the SQLite store. Each
# table also gets a `json` TEXT column holding the complete record. These are
# the columns `astria cache get/find/uses` query against — no API calls.
DB_COLUMNS = {
"tunes": ("id", "name", "title"),
"prompts": ("id", "text", "pack_id", "tune_id", "order_id", "num_images", "aspect_ratio", "resolution"),
"packs": ("id", "title", "slug", "main_class_name"),
}
CACHE_DB_NAME = "cache.db"
def cache_fetch(cfg, resource, workspace, out_dir):
"""Fetch one resource, write its `<resource>.json` snapshot, and return the
raw API data so the caller can also load it into the SQLite store."""
endpoint = "/users" if resource == "user" else f"/{resource}"
data = request(cfg, "GET", endpoint, workspace=workspace, no_workspace=(workspace is None))
fields = CACHE_FIELDS[resource]
if isinstance(data, list):
shaped = [{k: row.get(k) for k in fields} for row in data]
else:
shaped = {k: data.get(k) for k in fields}
(out_dir / f"{resource}.json").write_text(json.dumps(shaped, ensure_ascii=False, indent=2))
return data
def cache_db_path(cfg, override):
return cache_dir(cfg, override) / CACHE_DB_NAME
def open_cache_db(cfg, override, *, require=False):
"""Open the local cache SQLite DB. With require=True, die() if it is missing
so query commands point the user at `astria cache refresh`."""
path = cache_db_path(cfg, override)
if require and not path.is_file():
die(f"no local cache at {path} — run `astria cache refresh` first")
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
return conn
def db_init_schema(conn):
"""(Re)create the tunes/prompts/packs tables and their indexes."""
for table, columns in DB_COLUMNS.items():
conn.execute(f"DROP TABLE IF EXISTS {table}")
cols = ", ".join(f"{c} INTEGER PRIMARY KEY" if c == "id" else f"{c} TEXT" for c in columns)
conn.execute(f"CREATE TABLE {table} ({cols}, json TEXT NOT NULL)")
for col in columns:
if col != "id":
conn.execute(f"CREATE INDEX idx_{table}_{col} ON {table} ({col})")
def db_load_resource(conn, resource, data):
"""Replace one table's rows from raw API data (a list of records)."""
columns = DB_COLUMNS[resource]
conn.execute(f"DELETE FROM {resource}")
rows = data if isinstance(data, list) else [data]
placeholders = ", ".join(["?"] * (len(columns) + 1))
conn.executemany(
f"INSERT OR REPLACE INTO {resource} ({', '.join(columns)}, json) VALUES ({placeholders})",
[tuple(row.get(c) for c in columns) + (json.dumps(row, ensure_ascii=False),) for row in rows],
)
def cmd_cache(args, cfg):
if args.action == "path":
print(cache_dir(cfg, args.workspace))
return
if args.action == "get":
cmd_cache_get(args, cfg)
return
if args.action == "find":
cmd_cache_find(args, cfg)
return
if args.action == "uses":
cmd_cache_uses(args, cfg)
return
if args.action == "stats":
cmd_cache_stats(args, cfg)
return
# action == "refresh"
workspace = resolve_workspace(cfg, args.workspace)
out_dir = cache_dir(cfg, args.workspace)
out_dir.mkdir(parents=True, exist_ok=True)
stamp = out_dir / "_refreshed_at"
if args.resource:
data = cache_fetch(cfg, args.resource, workspace, out_dir)
if args.resource in DB_COLUMNS:
conn = open_cache_db(cfg, args.workspace)
conn.execute(f"CREATE TABLE IF NOT EXISTS {args.resource} "
f"({', '.join(DB_COLUMNS[args.resource])}, json TEXT NOT NULL)")
db_load_resource(conn, args.resource, data)
conn.commit()
conn.close()
stamp.write_text(str(int(time.time())))
print(f"refreshed {args.resource} → {out_dir}")
return
if not args.force and stamp.is_file():
age = time.time() - int(stamp.read_text() or 0)
if age < CACHE_MAX_AGE:
print(f"cache fresh ({int(age)}s old) — {out_dir}")
return
errors = []
fetched = {}
def worker(resource):
try:
fetched[resource] = cache_fetch(cfg, resource, workspace, out_dir)
except AstriaError as exc:
errors.append(str(exc))
threads = [threading.Thread(target=worker, args=(r,)) for r in CACHE_FIELDS]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
if errors:
die("; ".join(errors))
conn = open_cache_db(cfg, args.workspace)
db_init_schema(conn)
for resource in DB_COLUMNS:
db_load_resource(conn, resource, fetched[resource])
conn.commit()
conn.close()
stamp.write_text(str(int(time.time())))
print(f"refreshed tunes, prompts, packs, user → {out_dir}")
def cmd_cache_get(args, cfg):
conn = open_cache_db(cfg, args.workspace, require=True)
row = conn.execute(f"SELECT json FROM {args.table} WHERE id = ?", (args.id,)).fetchone()
conn.close()
if row is None:
die(f"no {args.table} record with id {args.id} in the local cache")
emit(json.loads(row["json"]))
def cmd_cache_find(args, cfg):
conn = open_cache_db(cfg, args.workspace, require=True)
# filter -> (column, mode): 'eq' for exact match, 'like' for substring
filters = {
"tunes": {"name": ("name", "like"), "title": ("title", "like")},
"prompts": {"pack_id": ("pack_id", "eq"), "tune_id": ("tune_id", "eq"), "order_id": ("order_id", "eq"),
"text": ("text", "like")},
"packs": {"main_class": ("main_class_name", "like"), "title": ("title", "like")},
}[args.table]
clauses, params = [], []
for arg_name, (column, mode) in filters.items():
value = getattr(args, arg_name, None)
if value is None:
continue
if mode == "like":
clauses.append(f"{column} LIKE ?")
params.append(f"%{value}%")
else:
clauses.append(f"{column} = ?")
params.append(value)
where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
rows = conn.execute(f"SELECT json FROM {args.table}{where} ORDER BY id DESC", params).fetchall()
conn.close()
emit([json.loads(r["json"]) for r in rows])
# Tunes are referenced inside prompt text via tokens like <faceid:1234:1> or
# <lora:1234:0.8> — the id is the first colon-separated field after the type.
TUNE_TOKEN_RE = re.compile(r"<[a-z0-9_]+:(\d+):[^>]*>", re.IGNORECASE)
def cmd_cache_uses(args, cfg):
conn = open_cache_db(cfg, args.workspace, require=True)
rows = conn.execute("SELECT json FROM prompts WHERE text LIKE ?",
(f"%{args.tune_id}%",)).fetchall()
conn.close()
target = str(args.tune_id)
matches = [
json.loads(r["json"]) for r in rows
if target in TUNE_TOKEN_RE.findall(json.loads(r["json"]).get("text") or "")
]
emit(matches)
def cmd_cache_stats(args, cfg):
path = cache_db_path(cfg, args.workspace)
if not path.is_file():
die(f"no local cache at {path} — run `astria cache refresh` first")
conn = open_cache_db(cfg, args.workspace, require=True)
counts = {t: conn.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0] for t in DB_COLUMNS}
conn.close()
stamp = cache_dir(cfg, args.workspace) / "_refreshed_at"
age = int(time.time() - int(stamp.read_text() or 0)) if stamp.is_file() else None
emit({
"db_path": str(path),
"counts": counts,
"refreshed_age_seconds": age,
"stale": age is None or age > CACHE_MAX_AGE,
})
# --------------------------------------------------------------------------
# commands — download prompt images
# --------------------------------------------------------------------------
# Asset extensions we recognise from a URL path; anything else falls back to .jpg.
DOWNLOAD_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".gif", ".mp4", ".webm", ".mov"}
DOWNLOAD_WORKERS = 6
def asset_ext(url):
"""Derive a file extension from an asset URL's path, defaulting to .jpg."""
path = url.split("?", 1)[0].split("#", 1)[0]
ext = os.path.splitext(path)[1].lower()
return ext if ext in DOWNLOAD_EXTS else ".jpg"
def collect_prompt_ids(args):
"""Gather prompt ids from positional args, --prompts-file and stdin.
Deduped, original order preserved."""
raw = list(args.ids or [])
if args.prompts_file:
if not Path(args.prompts_file).is_file():
die(f"file not found: {args.prompts_file}")
raw += Path(args.prompts_file).read_text().split()
if not raw and not sys.stdin.isatty():
raw += sys.stdin.read().split()
seen, ids = set(), []
for token in raw:
token = token.strip()
if token and token not in seen:
seen.add(token)
ids.append(token)
return ids
def download_asset(url, dest):
"""Fetch one asset URL to dest with curl. Raises AstriaError on failure."""
argv = ["curl", "-sSL", "--fail", "--max-time", "120", "-A", USER_AGENT, "-o", str(dest), url]
try:
proc = subprocess.run(argv, capture_output=True)
except FileNotFoundError:
die("`curl` not found on PATH — install curl")
if proc.returncode != 0:
raise AstriaError(proc.stderr.decode("utf-8", "replace").strip() or f"curl exit {proc.returncode}")
def cmd_download(args, cfg):
ids = collect_prompt_ids(args)
if not ids:
die("no prompt ids — pass them as arguments, via --prompts-file, or on stdin")
out_dir = Path(args.out)
out_dir.mkdir(parents=True, exist_ok=True)
results = {}
def worker(prompt_id):
entry = {"id": prompt_id, "images": 0, "saved": []}
try:
prompt = request(cfg, "GET", f"/prompts/{prompt_id}.json", workspace=args.workspace)
except AstriaError as exc:
entry["error"] = str(exc)
results[prompt_id] = entry
return
urls = (prompt or {}).get("images") or []
entry["images"] = len(urls)
if not urls:
entry["error"] = "no images yet"
results[prompt_id] = entry
return
width = max(2, len(str(len(urls))))
for idx, url in enumerate(urls):
dest = out_dir / f"prompt-{prompt_id}-{idx:0{width}d}{asset_ext(url)}"
try:
download_asset(url, dest)
entry["saved"].append(str(dest))
except AstriaError as exc:
entry.setdefault("errors", []).append(f"{url}: {exc}")
if "errors" in entry and not entry["saved"]:
entry["error"] = "; ".join(entry.pop("errors"))
results[prompt_id] = entry
pending = list(ids)
lock = threading.Lock()
def pump():
while True:
with lock:
if not pending:
return
prompt_id = pending.pop(0)
worker(prompt_id)
threads = [threading.Thread(target=pump) for _ in range(min(DOWNLOAD_WORKERS, len(ids)))]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
prompts = [results[pid] for pid in ids]
emit({
"out": str(out_dir),
"prompts": prompts,
"totals": {
"prompts": len(prompts),
"downloaded": sum(len(p["saved"]) for p in prompts),
"failed": sum(1 for p in prompts if p.get("error")),
},
})
# --------------------------------------------------------------------------
# commands — raw API escape hatch
# --------------------------------------------------------------------------
def cmd_api(args, cfg):
form = [tuple(item.split("=", 1)) for item in args.form or [] if "=" in item]
uploaded = iter(direct_uploads(cfg, [value[1:] for _, value in form if value.startswith("@")]))
form = [(key, next(uploaded) if value.startswith("@") else value) for key, value in form]
query = [tuple(item.split("=", 1)) for item in args.query or [] if "=" in item]
json_body = json.loads(args.data) if args.data else None
result = request(cfg, args.method.upper(), args.path, query=query, form=form, json_body=json_body,
workspace=args.workspace)
emit(result) if not isinstance(result, str) else sys.stdout.write(result + "\n")
# --------------------------------------------------------------------------
# hidden — SessionStart hook
# --------------------------------------------------------------------------
def cmd_session_hook(args, cfg):
# Best-effort: keep the /cli/models id cache warm (daily refresh).
record = load_models_cache()
try:
if models_cache_stale(record, cfg):
record = fetch_models(cfg) or record
except Exception:
pass
authed = bool(os.environ.get("ASTRIA_API_KEY") or os.environ.get("ASTRIA_AUTH_TOKEN") or cfg.get("api_key"))
out = {"hookSpecificOutput": {"hookEventName": "SessionStart"}}
context = []
if not authed:
context.append(
"The Astria plugin is installed but not authenticated. If the user asks to do "
"anything with Astria (generate images, manage tunes/packs), tell them to run "
f"`astria login` first (API key at {login_hint(cfg)})."
)
skills_context = skills_upgrade_context(record)
if skills_context:
context.append(skills_context)
if context:
out["hookSpecificOutput"]["additionalContext"] = "\n\n".join(context)
emit(out)
# --------------------------------------------------------------------------
# argument parser
# --------------------------------------------------------------------------
# --help enum text is built from the cached /cli/models record so usage
# strings track the live catalog — never hardcode model names or resolutions.
def image_model_help(rec, prefix="model name or tune id"):
models = rec.get("models") or {}
if not models:
return f"{prefix}; see 'astria models', omit for the platform default"
names = ", ".join(
f"{name} (default)" if meta.get("default") or name == rec.get("default") else name
for name, meta in models.items())
return f"{prefix} — {names}; omit for the default"
def resolution_help(rec, noun="output resolution"):
res = sorted({r for meta in (rec.get("models") or {}).values()
for r in (meta.get("resolutions") or [])})
if not res:
return f"{noun}; see 'astria models' for per-model supported values"
return (f"{noun}, model-dependent — " + ", ".join(res)
+ "; see 'astria models' for each model's set")
def video_model_help(rec):
video_models = rec.get("video_models") or {}
if not video_models:
return "video model name; see 'astria models' for the list"
default = rec.get("default_video_model")
names = ", ".join(f"{name} (default)" if name == default else name
for name in video_models)
return "video model — " + names
def video_capability_help(rec, flag, base):
models = [name for name, meta in (rec.get("video_models") or {}).items()
if meta.get(flag)]
if not models:
return base
return f"{base}; supported by {', '.join(models)}"
def build_parser():
models_record = load_models_cache() or {}
parser = argparse.ArgumentParser(prog="astria", description="Astria API command-line client.")
parser.add_argument("--profile", metavar="NAME",
help="config profile to use (also ASTRIA_PROFILE env); default: 'default'")
parser.add_argument("-v", "--version", action="version", version=f"astria {VERSION}")
sub = parser.add_subparsers(dest="command", required=True)
# shared --workspace for API commands
ws_parent = argparse.ArgumentParser(add_help=False)
ws_parent.add_argument("-w", "--workspace", help="workspace id, 'all', or 'personal'")
# auth
p = sub.add_parser("login", help="store an API key in ~/.astria/config.json")
p.add_argument("--api-key", help="API key (else prompted / read from stdin)")
p.add_argument("--base-url", help="override API base URL")
p.add_argument("--workspace", help="default workspace id")
p.set_defaults(func=cmd_login)
sub.add_parser("logout", help="remove stored credentials").set_defaults(func=cmd_logout)
sub.add_parser("whoami", parents=[ws_parent], help="show the authenticated account").set_defaults(func=cmd_whoami)
p = sub.add_parser("config", help="show or change config")
p.add_argument("action", nargs="?", choices=["get", "set"], help="get (default) or set")
p.add_argument("key", nargs="?", help="config key for `set`")
p.add_argument("value", nargs="?", help="config value for `set`")
p.set_defaults(func=cmd_config)
p = sub.add_parser("models", help="list available image + video models (name -> tune id)")
p.add_argument("--refresh", action="store_true", help="force a refresh from the server")
p.set_defaults(func=cmd_models)
p = sub.add_parser("upgrade", help="upgrade this astria executable in place")
p.add_argument("--url", help=argparse.SUPPRESS)
p.add_argument("--path", help=argparse.SUPPRESS)
p.add_argument("--force", action="store_true", help="install even if the downloaded version is not newer")
p.set_defaults(func=cmd_upgrade)
# tunes
tunes = sub.add_parser("tunes", help="references / fine-tuned models").add_subparsers(dest="sub", required=True)
p = tunes.add_parser("list", parents=[ws_parent], help="list tunes")
p.add_argument("--title", help="filter by title / product name / SKU substring")
p.add_argument("--name", action="append", help="filter by class name (repeatable)")
p.add_argument("--model-type", help="filter by model_type (e.g. faceid)")
p.add_argument("--branch", help="filter by branch (e.g. partner-1 to discover partner models)")
p.add_argument("--gallery", action="store_true", help="query the public gallery instead")
p.add_argument("--limit")
p.add_argument("--offset")
p.set_defaults(func=cmd_tunes_list)
p = tunes.add_parser("get", parents=[ws_parent], help="get one tune")
p.add_argument("id")
p.set_defaults(func=cmd_tunes_get)
p = tunes.add_parser("create", parents=[ws_parent], help="create a tune / reference")
p.add_argument("--title", required=True)
p.add_argument("--name", required=True, help="subject class (man, woman, dress, shoes, pose, ...)")
p.add_argument("--model", default=None, help=image_model_help(models_record, "base model for the reference"))
p.add_argument("--model-type", default="faceid")
p.add_argument("--description", help="short description (stored in characteristics)")
p.add_argument("--image-url", action="append", help="training image URL (repeatable)")
p.add_argument("--image", action="append", help="local training image file (repeatable)")
p.set_defaults(func=cmd_tunes_create)
p = tunes.add_parser("update", parents=[ws_parent], help="update a tune")
p.add_argument("id")
p.add_argument("--title")
p.add_argument("--name")
p.add_argument("--description")
p.set_defaults(func=cmd_tunes_update)
# prompts
prompts = sub.add_parser("prompts", help="image/video generation records").add_subparsers(dest="sub", required=True)
p = prompts.add_parser("list", parents=[ws_parent], help="list prompts")
p.add_argument("--tune-id", help="scope to one tune's prompts")
p.add_argument("--pack-id", help="template prompts owned by a pack")
p.add_argument("--base-pack-id", help="prompts generated from a pack")
p.add_argument("--order-id", help="prompts generated for one order")
p.add_argument("--user-id")
p.add_argument("--orig-prompt-id", help="prompts cloned from this prompt id")
p.add_argument("--text", help="filter by prompt text substring")
p.add_argument("--liked", action="store_true", help="only prompts that have likes")
p.add_argument("--today", action="store_true", help="only prompts created today")
p.add_argument("--is-video", action="store_true", help="only video prompts")
p.add_argument("--is-api", action="store_true", help="only prompts created via the API")
p.add_argument("--expand", action="append", choices=["debug_images", "prompt.debug_images"],
help="expand a prompt field, e.g. debug_images (repeatable)")
p.add_argument("--limit")
p.add_argument("--offset")
p.set_defaults(func=cmd_prompts_list)
p = prompts.add_parser("get", parents=[ws_parent], help="get one prompt")
p.add_argument("id")
p.add_argument("--tune", help="tune id the prompt belongs to")
p.add_argument("--model", help="model name instead of --tune")
p.add_argument("--expand", action="append", choices=["debug_images", "prompt.debug_images"],
help="expand a prompt field, e.g. debug_images (repeatable)")
p.set_defaults(func=cmd_prompts_get)
p = prompts.add_parser("update", parents=[ws_parent], help="update a prompt (assign pack / move workspace)")
p.add_argument("id")
p.add_argument("--tune", help="tune id the prompt belongs to")
p.add_argument("--model", help="model name instead of --tune")
p.add_argument("--pack-id", help="assign as a pack template prompt")
p.add_argument("--base-pack-id", help="bind as a pack one-off")
p.add_argument("--workspace-id")
p.set_defaults(func=cmd_prompts_update)
p = prompts.add_parser("wait", parents=[ws_parent],
help="block until one or more prompts have images or a user_error")
p.add_argument("ids", nargs="*", help="prompt ids (also accepted via --prompts-file or stdin)")
p.add_argument("--prompts-file", help="file of prompt ids (one per line or whitespace-separated)")
p.add_argument("--timeout", type=int, default=WAIT_TIMEOUT_SEC,
help=f"total wall-clock cap in seconds (default: {WAIT_TIMEOUT_SEC})")
p.add_argument("--interval", type=int, default=WAIT_INTERVAL_SEC,
help=f"polling interval in seconds (default: {WAIT_INTERVAL_SEC}, min {WAIT_INTERVAL_MIN})")
p.set_defaults(func=cmd_prompts_wait)
# generate
p = sub.add_parser(
"generate", parents=[ws_parent], help="generate images",
formatter_class=argparse.RawTextHelpFormatter,
epilog=(
"examples:\n"
" astria generate --model wan-2-7 --text \"plain white background\" \\\n"
" --reference \"dress=./dress.jpg\" --reference \"woman=./woman.jpg\" --num-images 1 --wait"
),
)
p.add_argument("--model", default=None, help=image_model_help(models_record))
p.add_argument("--text", help="prompt text; embed references as <faceid:ID:1>")
p.add_argument("--video-prompt", help=argparse.SUPPRESS)
p.add_argument("--num-images", help="override config default_num_images; otherwise 2, or 1 with --input-image")
p.add_argument("--seed", help="generation seed; also makes an otherwise-identical prompt distinct")
p.add_argument("--film-grain", "--film_grain", action=argparse.BooleanOptionalAction, default=None,
help="enable film-grain post-processing as a separate prompt attribute; "
"--film_grain is an alias")
p.add_argument("--aspect-ratio", default="3:4", help="e.g. 3:4, 1:1, 16:9 (default: %(default)s)")
p.add_argument("--resolution", help=resolution_help(models_record))
p.add_argument("--input-image", help="URL or local file to edit")
p.add_argument("--mask-image", help="URL or local file to use as an inpainting mask")
p.add_argument("--reference", "--images", dest="references", action="append", metavar="NAME=PATH_OR_URL",
help="create and attach a reference; repeat for each NAME=PATH_OR_URL (--images is an alias)")
p.add_argument("--pack-id", help="author the prompt as a pack template prompt")
p.add_argument("--base-pack-id", help="bind as a pack one-off")
p.add_argument("--wait", action="store_true", help="block until images are ready")
p.set_defaults(func=cmd_generate)
# video
p = sub.add_parser(
"video", parents=[ws_parent], help="generate video",
formatter_class=argparse.RawTextHelpFormatter,
epilog=(
"references use the same syntax as image prompts:\n"
" <faceid:TUNE_ID:1> TUNE_NAME\n"
"The tune name must immediately follow its token. Put existing tune mentions in\n"
"--video-prompt. --reference NAME=PATH_OR_URL creates a new reference and adds\n"
"the correctly formatted mention to --text and --video-prompt automatically.\n\n"
"examples:\n"
" # Existing reference tune\n"
" astria video --video-model seedance2_fast_720p \\\n"
" --video-prompt \"<faceid:1234:1> woman walks down a runway\" --duration 5 --wait\n\n"
" # Create and use references from local files or URLs\n"
" astria video --video-model seedance2_fast_480p \\\n"
" --video-prompt \"woman wearing dress, plain white background\" \\\n"
" --reference \"dress=./dress.jpg\" --reference \"woman=./woman.jpg\" \\\n"
" --num-images 1 --wait\n\n"
" # Attach ordered raw image references without creating tunes\n"
" astria video --video-model seedance2_fast_720p \\\n"
" --video-prompt \"transition through these looks in order\" \\\n"
" --image-reference ./look-1.jpg --image-reference ./look-2.jpg --wait"
),
)
p.add_argument("--model", default=None, help=image_model_help(models_record, "first-frame image model"))
p.add_argument("--text", help="first-frame prompt text")
p.add_argument("--video-model", required=True, help=video_model_help(models_record))
p.add_argument("--video-prompt", required=True,
help="motion / action description; references use <faceid:TUNE_ID:1> TUNE_NAME")
p.add_argument("--duration", help="seconds (model-dependent)")
p.add_argument("--aspect-ratio", default="3:4", help="e.g. 3:4, 1:1, 16:9 (default: %(default)s)")
p.add_argument("--resolution", help=resolution_help(
models_record, "first-frame image resolution (video resolution is set by --video-model)"))
p.add_argument("--num-images", help="override config default_num_images; otherwise 2")
p.add_argument("--input-image", help=argparse.SUPPRESS)
p.add_argument("--reference", "--images", dest="references", action="append", metavar="NAME=PATH_OR_URL",
help="create and attach a reference; repeat for each NAME=PATH_OR_URL (--images is an alias)")
p.add_argument("--image-reference", dest="image_references", action="append", metavar="PATH_OR_URL",
help=video_capability_help(
models_record, "supports_image_references",
"attach an ordered raw image reference; repeat for each image"))
p.add_argument("--first-frame", help="URL or file overriding the rendered first frame")
p.add_argument("--last-frame", help="URL or file for keyframe models")
p.add_argument("--input-video", help="URL or file for motion-control models")
p.add_argument("--audio-reference", help=video_capability_help(
models_record, "supports_audio_reference",
"URL or file — reference audio track the video is driven by"))
p.add_argument("--generate-audio", action=argparse.BooleanOptionalAction, default=None,
help=video_capability_help(
models_record, "generates_audio",
"generate a soundtrack; omit to use the model default"))
p.add_argument("--wait", action="store_true")
p.set_defaults(func=cmd_video)
p = sub.add_parser(
"variate", parents=[ws_parent],
help="edit a source video while preserving its timing and performance",
formatter_class=argparse.RawTextHelpFormatter,
epilog=(
"examples:\n"
" astria variate ./source.mp4 --brief 'Change the end card to Astria' --wait\n"
" astria variate ./source.mp4 --reference ./dress.jpg --reference woman=./model.jpg --wait\n"
" astria variate https://example.com/source.mp4 --tune-id 123 --brief 'Replace the presenter'"
),
)
p.add_argument("source", help="local MP4/MOV file or public HTTPS video URL")
p.add_argument("--brief", default="", help="requested edit; optional when references are supplied")
p.add_argument("--tune-id", dest="tune_ids", action="append", metavar="ID",
help="existing replacement reference (repeatable)")
p.add_argument("--reference", dest="references", action="append", metavar="[NAME=]PATH_OR_URL",
help="create a replacement reference; omit NAME to auto-detect it (repeatable)")
description = p.add_mutually_exclusive_group()
description.add_argument("--description", help="source-video description; skips video inspection")
description.add_argument("--description-file", help="read source-video description from a UTF-8 file")
p.add_argument("--base-pack-id", type=int, help="bind the result as a pack one-off")
p.add_argument("--order-id", type=int, help="associate the result with an accessible order")
p.add_argument("--wait", action="store_true", help="block until the generated video is ready")
p.set_defaults(func=cmd_variate)
p = sub.add_parser("inspect-video", parents=[ws_parent],
help="describe a video as timestamped text-to-video prompt text")
p.add_argument("source", help="local video file or public HTTPS URL")
p.add_argument("--tune-id", action="append", metavar="ID",
help="reference whose appearance should be removed from the description (repeatable)")
p.set_defaults(func=cmd_inspect_video)
p = sub.add_parser("inspect", parents=[ws_parent],
help="describe an image as image-generation prompt text")
p.add_argument("source", help="local image file or public HTTPS image URL")
p.add_argument("--name", action="append",
help="reference name whose appearance should be removed from the description (repeatable)")
p.set_defaults(func=cmd_inspect)
# download — pull a prompt's rendered images/videos to a local directory
p = sub.add_parser("download", parents=[ws_parent],
help="download the images/videos for one or more prompt ids")
p.add_argument("ids", nargs="*", help="prompt ids (also accepted via --prompts-file or stdin)")
p.add_argument("--out", default="./astria-downloads", help="target directory (created if missing)")
p.add_argument("--prompts-file", help="file of prompt ids (one per line or whitespace-separated)")
p.set_defaults(func=cmd_download)
# packs
packs = sub.add_parser("packs", help="prompt-template collections").add_subparsers(dest="sub", required=True)
p = packs.add_parser("list", parents=[ws_parent], help="list packs")
p.add_argument("--limit")
p.add_argument("--offset")
p.set_defaults(func=cmd_packs_list)
p = packs.add_parser("get", parents=[ws_parent], help="get a pack and its template prompts")
p.add_argument("slug", help="pack slug or id")
p.set_defaults(func=cmd_packs_get)
p = packs.add_parser("create", parents=[ws_parent], help="create a pack")
p.add_argument("--title", required=True)
p.add_argument("--subtitle")
p.add_argument("--model", default=None, help=image_model_help(models_record, "base model for the pack"))
p.add_argument("--model-type", default="faceid")
p.set_defaults(func=cmd_packs_create)
p = packs.add_parser(
"run", parents=[ws_parent],
formatter_class=argparse.RawTextHelpFormatter,
help="run a pack (template): fire its prompts against tunes (or a fresh training set)",
epilog=(
"examples:\n"
" # multi pack — run against existing tunes, with overrides\n"
" astria packs run spring-lookbook --tune-id 123 --tune-id 456 \\\n"
" --brief \"golden hour, Lisbon\" --aspect-ratio 3:4 --inpaint-faces\n"
" # only a subset of the pack's template prompts\n"
" astria packs run spring-lookbook --tune-id 123 --prompt-ids 501,502\n"
" # regular pack — train a fresh tune from photos, then generate\n"
" astria packs run my-pack --title Jane --name woman --image ./a.jpg --image ./b.jpg"
),
)
p.add_argument("slug", help="pack slug or id")
p.add_argument("--tune-id", action="append", metavar="ID",
help="existing tune to run the pack against (repeatable; multi packs need >=1)")
p.add_argument("--prompt-ids", help="subset of the pack's template prompt ids, comma-separated")
p.add_argument("--brief", help="art-direction brief applied to the generated prompts")
p.add_argument("--title", help="[train] tune title, e.g. the subject's name")
p.add_argument("--name", help="[train] subject class: man, woman, dress, shoes, ...")
p.add_argument("--image", action="append", metavar="FILE", help="[train] local image (repeatable)")
p.add_argument("--image-url", action="append", metavar="URL", help="[train] remote image (repeatable)")
p.add_argument("--num-images", help="override: images per prompt")
p.add_argument("--aspect-ratio", help="override: e.g. 3:4, 1:1, 16:9")
p.add_argument("--resolution", help="override: e.g. 1K, 2K")
p.add_argument("--inpaint-faces", action=argparse.BooleanOptionalAction, default=None,
help="override: face inpainting on/off")
p.add_argument("--attr", action="append", metavar="KEY=VALUE",
help="any other prompt attribute override (repeatable), e.g. --attr super_resolution=true")
p.set_defaults(func=cmd_packs_run)
# workspaces
workspaces = sub.add_parser("workspaces", help="workspaces").add_subparsers(dest="sub", required=True)
workspaces.add_parser("list", help="list workspaces").set_defaults(func=cmd_workspaces_list)
wc = workspaces.add_parser("create", help="create a workspace")
wc.add_argument("--title", required=True)
wc.set_defaults(func=cmd_workspaces_create)
# agent handoff
agent = sub.add_parser("agent", help="Astria embedded agent").add_subparsers(dest="sub", required=True)
p = agent.add_parser("handoff", parents=[ws_parent], help="hand work from another agent into Astria")
p.add_argument("--handoff", required=True, metavar="HANDOFF.md", help="UTF-8 handoff summary")
p.add_argument("--attach", action="append", metavar="PATH", help="attach a file or directory (repeatable)")
p.add_argument("--skill", action="append", metavar="DIR", help="include a skill directory containing SKILL.md (repeatable)")
p.add_argument("--source", choices=["chatgpt", "claude-code", "claude-desktop", "cursor", "other"])
p.add_argument("--title", help="title for the imported Astria chat")
p.add_argument("--open", action="store_true", help="open the Astria deep link after upload")
p.set_defaults(func=cmd_agent_handoff)
# landing page
landing = sub.add_parser("landing", help="workspace landing page").add_subparsers(dest="sub", required=True)
p = landing.add_parser("get", parents=[ws_parent], help="fetch the workspace JSON (incl. landing_page_html)")
p.add_argument("--html", action="store_true", help="print only landing_page_html (raw)")
p.set_defaults(func=cmd_landing_get)
p = landing.add_parser("set", parents=[ws_parent], help="update the landing page")
p.add_argument("--html-file", help="path to the edited inner-body HTML")
p.add_argument("--brief", help="regenerate the whole page from this brief (Gemini)")
p.set_defaults(func=cmd_landing_set)
# cache — local snapshot (JSON + SQLite) of tunes/prompts/packs/user.
# Query subcommands (get/find/uses/stats) read ONLY the local DB, never the API.
cache = sub.add_parser("cache", parents=[ws_parent],
help="local snapshot + query layer for tunes/prompts/packs")
cache_sub = cache.add_subparsers(dest="action")
cache.set_defaults(action="refresh", resource=None, force=False, func=cmd_cache)
p = cache_sub.add_parser("refresh", parents=[ws_parent],
help="pull tunes/prompts/packs/user into JSON + SQLite")
p.add_argument("resource", nargs="?", choices=list(CACHE_FIELDS), help="refresh only one resource")
p.add_argument("--force", action="store_true", help="refresh even if the cache is fresh")
p.set_defaults(func=cmd_cache)
p = cache_sub.add_parser("path", parents=[ws_parent], help="print the cache directory")
p.set_defaults(func=cmd_cache, resource=None, force=False)
p = cache_sub.add_parser("get", parents=[ws_parent], help="get one cached record by id (local DB)")
p.add_argument("table", choices=list(DB_COLUMNS), help="tunes / prompts / packs")
p.add_argument("id", help="record id")
p.set_defaults(func=cmd_cache, resource=None, force=False)
p = cache_sub.add_parser("find", parents=[ws_parent], help="filter cached records (local DB)")
find_sub = p.add_subparsers(dest="table", required=True)
f = find_sub.add_parser("tunes", parents=[ws_parent], help="find tunes by name / title substring")
f.add_argument("--name", help="class-name substring filter")
f.add_argument("--title", help="title substring filter")
f.set_defaults(func=cmd_cache, action="find", resource=None, force=False)
f = find_sub.add_parser("prompts", parents=[ws_parent], help="find prompts by pack / tune / text")
f.add_argument("--pack-id", dest="pack_id", help="exact pack id")
f.add_argument("--tune-id", dest="tune_id", help="exact tune id")
f.add_argument("--order-id", dest="order_id", help="exact order id")
f.add_argument("--text", help="prompt text substring filter")
f.set_defaults(func=cmd_cache, action="find", resource=None, force=False)
f = find_sub.add_parser("packs", parents=[ws_parent], help="find packs by main-class / title")
f.add_argument("--main-class", dest="main_class", help="main_class_name substring filter")
f.add_argument("--title", help="title substring filter")
f.set_defaults(func=cmd_cache, action="find", resource=None, force=False)
p = cache_sub.add_parser("uses", parents=[ws_parent],
help="prompts whose text references a tune id (<faceid:ID:..>)")
p.add_argument("tune_id", help="tune id to cross-reference")
p.set_defaults(func=cmd_cache, resource=None, force=False)
p = cache_sub.add_parser("stats", parents=[ws_parent], help="row counts per table + cache age")
p.set_defaults(func=cmd_cache, resource=None, force=False)
# raw api
p = sub.add_parser("api", parents=[ws_parent], help="raw API call (escape hatch)")
p.add_argument("method", help="GET / POST / PATCH / DELETE")
p.add_argument("path", help="API path, e.g. /tunes?limit=5")
p.add_argument("--query", action="append", help="query param k=v (repeatable)")
p.add_argument("--form", action="append",
help="form field k=v (repeatable; @file values are direct-uploaded first)")
p.add_argument("--data", help="raw JSON request body")
p.set_defaults(func=cmd_api)
# hidden hook
p = sub.add_parser("_session-hook")
p.set_defaults(func=cmd_session_hook)
return parser
def main():
# Resolve the profile up front so build_parser() reads that profile's
# cached model catalog for the --help enum text; re-resolve from the
# parsed args afterwards so the flag stays authoritative.
sys.argv[1:] = normalize_meta_argv(sys.argv[1:])
apply_profile(resolve_profile(argv_profile()))
args = build_parser().parse_args()
apply_profile(resolve_profile(args.profile))
cfg = load_config()
if args.command != "upgrade":
check_cli_version(load_models_cache(), auto_upgrade=True)
try:
args.func(args, cfg)
except AstriaError as exc:
die(str(exc))
except BrokenPipeError:
sys.exit(0)
except KeyboardInterrupt:
die("interrupted", code=130)
if __name__ == "__main__":
main()
SHA-256: af0acd4b8532f13c5bf01c01f29294c58d166657e2eca603e58d99edd9c46465