wehub-resource-sync
733 lines
27 KiB
Python
733 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""Small CLI for Agnes AI text, image, and video generation APIs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
import urllib.parse
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
|
|
BASE_URL = "https://apihub.agnes-ai.com"
|
|
TEXT_MODEL = "agnes-2.0-flash"
|
|
IMAGE_MODEL = "agnes-image-2.1-flash"
|
|
VIDEO_MODEL = "agnes-video-v2.0"
|
|
SIZE_RE = re.compile(r"^[1-9]\d*x[1-9]\d*$")
|
|
|
|
|
|
def get_api_key() -> str:
|
|
for name in ("AGNES_API_KEY", "AGNES_API_TOKEN", "APIHUB_AGNES_API_KEY"):
|
|
value = os.environ.get(name)
|
|
if value:
|
|
return value
|
|
raise SystemExit(
|
|
"Missing API key. Set AGNES_API_KEY, AGNES_API_TOKEN, or APIHUB_AGNES_API_KEY."
|
|
)
|
|
|
|
|
|
def request_json(method: str, path: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
body = None if payload is None else json.dumps(payload).encode("utf-8")
|
|
req = urllib.request.Request(
|
|
BASE_URL + path,
|
|
data=body,
|
|
method=method,
|
|
headers={
|
|
"Authorization": f"Bearer {get_api_key()}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=120) as resp:
|
|
text = resp.read().decode("utf-8")
|
|
return json.loads(text) if text else {}
|
|
except urllib.error.HTTPError as exc:
|
|
detail = exc.read().decode("utf-8", errors="replace")
|
|
raise SystemExit(f"HTTP {exc.code} from {path}: {detail}") from exc
|
|
except urllib.error.URLError as exc:
|
|
raise SystemExit(f"Request failed for {path}: {exc}") from exc
|
|
|
|
|
|
def request_text(method: str, path: str, payload: dict[str, Any] | None = None) -> str:
|
|
body = None if payload is None else json.dumps(payload).encode("utf-8")
|
|
req = urllib.request.Request(
|
|
BASE_URL + path,
|
|
data=body,
|
|
method=method,
|
|
headers={
|
|
"Authorization": f"Bearer {get_api_key()}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=120) as resp:
|
|
return resp.read().decode("utf-8", errors="replace")
|
|
except urllib.error.HTTPError as exc:
|
|
detail = exc.read().decode("utf-8", errors="replace")
|
|
raise SystemExit(f"HTTP {exc.code} from {path}: {detail}") from exc
|
|
except urllib.error.URLError as exc:
|
|
raise SystemExit(f"Request failed for {path}: {exc}") from exc
|
|
|
|
|
|
def stream_summary(payload: dict[str, Any]) -> dict[str, Any]:
|
|
raw = request_text("POST", "/v1/chat/completions", payload)
|
|
event_count = 0
|
|
done = False
|
|
content_parts: list[str] = []
|
|
for line in raw.splitlines():
|
|
line = line.strip()
|
|
if not line.startswith("data:"):
|
|
continue
|
|
data = line.removeprefix("data:").strip()
|
|
if data == "[DONE]":
|
|
done = True
|
|
elif data:
|
|
event_count += 1
|
|
try:
|
|
event = json.loads(data)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
try:
|
|
delta = event["choices"][0].get("delta", {})
|
|
except (KeyError, IndexError, TypeError, AttributeError):
|
|
continue
|
|
content = delta.get("content")
|
|
if isinstance(content, str):
|
|
content_parts.append(content)
|
|
return {
|
|
"type": "text-stream",
|
|
"content": "".join(content_parts) or None,
|
|
"events": event_count,
|
|
"done": done,
|
|
"raw_prefix": raw[:200],
|
|
}
|
|
|
|
|
|
def print_json(data: Any) -> None:
|
|
print(json.dumps(data, ensure_ascii=False, indent=2))
|
|
|
|
|
|
def parse_json_arg(name: str, value: str) -> Any:
|
|
try:
|
|
return json.loads(value)
|
|
except json.JSONDecodeError as exc:
|
|
raise SystemExit(f"Invalid JSON for {name}: {exc.msg} at position {exc.pos}") from exc
|
|
|
|
|
|
def needs_english_translation(prompt: str) -> bool:
|
|
return any(ord(ch) > 127 for ch in prompt)
|
|
|
|
|
|
def translate_prompt_to_english(prompt: str) -> str:
|
|
payload = {
|
|
"model": TEXT_MODEL,
|
|
"messages": [
|
|
{
|
|
"role": "system",
|
|
"content": (
|
|
"Translate the user's image/video generation prompt into fluent English. "
|
|
"Preserve all concrete visual details, style words, camera motion, lighting, "
|
|
"composition constraints, and negative instructions. Return only the English prompt."
|
|
),
|
|
},
|
|
{"role": "user", "content": prompt},
|
|
],
|
|
"temperature": 0,
|
|
"max_tokens": 800,
|
|
}
|
|
data = request_json("POST", "/v1/chat/completions", payload)
|
|
try:
|
|
translated = data["choices"][0]["message"]["content"].strip()
|
|
except (KeyError, IndexError, TypeError) as exc:
|
|
raise SystemExit(f"Prompt translation failed: {json.dumps(data, ensure_ascii=False)}") from exc
|
|
if not translated:
|
|
raise SystemExit("Prompt translation failed: empty translated prompt")
|
|
return translated
|
|
|
|
|
|
def prepare_generation_prompt(prompt: str, translate: bool = True) -> tuple[str, str | None]:
|
|
if translate and needs_english_translation(prompt):
|
|
translated = translate_prompt_to_english(prompt)
|
|
return translated, translated
|
|
return prompt, None
|
|
|
|
|
|
def extract_text_content(data: dict[str, Any]) -> str | None:
|
|
try:
|
|
content = data["choices"][0]["message"].get("content")
|
|
except (KeyError, IndexError, TypeError, AttributeError):
|
|
return None
|
|
return content if isinstance(content, str) else None
|
|
|
|
|
|
def output_result(
|
|
result_type: str,
|
|
raw: dict[str, Any],
|
|
*,
|
|
prompt_used: str | None = None,
|
|
translated_prompt: str | None = None,
|
|
urls: list[str] | None = None,
|
|
status: str | None = None,
|
|
next_steps: list[str] | None = None,
|
|
raw_only: bool = False,
|
|
) -> None:
|
|
if raw_only:
|
|
print_json(raw)
|
|
return
|
|
summary: dict[str, Any] = {"type": result_type}
|
|
if status:
|
|
summary["status"] = status
|
|
if urls:
|
|
summary["urls"] = urls
|
|
if prompt_used:
|
|
summary["prompt_used"] = prompt_used
|
|
if translated_prompt:
|
|
summary["translated_prompt"] = translated_prompt
|
|
if next_steps:
|
|
summary["next_steps"] = next_steps
|
|
summary["raw"] = raw
|
|
print_json(summary)
|
|
|
|
|
|
def extract_image_urls(data: dict[str, Any]) -> list[str]:
|
|
urls = []
|
|
if isinstance(data.get("url"), str):
|
|
urls.append(data["url"])
|
|
if isinstance(data.get("image_url"), str):
|
|
urls.append(data["image_url"])
|
|
if isinstance(data.get("data"), list):
|
|
for item in data["data"]:
|
|
if isinstance(item, dict):
|
|
for key in ("url", "image_url"):
|
|
if isinstance(item.get(key), str):
|
|
urls.append(item[key])
|
|
return urls
|
|
|
|
|
|
def extract_video_urls(data: dict[str, Any]) -> list[str]:
|
|
urls = []
|
|
for key in ("video_url", "url", "remixed_from_video_id"):
|
|
value = data.get(key)
|
|
if isinstance(value, str) and value.startswith(("http://", "https://")):
|
|
urls.append(value)
|
|
if isinstance(data.get("data"), list):
|
|
for item in data["data"]:
|
|
if isinstance(item, dict):
|
|
urls.extend(extract_video_urls(item))
|
|
return list(dict.fromkeys(urls))
|
|
|
|
|
|
def video_lookup_id(data: dict[str, Any]) -> tuple[str | None, str | None]:
|
|
video_id = data.get("video_id")
|
|
if isinstance(video_id, str) and video_id:
|
|
return video_id, "video_id"
|
|
for key in ("task_id", "id"):
|
|
value = data.get(key)
|
|
if isinstance(value, str) and value:
|
|
return value, "task_id"
|
|
return None, None
|
|
|
|
|
|
def retrieve_video(identifier: str, model_name: str | None = VIDEO_MODEL) -> dict[str, Any]:
|
|
if identifier.startswith("video_"):
|
|
query = {"video_id": identifier}
|
|
if model_name:
|
|
query["model_name"] = model_name
|
|
return request_json("GET", "/agnesapi?" + urllib.parse.urlencode(query))
|
|
return request_json("GET", f"/v1/videos/{urllib.parse.quote(identifier, safe='')}")
|
|
|
|
|
|
def validate_size(value: str | None, name: str = "size") -> None:
|
|
if value and not SIZE_RE.match(value):
|
|
raise SystemExit(f"Invalid {name}: {value}. Expected WIDTHxHEIGHT, for example 1024x768.")
|
|
|
|
|
|
def validate_video_args(args: argparse.Namespace) -> None:
|
|
if args.num_frames is not None:
|
|
if args.num_frames > 441 or (args.num_frames - 1) % 8 != 0:
|
|
raise SystemExit("Invalid --num-frames: must be <= 441 and satisfy 8n + 1, for example 81 or 121.")
|
|
if args.frame_rate is not None and not (1 <= args.frame_rate <= 60):
|
|
raise SystemExit("Invalid --frame-rate: supported range is 1-60.")
|
|
for name in ("height", "width"):
|
|
value = getattr(args, name)
|
|
if value is not None and value <= 0:
|
|
raise SystemExit(f"Invalid --{name.replace('_', '-')}: must be a positive integer.")
|
|
|
|
|
|
def cmd_text(args: argparse.Namespace) -> None:
|
|
messages = []
|
|
if args.system:
|
|
messages.append({"role": "system", "content": args.system})
|
|
messages.append({"role": "user", "content": args.prompt})
|
|
payload: dict[str, Any] = {
|
|
"model": TEXT_MODEL,
|
|
"messages": messages,
|
|
"temperature": args.temperature,
|
|
"max_tokens": args.max_tokens,
|
|
}
|
|
if args.top_p is not None:
|
|
payload["top_p"] = args.top_p
|
|
if args.stream:
|
|
payload["stream"] = True
|
|
if args.tools_json:
|
|
payload["tools"] = parse_json_arg("--tools-json", args.tools_json)
|
|
if args.tool_choice_json:
|
|
payload["tool_choice"] = parse_json_arg("--tool-choice-json", args.tool_choice_json)
|
|
if args.stream:
|
|
print_json(stream_summary(payload))
|
|
else:
|
|
data = request_json("POST", "/v1/chat/completions", payload)
|
|
content = extract_text_content(data)
|
|
wrapped = {
|
|
"type": "text",
|
|
"content": content,
|
|
"raw": data,
|
|
}
|
|
print_json(data if args.raw else wrapped)
|
|
|
|
|
|
def cmd_image(args: argparse.Namespace) -> None:
|
|
validate_size(args.size)
|
|
prompt, translated_prompt = prepare_generation_prompt(args.prompt, not args.no_translate_prompt)
|
|
payload: dict[str, Any] = {
|
|
"model": IMAGE_MODEL,
|
|
"prompt": prompt,
|
|
}
|
|
if args.size:
|
|
payload["size"] = args.size
|
|
extra: dict[str, Any] = {"response_format": "url"}
|
|
if args.image:
|
|
extra["image"] = args.image
|
|
if extra:
|
|
payload["extra_body"] = extra
|
|
data = request_json("POST", "/v1/images/generations", payload)
|
|
urls = extract_image_urls(data)
|
|
output_result(
|
|
"image-to-image" if args.image else "text-to-image",
|
|
data,
|
|
prompt_used=prompt,
|
|
translated_prompt=translated_prompt,
|
|
urls=urls,
|
|
raw_only=args.raw,
|
|
)
|
|
|
|
|
|
def video_payload(args: argparse.Namespace) -> dict[str, Any]:
|
|
validate_video_args(args)
|
|
prompt, translated_prompt = prepare_generation_prompt(args.prompt, not args.no_translate_prompt)
|
|
args._prompt_used = prompt
|
|
args._translated_prompt = translated_prompt
|
|
payload: dict[str, Any] = {
|
|
"model": VIDEO_MODEL,
|
|
"prompt": prompt,
|
|
}
|
|
for name in (
|
|
"height",
|
|
"width",
|
|
"num_frames",
|
|
"frame_rate",
|
|
"num_inference_steps",
|
|
"seed",
|
|
"negative_prompt",
|
|
):
|
|
value = getattr(args, name)
|
|
if value is not None:
|
|
payload[name] = value
|
|
if args.mode:
|
|
payload["mode"] = args.mode
|
|
if args.image:
|
|
if len(args.image) == 1 and args.mode != "keyframes":
|
|
payload["image"] = args.image[0]
|
|
else:
|
|
payload["extra_body"] = {"image": args.image}
|
|
if args.mode:
|
|
payload["extra_body"]["mode"] = args.mode
|
|
return payload
|
|
|
|
|
|
def poll_video(identifier: str, timeout: int, interval: int) -> dict[str, Any]:
|
|
deadline = time.time() + timeout
|
|
last: dict[str, Any] = {}
|
|
while time.time() < deadline:
|
|
last = retrieve_video(identifier)
|
|
if last.get("error"):
|
|
raise SystemExit(f"Video {identifier} returned error: {json.dumps(last, ensure_ascii=False)}")
|
|
status = str(last.get("status", "")).lower()
|
|
progress = last.get("progress")
|
|
if status:
|
|
print(f"video {identifier}: status={status} progress={progress}", file=sys.stderr)
|
|
if status in {"completed", "failed"}:
|
|
return last
|
|
time.sleep(interval)
|
|
raise SystemExit(f"Timed out waiting for video {identifier}. Last response: {json.dumps(last)}")
|
|
|
|
|
|
def cmd_video(args: argparse.Namespace) -> None:
|
|
created = request_json("POST", "/v1/videos", video_payload(args))
|
|
if not args.poll:
|
|
identifier, id_kind = video_lookup_id(created)
|
|
next_steps = []
|
|
if identifier:
|
|
next_steps.append(f"python scripts/agnes_api.py video-get {identifier}")
|
|
next_steps.append(f"python scripts/agnes_api.py video-get {identifier} # repeat until status is completed")
|
|
output_result(
|
|
"video-task",
|
|
created,
|
|
prompt_used=getattr(args, "_prompt_used", None),
|
|
translated_prompt=getattr(args, "_translated_prompt", None),
|
|
status=str(created.get("status", "")) if created.get("status") is not None else None,
|
|
next_steps=next_steps,
|
|
raw_only=args.raw,
|
|
)
|
|
if id_kind == "task_id":
|
|
print("warning: create response did not include video_id; falling back to legacy task_id lookup", file=sys.stderr)
|
|
return
|
|
identifier, id_kind = video_lookup_id(created)
|
|
if not identifier:
|
|
raise SystemExit(f"Video create response did not include video_id, task_id, or id: {json.dumps(created)}")
|
|
if id_kind == "task_id":
|
|
print("warning: create response did not include video_id; falling back to legacy task_id lookup", file=sys.stderr)
|
|
data = poll_video(identifier, args.timeout, args.interval)
|
|
urls = extract_video_urls(data)
|
|
output_result(
|
|
"video-result",
|
|
data,
|
|
prompt_used=getattr(args, "_prompt_used", None),
|
|
translated_prompt=getattr(args, "_translated_prompt", None),
|
|
urls=urls,
|
|
status=str(data.get("status", "")) if data.get("status") is not None else None,
|
|
raw_only=args.raw,
|
|
)
|
|
|
|
|
|
def cmd_video_get(args: argparse.Namespace) -> None:
|
|
data = retrieve_video(args.identifier, None if args.no_model_name else args.model_name)
|
|
urls = extract_video_urls(data)
|
|
output_result(
|
|
"video-result",
|
|
data,
|
|
urls=urls,
|
|
status=str(data.get("status", "")) if data.get("status") is not None else None,
|
|
next_steps=[] if urls else [f"python scripts/agnes_api.py video-get {args.identifier}"],
|
|
raw_only=args.raw,
|
|
)
|
|
if data.get("error"):
|
|
raise SystemExit(1)
|
|
|
|
|
|
def require_ok(name: str, data: dict[str, Any], keys: tuple[str, ...]) -> None:
|
|
missing = [key for key in keys if key not in data]
|
|
if missing:
|
|
raise SystemExit(f"{name} response missing {missing}: {json.dumps(data)}")
|
|
print(f"{name}: ok")
|
|
|
|
|
|
def require_video_ok(name: str, data: dict[str, Any], completed: bool = False) -> None:
|
|
require_ok(name, data, ("id", "status"))
|
|
if data.get("error"):
|
|
raise SystemExit(f"{name} returned error: {json.dumps(data, ensure_ascii=False)}")
|
|
status = str(data.get("status", "")).lower()
|
|
if status == "failed":
|
|
raise SystemExit(f"{name} failed: {json.dumps(data, ensure_ascii=False)}")
|
|
if completed and status != "completed":
|
|
raise SystemExit(f"{name} did not complete: {json.dumps(data, ensure_ascii=False)}")
|
|
if completed and not extract_video_urls(data):
|
|
raise SystemExit(f"{name} completed without a video URL: {json.dumps(data, ensure_ascii=False)}")
|
|
|
|
|
|
def check_tool_call(name: str, data: dict[str, Any], strict: bool = False) -> None:
|
|
try:
|
|
tool_calls = data["choices"][0]["message"].get("tool_calls")
|
|
except (KeyError, IndexError, TypeError, AttributeError):
|
|
tool_calls = None
|
|
if not tool_calls:
|
|
message = f"{name}: request accepted, but response did not include tool_calls"
|
|
if strict:
|
|
raise SystemExit(f"{message}: {json.dumps(data, ensure_ascii=False)}")
|
|
print(message, file=sys.stderr)
|
|
return
|
|
print(f"{name}: ok")
|
|
|
|
|
|
def extract_image_url(data: dict[str, Any]) -> str:
|
|
candidates = extract_image_urls(data)
|
|
if not candidates:
|
|
raise SystemExit(f"Could not find image URL in response: {json.dumps(data, ensure_ascii=False)}")
|
|
return candidates[0]
|
|
|
|
|
|
def create_video_case(name: str, payload: dict[str, Any], args: argparse.Namespace) -> dict[str, Any]:
|
|
created = request_json("POST", "/v1/videos", payload)
|
|
require_video_ok(f"{name}-create", created)
|
|
identifier, id_kind = video_lookup_id(created)
|
|
if not identifier:
|
|
raise SystemExit(f"{name}-create response missing video_id/task_id/id: {json.dumps(created, ensure_ascii=False)}")
|
|
if id_kind == "task_id":
|
|
print(f"{name}-create: warning: falling back to legacy task_id lookup", file=sys.stderr)
|
|
retrieved = (
|
|
poll_video(identifier, args.video_timeout, args.video_interval)
|
|
if args.poll_video
|
|
else retrieve_video(identifier)
|
|
)
|
|
require_video_ok(f"{name}-get", retrieved, completed=args.poll_video)
|
|
return {"create": created, "get": retrieved}
|
|
|
|
|
|
VIDEO_CASES = ("text-to-video", "image-to-video", "multi-image", "keyframes")
|
|
|
|
|
|
def cmd_smoke_test(args: argparse.Namespace) -> None:
|
|
validate_size(args.image_size, "image-size")
|
|
validate_video_args(
|
|
argparse.Namespace(
|
|
num_frames=args.video_num_frames,
|
|
frame_rate=args.video_frame_rate,
|
|
height=args.video_height,
|
|
width=args.video_width,
|
|
)
|
|
)
|
|
text = request_json(
|
|
"POST",
|
|
"/v1/chat/completions",
|
|
{
|
|
"model": TEXT_MODEL,
|
|
"messages": [{"role": "user", "content": "Reply with exactly: Agnes text ok"}],
|
|
"max_tokens": 20,
|
|
"temperature": 0,
|
|
},
|
|
)
|
|
require_ok("text", text, ("choices",))
|
|
|
|
text_stream = stream_summary(
|
|
{
|
|
"model": TEXT_MODEL,
|
|
"messages": [{"role": "user", "content": "Reply with exactly: Agnes stream ok"}],
|
|
"max_tokens": 20,
|
|
"temperature": 0,
|
|
"stream": True,
|
|
}
|
|
)
|
|
if text_stream["events"] < 1 and not text_stream["done"]:
|
|
raise SystemExit(f"text-stream response did not look like SSE: {json.dumps(text_stream)}")
|
|
print("text-stream: ok")
|
|
|
|
text_tools = request_json(
|
|
"POST",
|
|
"/v1/chat/completions",
|
|
{
|
|
"model": TEXT_MODEL,
|
|
"messages": [{"role": "user", "content": "Use the get_test_value tool."}],
|
|
"max_tokens": 128,
|
|
"temperature": 0,
|
|
"tools": [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "get_test_value",
|
|
"description": "Return a deterministic smoke test value.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"label": {"type": "string", "description": "test label"}
|
|
},
|
|
"required": ["label"],
|
|
},
|
|
},
|
|
}
|
|
],
|
|
"tool_choice": {"type": "function", "function": {"name": "get_test_value"}},
|
|
},
|
|
)
|
|
check_tool_call("text-tools", text_tools, strict=args.strict_tools)
|
|
|
|
image_text = request_json(
|
|
"POST",
|
|
"/v1/images/generations",
|
|
{
|
|
"model": IMAGE_MODEL,
|
|
"prompt": "A simple red square icon centered on a white background",
|
|
"size": args.image_size,
|
|
"extra_body": {"response_format": "url"},
|
|
},
|
|
)
|
|
require_ok("image-text-to-image", image_text, ("data",))
|
|
generated_image_url = extract_image_url(image_text)
|
|
|
|
image_edit = None
|
|
edited_image_url = None
|
|
selected_cases = set(args.video_case or [])
|
|
needs_second_image = bool(selected_cases.intersection({"multi-image", "keyframes"}))
|
|
if args.include_image_edit or needs_second_image:
|
|
image_edit = request_json(
|
|
"POST",
|
|
"/v1/images/generations",
|
|
{
|
|
"model": IMAGE_MODEL,
|
|
"prompt": "Turn this into a clean blue square icon while preserving the centered composition",
|
|
"size": args.image_size,
|
|
"extra_body": {"image": [generated_image_url], "response_format": "url"},
|
|
},
|
|
)
|
|
require_ok("image-to-image", image_edit, ("data",))
|
|
edited_image_url = extract_image_url(image_edit)
|
|
|
|
video_common = {
|
|
"model": VIDEO_MODEL,
|
|
}
|
|
for key, value in (
|
|
("height", args.video_height),
|
|
("width", args.video_width),
|
|
("num_frames", args.video_num_frames),
|
|
("frame_rate", args.video_frame_rate),
|
|
):
|
|
if value is not None:
|
|
video_common[key] = value
|
|
video_results = {}
|
|
if "text-to-video" in selected_cases:
|
|
video_results["text_to_video"] = create_video_case(
|
|
"video-text-to-video",
|
|
{
|
|
**video_common,
|
|
"prompt": "A simple cinematic shot of a red square gently moving on a white background",
|
|
},
|
|
args,
|
|
)
|
|
if "image-to-video" in selected_cases:
|
|
video_results["image_to_video"] = create_video_case(
|
|
"video-image-to-video",
|
|
{
|
|
**video_common,
|
|
"prompt": "Animate the icon with subtle floating motion, stable centered composition",
|
|
"image": generated_image_url,
|
|
},
|
|
args,
|
|
)
|
|
if "multi-image" in selected_cases:
|
|
if not edited_image_url:
|
|
raise SystemExit("multi-image test requires an edited image URL")
|
|
video_results["multi_image"] = create_video_case(
|
|
"video-multi-image",
|
|
{
|
|
**video_common,
|
|
"prompt": "Create a smooth transformation from the first icon to the second icon, stable centered composition",
|
|
"extra_body": {"image": [generated_image_url, edited_image_url]},
|
|
},
|
|
args,
|
|
)
|
|
if "keyframes" in selected_cases:
|
|
if not edited_image_url:
|
|
raise SystemExit("keyframes test requires an edited image URL")
|
|
video_results["keyframes"] = create_video_case(
|
|
"video-keyframes",
|
|
{
|
|
**video_common,
|
|
"prompt": "Create a smooth keyframe transition between the two icons, stable centered composition",
|
|
"extra_body": {"image": [generated_image_url, edited_image_url], "mode": "keyframes"},
|
|
},
|
|
args,
|
|
)
|
|
print_json(
|
|
{
|
|
"text": text,
|
|
"text_stream": text_stream,
|
|
"text_tools": text_tools,
|
|
"image_text_to_image": image_text,
|
|
"image_to_image": image_edit,
|
|
"video": video_results,
|
|
}
|
|
)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Call Agnes AI generation APIs.")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
text = sub.add_parser("text", help="Create a chat completion.")
|
|
text.add_argument("--prompt", required=True)
|
|
text.add_argument("--system")
|
|
text.add_argument("--temperature", type=float, default=0.7)
|
|
text.add_argument("--top-p", type=float)
|
|
text.add_argument("--max-tokens", type=int, default=1024)
|
|
text.add_argument("--stream", action="store_true")
|
|
text.add_argument("--tools-json", help="JSON array for OpenAI-compatible tool definitions.")
|
|
text.add_argument("--tool-choice-json", help="JSON object/string for OpenAI-compatible tool_choice.")
|
|
text.add_argument("--raw", action="store_true", help="Print the raw provider response.")
|
|
text.set_defaults(func=cmd_text)
|
|
|
|
image = sub.add_parser("image", help="Generate or edit an image.")
|
|
image.add_argument("--prompt", required=True)
|
|
image.add_argument("--size", default="1024x768")
|
|
image.add_argument("--image", action="append", help="Input image URL. Repeat for multiple images.")
|
|
image.add_argument(
|
|
"--no-translate-prompt",
|
|
action="store_true",
|
|
help="Do not translate non-English prompts before sending to the image API.",
|
|
)
|
|
image.add_argument("--raw", action="store_true", help="Print the raw provider response.")
|
|
image.set_defaults(func=cmd_image)
|
|
|
|
video = sub.add_parser("video", help="Create a video task.")
|
|
video.add_argument("--prompt", required=True)
|
|
video.add_argument("--image", action="append", help="Input image URL. Repeat for multi-image or keyframes.")
|
|
video.add_argument("--mode", choices=("ti2vid", "keyframes"))
|
|
video.add_argument("--height", type=int)
|
|
video.add_argument("--width", type=int)
|
|
video.add_argument("--num-frames", type=int, default=121)
|
|
video.add_argument("--frame-rate", type=float, default=24)
|
|
video.add_argument("--num-inference-steps", type=int)
|
|
video.add_argument("--seed", type=int)
|
|
video.add_argument("--negative-prompt")
|
|
video.add_argument(
|
|
"--no-translate-prompt",
|
|
action="store_true",
|
|
help="Do not translate non-English prompts before sending to the video API.",
|
|
)
|
|
video.add_argument("--poll", action="store_true")
|
|
video.add_argument("--timeout", type=int, default=900)
|
|
video.add_argument("--interval", type=int, default=10)
|
|
video.add_argument("--raw", action="store_true", help="Print the raw provider response.")
|
|
video.set_defaults(func=cmd_video)
|
|
|
|
video_get = sub.add_parser("video-get", help="Retrieve a video result by video_id, or by legacy task_id.")
|
|
video_get.add_argument("identifier", help="Prefer video_id. Legacy task_id is still accepted.")
|
|
video_get.add_argument("--model-name", default=VIDEO_MODEL, help="Model name for video_id result lookup.")
|
|
video_get.add_argument("--no-model-name", action="store_true", help="Do not pass model_name for video_id lookup.")
|
|
video_get.add_argument("--raw", action="store_true", help="Print the raw provider response.")
|
|
video_get.set_defaults(func=cmd_video_get)
|
|
|
|
smoke = sub.add_parser("smoke-test", help="Run live text, image, and video API tests.")
|
|
smoke.add_argument("--image-size", default="1024x768")
|
|
smoke.add_argument("--video-height", type=int)
|
|
smoke.add_argument("--video-width", type=int)
|
|
smoke.add_argument("--video-num-frames", type=int, default=81)
|
|
smoke.add_argument("--video-frame-rate", type=float, default=24)
|
|
smoke.add_argument("--include-image-edit", action="store_true", help="Also test image-to-image editing.")
|
|
smoke.add_argument("--strict-tools", action="store_true", help="Fail if the tool-calling response has no tool_calls.")
|
|
smoke.add_argument("--poll-video", action="store_true")
|
|
smoke.add_argument("--video-timeout", type=int, default=900)
|
|
smoke.add_argument("--video-interval", type=int, default=10)
|
|
smoke.add_argument(
|
|
"--video-case",
|
|
action="append",
|
|
choices=VIDEO_CASES,
|
|
help="Video case to test. Repeat to test multiple cases. Omit to skip video creation.",
|
|
)
|
|
smoke.set_defaults(func=cmd_smoke_test)
|
|
|
|
return parser
|
|
|
|
|
|
def main() -> None:
|
|
args = build_parser().parse_args()
|
|
args.func(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|