# PixZoo: generate images and videos from one link Use this guide to complete the user's media request. Reply in the user's language. You need network access, Python 3.9+ (standard library only), and permission to read/write the requested local files. No repository, package installation, skill, CLI, or MCP connection is required. ## Quick start 1. Reuse saved PixZoo tool credentials. If PIXZOO_API_KEY is already set, it remains supported; otherwise use the downloadable Python client below to start browser authorization. Show the user its authorization link and code, let them sign in and approve, and keep the client polling while they do so. Do not approve on the user's behalf or read their browser cookies. The client automatically stores credentials privately and renews them in the same persistent execution environment. No environment variable configuration is needed. Never print tokens, put them in code or command arguments, or ask the user to paste them in chat. Generating media spends the user's existing PixZoo credits. 2. Read [current models](/openapi/models.json) to choose a site-enabled canonical model ID and supported parameters. If unspecified, default to nano-banana-2 for images or seedance-2.0 for videos. Use that model's defaults, and generate one result unless requested otherwise. If the model is unavailable or cannot satisfy the input, ask the user rather than silently substituting models or dropping references. Pricing rules are informational; the existing server determines the actual charge. Display credit amounts rounded up to integers. 3. For a local reference file, use POST /api/storage/presign/ with filename, contentType, size, and purpose="media"; PUT the original bytes to data.uploadUrl using only data.headers; then POST /api/storage/complete/ with uploadId and use data.url. Respect the uploadTypes MIME/size limits and selected model's tighter reference limits. Preserve reference order and firstFrame/lastFrame/reference roles. Existing reachable HTTP(S) URLs need no upload. A local path or file:// URL is not a generation input. Ordinary uploads do not create material-library records. Never forward the PixZoo key to storage. Audio/video references are usable only with models that explicitly support them; speech/music generation is outside this guide. 4. Submit POST /api/ai/generate/ once, using a canonical schemaVersion: 1 request. Save the returned data.id immediately to a local recovery file; this is the SITE task ID. Do not use data.taskId or an upstream provider ID. Submission is NOT idempotent: on timeout, disconnect, malformed response, or ambiguous failure, do not resubmit. Check the user's task history to identify the existing task first. callbackUrl is not implemented. 5. POST /api/ai/query/ with {"taskId": "the saved site ID"} every 10 seconds. queuing/pending/processing are in progress; success is terminal success; failed/canceled are terminal failures. Stop on unknown status. Default waiting limits: images 10 minutes, videos 30 minutes. On timeout, retain the ID and resume querying; do not generate again. 6. On success, use data.outputImages as the original result URL list for BOTH images and videos (legacy field name). If empty, JSON-parse data.taskInfo and read images[*].imageUrl, as the existing backend does. data.mediaType determines the kind; a video URL in outputImages is still a video. Do not scan arbitrary provider payload fields. Treat success without usable outputs as an error, not permission to regenerate. Download all requested original results without API authorization headers, preserve MIME-derived extensions, and report the local paths. Retry only failed downloads from the saved URL snapshot, never generation. URLs may expire; query the same task again if a saved URL can no longer be used. Media API requests use Authorization: Bearer with the tool access token (or PIXZOO_API_KEY) and JSON Content-Type. Check BOTH HTTP status AND JSON code: code 0 is success; code -1 is a business error even on HTTP 200. On invalid credentials, insufficient credits, invalid parameters, or forbidden/missing tasks, stop and explain the issue. Never print raw authenticated responses, signed upload URLs, or provider diagnostics to chat. Return useful status and local file paths; share result URLs only when needed for the user's requested delivery. ## Credential permissions Existing API keys allow uploads, media generation (including image processing), and querying only the owning user’s tasks. They cannot purchase or recharge credits, change account settings, or create/manage API keys. Those account actions require the user’s interactive login. Generation still spends existing account credits; a key does not grant free generation. Browser-authorized tool credentials obey the same boundary. Only an interactive, same-origin signed-in user can approve or revoke tool access; API keys and tool tokens cannot manage authorizations. ## Sign in once; no manual API key configuration Download [the Python client](/api/openapi/client/) from the SAME deployment as this guide and save it as pixzoo_example.py in your working directory. It is the exact source embedded below; no SDK installation is needed. Review it before execution. Run python3 pixzoo_example.py to authorize first, then use the generation snippets below. The client always sends credentials and paid media requests to https://pixzoo.ai, including when this guide is read on a preview deployment. Never change BASE to an untrusted origin. The client displays only a PixZoo verification URL and a user code. Present both to the user and keep the authorization process alive while they sign in and approve. Never approve for them. The URL carries only the user code, never the secret device code. Approval is not implied by login. Access tokens last 15 minutes; refresh tokens rotate after each use, expire after 30 days without renewal, and cannot extend the grant beyond 90 days. Revocation blocks both access and renewal immediately. Manage grants at https://pixzoo.ai/openapi/#tool-authorizations. Credentials are stored in the execution user's private config directory: XDG_CONFIG_HOME/pixzoo (or ~/.config/pixzoo), or APPDATA/pixzoo on Windows. The directory is 0700 and the credential file is 0600 on POSIX. File permissions on Windows follow the user's profile ACL. Do not display, copy to chat, commit, or put these files in shared workspaces. Processes sharing these credentials must use the same client lock; never race refresh requests. New computers, temporary cloud environments without persistence, expired or revoked grants need fresh approval. To explicitly reconnect, call p.login(force=True); an environment API key takes precedence and must be replaced separately if invalid. If a token response is lost, do not replay the device exchange or refresh token blindly. The client preserves unknown refresh state and requires a new authorization on the next run. A device exchange may already have issued a grant; revoke unused grants in the website. None of these recovery operations resubmit media generation. For other clients: POST /api/agent/device/ with client_id="pixzoo-agent", client_name (a label, not verified identity), and scope="media:upload media:generate tasks:read". Display verification_uri_complete AND user_code. Poll POST /api/agent/token/ with client_id, grant_type="urn:ietf:params:oauth:grant-type:device_code", and secret device_code at the returned interval. On authorization_pending continue; on slow_down add at least 5 seconds permanently; stop on access_denied or expiry. Refresh with grant_type="refresh_token" and refresh_token. Persist the replacement refresh token atomically. These endpoints accept form-urlencoded or JSON and return plain OAuth JSON with HTTP error statuses, not the media API code/data envelope. They do not receive existing API keys or login cookies. See /openapi.json for their schemas. ## Upload API reference Both metadata endpoints require Authorization: Bearer from PIXZOO_API_KEY and Content-Type: application/json. They accept JSON metadata, not multipart file bytes. Existing HTTP(S) reference URLs can be used directly without uploading again. ### 1. POST /api/storage/presign/ | Field | Type | Required | Meaning | | --- | --- | --- | --- | | filename | string | yes | Original filename, 1–255 characters | | contentType | string | yes | MIME type from uploadTypes in the current model directory | | size | integer | yes | Actual original file length in bytes, greater than zero | | purpose | string | use media | Ordinary upload; no material-library record | Request: ```json { "filename": "reference.png", "contentType": "image/png", "size": 1024, "purpose": "media" } ``` Response example (URLs, signature, timestamps and IDs are placeholders): ```json { "code": 0, "message": "ok", "data": { "uploadId": "11111111-1111-4111-8111-111111111111", "uploadUrl": "https://storage.example.test/pending/reference.png?signature=example", "headers": { "Content-Type": "image/png" }, "expiresAt": "2030-01-01T00:05:00.000Z" } } ``` ### 2. PUT data.uploadUrl Send the original file bytes as the entire request body to the returned storage URL. Use exactly data.headers (and the actual Content-Length if required by your HTTP client). Do not send JSON, multipart wrappers, cookies, or the PixZoo API key to storage. Check the PUT HTTP status before completing. A successful PUT alone is not a usable final reference URL. ### 3. POST /api/storage/complete/ The only request field is uploadId (required UUID), returned by presign. Use the same PixZoo API key user that reserved the upload. ```json { "uploadId": "11111111-1111-4111-8111-111111111111" } ``` Response: ```json { "code": 0, "message": "ok", "data": { "uploadId": "11111111-1111-4111-8111-111111111111", "key": "uploads/example/reference.png", "url": "https://media.example.test/uploads/reference.png", "filename": "reference.png", "size": 1024, "materialId": null } } ``` Use data.url in options.referenceImages (objects with url and optional role), referenceVideos, referenceAudios or maskUrl according to the model capability. Never use the temporary uploadUrl as a generation input. Completion is idempotent: after a lost completion response, retry the same uploadId without uploading again. Presigned PUT URLs expire after 5 minutes; the completion window is 1 hour. The service enforces MIME, original file size, signatures, ownership and account upload quotas. See [current upload limits](/openapi/models.json) for exact MIME-specific maximum bytes; the selected model can impose tighter reference limits. ## Runnable workflow Save the following block as pixzoo_example.py in the task's working directory. It defines functions and performs no requests when imported. Choose ONE of the examples below to execute; do not run every example. The API origin is fixed to https://pixzoo.ai. If a request fails, report a safe error summary rather than an unfiltered Python traceback containing a signed URL. Recovery state may contain private result URLs: keep it local. ```python import json import mimetypes import os from pathlib import Path import tempfile import time from urllib.error import HTTPError, URLError from urllib.parse import urlsplit from urllib.request import Request, build_opener, HTTPRedirectHandler, urlopen BASE = "https://pixzoo.ai" class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None # Never forward credentials through redirects, including to another origin. API_OPENER = build_opener(NoRedirect()) class ApiError(Exception): def __init__(self, status, code=None, message=""): reason = {401: "authentication required", 402: "insufficient credits", 403: "permission denied", 429: "rate limited"}.get(status, "request rejected") # Report only known safe categories, never raw provider messages or URLs. if isinstance(message, str): for prefix in ("insufficient credits", "no auth", "no permission", "task not found", "invalid params"): if message.lower().startswith(prefix): reason = prefix break super().__init__("PixZoo: %s (HTTP %s, code %s)" % (reason, status, code)) self.status = status def api(path, body=None, retry=False, authenticated=True): if not path.startswith("/") or path.startswith("//"): raise ValueError("Expected an API path") headers = {"Accept": "application/json"} if authenticated: key = get_access_token() headers["Authorization"] = "Bearer " + key data = None if body is None else json.dumps(body).encode("utf-8") if data is not None: headers["Content-Type"] = "application/json" for attempt in range(3 if retry else 1): try: with API_OPENER.open(Request(BASE + path, data=data, headers=headers), timeout=180) as response: payload = json.load(response) if not isinstance(payload, dict) or payload.get("code") != 0: raise ApiError(response.status, payload.get("code") if isinstance(payload, dict) else None, payload.get("message", "") if isinstance(payload, dict) else "") return payload["data"] except HTTPError as error: if not retry or error.code not in (429, 500, 502, 503, 504) or attempt == 2: raise ApiError(error.code) from None except (URLError, TimeoutError, OSError): if not retry or attempt == 2: raise RuntimeError("Network request interrupted; do not resubmit generation") from None time.sleep(2 ** attempt) def catalog(): # Public documents are plain JSON, not the authenticated API envelope. with API_OPENER.open(BASE + "/openapi/models.json", timeout=30) as response: return json.load(response) def upload_file(filename, content_type=None): path = Path(filename) content_type = content_type or mimetypes.guess_type(path.name)[0] policy = catalog()["uploadTypes"].get(content_type) size = path.stat().st_size if not policy or not 0 < size <= policy["maxBytes"]: raise ValueError("Unsupported reference file type or size; check the model and upload limits") signed = api("/api/storage/presign/", { "filename": path.name, "contentType": content_type, "size": size, "purpose": "media", }) if urlsplit(signed["uploadUrl"]).scheme != "https": raise RuntimeError("Expected an HTTPS upload URL") headers = dict(signed["headers"]) if any(key.lower() == "authorization" for key in headers): raise RuntimeError("Unexpected upload authorization header") headers["Content-Length"] = str(size) try: with path.open("rb") as source: # Only storage-provided headers; NEVER attach PIXZOO_API_KEY here. request = Request(signed["uploadUrl"], data=source, headers=headers, method="PUT") with API_OPENER.open(request, timeout=900): pass except (URLError, TimeoutError, OSError): raise RuntimeError("Upload interrupted before generation; uploadId=" + signed["uploadId"]) from None # Completion is idempotent and may be retried without uploading again. try: return api("/api/storage/complete/", {"uploadId": signed["uploadId"]}, retry=True)["url"] except Exception: raise RuntimeError("Upload completion interrupted; retry complete with uploadId=" + signed["uploadId"]) from None def save_state(directory, state): path = Path(directory) / "task.json" temporary = path.with_suffix(".tmp") with temporary.open("w", encoding="utf-8") as target: os.chmod(temporary, 0o600) json.dump(state, target, ensure_ascii=False, indent=2) temporary.replace(path) def submit(request, parent="."): if request.get("schemaVersion") != 1 or request.get("mediaType") not in ("image", "video"): raise ValueError("Use a canonical v1 image or video request") get_access_token() # Finish authorization before creating an unknown-submission marker. # The directory is unique. Never reuse another request's state or overwrite results. directory = Path(tempfile.mkdtemp(prefix="pixzoo-", dir=parent)).resolve() state = {"mediaType": request["mediaType"], "status": "submission_unknown"} save_state(directory, state) # Exactly ONE submission attempt. Any exception leaves a durable unknown marker. try: result = api("/api/ai/generate/", request) except Exception as error: message = str(error) if isinstance(error, ApiError) else "Submission outcome unknown" raise RuntimeError(message + "; do not resubmit. Recovery directory: " + str(directory)) from None task_id = result.get("id") if isinstance(result, dict) else None if not isinstance(task_id, str) or not task_id: raise RuntimeError("Submission outcome unknown; inspect task history, do not resubmit") state.update({"id": task_id, "status": result.get("status", "unknown")}) save_state(directory, state) return str(directory) def output_urls(task): # Current backend uses these fields for BOTH images and videos. # outputImages is the stored original-media URL list, despite its legacy name. urls = task.get("outputImages") if not isinstance(urls, list) or not urls: info = task.get("taskInfo") info = json.loads(info) if isinstance(info, str) else info entries = info.get("images", []) if isinstance(info, dict) else [] urls = [entry.get("imageUrl") for entry in entries if isinstance(entry, dict)] if not isinstance(urls, list): return [] if any(not isinstance(url, str) or urlsplit(url).scheme != "https" for url in urls): raise RuntimeError("Unexpected output URL; inspect the saved task, do not drop outputs or regenerate") return list(dict.fromkeys(urls)) def wait(directory, timeout=None, interval=10): directory = Path(directory) state = json.loads((directory / "task.json").read_text(encoding="utf-8")) task_id = state.get("id") if not task_id: raise RuntimeError("No known task ID: inspect task history; do not resubmit") timeout = timeout if timeout is not None else (1800 if state["mediaType"] == "video" else 600) deadline = time.monotonic() + timeout while time.monotonic() < deadline: task = api("/api/ai/query/", {"taskId": task_id}, retry=True) if task.get("id") != task_id or task.get("mediaType") != state["mediaType"]: raise RuntimeError("Unexpected task identity or media type") status = task.get("status") if status not in ("queuing", "pending", "processing", "success", "failed", "canceled"): raise RuntimeError("Unknown task status; stop and inspect") state["status"] = status if status == "success": state["urls"] = output_urls(task) save_state(directory, state) if not state["urls"]: raise RuntimeError("Task succeeded without usable outputs; do not regenerate") return state save_state(directory, state) if status in ("failed", "canceled"): raise RuntimeError("Task " + status + "; do not automatically generate again") time.sleep(max(10, interval)) raise TimeoutError("Task is still running. Resume wait() using the SAME directory") def download(directory): directory = Path(directory) state = json.loads((directory / "task.json").read_text(encoding="utf-8")) urls = state.get("urls", []) if state.get("status") != "success" or not urls: raise RuntimeError("Wait for successful outputs before downloading") files = state.setdefault("files", {}) extensions = {"image/png": ".png", "image/jpeg": ".jpg", "image/webp": ".webp", "image/gif": ".gif", "image/avif": ".avif", "image/svg+xml": ".svg", "video/mp4": ".mp4", "video/webm": ".webm", "video/quicktime": ".mov"} for index, url in enumerate(urls): previous = files.get(str(index)) if previous and (directory / previous).is_file(): continue # Retry only missing downloads, using the saved output snapshot. if urlsplit(url).scheme != "https": raise RuntimeError("Expected HTTPS media URL") for attempt in range(3): partial = directory / ("output-%s.part" % index) try: # No API authorization header on downloads or their redirects. with urlopen(Request(url, headers={"Accept": state["mediaType"] + "/*"}), timeout=180) as response: mime = response.headers.get_content_type() if not mime.startswith(state["mediaType"] + "/") or mime not in extensions: raise RuntimeError("Unexpected result MIME type; keep task ID and inspect") name = "output-%s%s" % (index, extensions[mime]) destination = directory / name if destination.exists(): raise FileExistsError("Refusing to overwrite an existing result") with partial.open("xb") as target: while True: chunk = response.read(1024 * 1024) if not chunk: break target.write(chunk) if partial.stat().st_size == 0: raise RuntimeError("Empty result download") declared = response.headers.get("Content-Length") if declared is not None and partial.stat().st_size != int(declared): raise OSError("Incomplete download") # Hard-link creates exclusively rather than replacing an existing file. os.link(partial, destination) files[str(index)] = name save_state(directory, state) break except HTTPError as error: if error.code not in (429, 500, 502, 503, 504) or attempt == 2: raise RuntimeError("Media download failed (HTTP %s); retry download(), not generation" % error.code) from None except (URLError, TimeoutError, OSError): if attempt == 2: raise RuntimeError("Media download interrupted; retry download(), not generation") from None finally: partial.unlink(missing_ok=True) time.sleep(2 ** attempt) return [str(directory / files[str(index)]) for index in range(len(urls))] # Browser authorization. Credentials stay in this execution environment, never # in the task directory, chat, repository, browser storage, or command arguments. import contextlib import re import stat from urllib.parse import urlencode CLIENT_ID = "pixzoo-agent" SCOPE = "media:upload media:generate tasks:read" DEVICE_GRANT = "urn:ietf:params:oauth:grant-type:device_code" class AuthError(Exception): def __init__(self, code): self.code = code super().__init__("PixZoo authorization: " + code) def auth_post(operation, body): request = Request(BASE + "/api/agent/" + operation + "/", data=urlencode(body).encode(), headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") try: with API_OPENER.open(request, timeout=30) as response: result = json.load(response) except HTTPError as error: try: code = json.load(error).get("error") except Exception: code = None known = ("authorization_pending", "slow_down", "expired_token", "access_denied", "invalid_grant", "credential_limit_reached", "invalid_client", "invalid_scope") raise AuthError(code if code in known else "request_failed") from None except (URLError, TimeoutError, OSError, ValueError): # Token issuance/rotation is never blindly retried after a lost response. raise RuntimeError("Authorization response interrupted. Run login again; never print credential files.") from None if not isinstance(result, dict): raise RuntimeError("Invalid authorization response") return result def credential_directory(): root = Path(os.environ.get("APPDATA") or os.environ.get("XDG_CONFIG_HOME") or (Path.home() / ".config")) directory = root / "pixzoo" if directory.is_symlink(): raise RuntimeError("Credential directory must not be a symlink") directory.mkdir(mode=0o700, parents=True, exist_ok=True) info = directory.stat() if hasattr(os, "getuid") and info.st_uid != os.getuid(): raise RuntimeError("Credential directory belongs to another user") directory.chmod(0o700) return directory @contextlib.contextmanager def credential_lock(directory): path = directory / "credentials.lock" if path.is_symlink(): raise RuntimeError("Credential lock must not be a symlink") fd = os.open(path, os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0), 0o600) try: info = os.fstat(fd) if not stat.S_ISREG(info.st_mode) or (hasattr(os, "getuid") and info.st_uid != os.getuid()): raise RuntimeError("Invalid credential lock file") if os.name == "nt": import msvcrt if info.st_size == 0: os.write(fd, b"0") os.lseek(fd, 0, os.SEEK_SET) try: msvcrt.locking(fd, msvcrt.LK_NBLCK, 1) except OSError: raise RuntimeError("Another PixZoo process is authorizing or refreshing; wait for it to finish") from None else: import fcntl try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) except OSError: raise RuntimeError("Another PixZoo process is authorizing or refreshing; wait for it to finish") from None yield finally: # Closing releases the OS lock even after a crash; never unlink the inode. os.close(fd) def read_credentials(directory): path = directory / "credentials.json" if not path.exists() and not path.is_symlink(): return {} if path.is_symlink(): raise RuntimeError("Credential file must not be a symlink") fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) with os.fdopen(fd, "r", encoding="utf-8") as source: info = os.fstat(source.fileno()) if not stat.S_ISREG(info.st_mode) or info.st_size > 16384: raise RuntimeError("Invalid credential file") if hasattr(os, "getuid") and (info.st_uid != os.getuid() or info.st_mode & 0o077): raise RuntimeError("Credential file must be owned by you and readable only by you (0600)") data = json.load(source) if not isinstance(data, dict) or data.get("issuer") != BASE or data.get("version") != 1: raise RuntimeError("Invalid credential file issuer or version") return data def write_credentials(directory, data): data = dict(data, version=1, issuer=BASE) fd, temporary = tempfile.mkstemp(prefix=".credentials-", dir=directory) try: with os.fdopen(fd, "w", encoding="utf-8") as target: os.chmod(temporary, 0o600) json.dump(data, target) target.flush() os.fsync(target.fileno()) os.replace(temporary, directory / "credentials.json") finally: if os.path.exists(temporary): os.unlink(temporary) def token_record(result): for field, prefix in (("access_token", "pz_at_"), ("refresh_token", "pz_rt_")): if not isinstance(result.get(field), str) or not re.fullmatch(prefix + r"[A-Za-z0-9_-]{43}", result[field]): raise RuntimeError("Invalid token response") if result.get("scope") != SCOPE or result.get("token_type") != "Bearer": raise RuntimeError("Unexpected token permissions") lifetime = result.get("expires_in") refresh_lifetime = result.get("refresh_expires_in") if (type(lifetime) is not int or not 0 < lifetime <= 900 or type(refresh_lifetime) is not int or not 0 < refresh_lifetime <= 30 * 86400): raise RuntimeError("Invalid token lifetime") return {"access_token": result["access_token"], "refresh_token": result["refresh_token"], "expires_at": time.time() + lifetime, "refresh_expires_at": time.time() + refresh_lifetime} def authorize_device(directory, saved): pending = saved.get("pending", {}) if not isinstance(pending, dict) or pending.get("expires_at", 0) <= time.time(): response = auth_post("device", {"client_id": CLIENT_ID, "client_name": "PixZoo Python tool", "scope": SCOPE}) code = response.get("device_code") user_code = response.get("user_code") if (not isinstance(code, str) or not re.fullmatch(r"pz_dc_[A-Za-z0-9_-]{43}", code) or not isinstance(user_code, str) or not re.fullmatch(r"[23456789ABCDEFGHJKLMNPQRSTUVWXYZ]{4}(?:-[23456789ABCDEFGHJKLMNPQRSTUVWXYZ]{4})", user_code) or type(response.get("expires_in")) is not int or not 0 < response["expires_in"] <= 600): raise RuntimeError("Invalid device authorization response") pending = {"device_code": code, "user_code": user_code, "expires_at": time.time() + response["expires_in"], "interval": 5} write_credentials(directory, {"pending": pending}) # Build the user-facing URL ourselves. Never follow a server-provided login host. print("Open this PixZoo link, sign in, and approve the request you initiated:", flush=True) print(BASE + "/openapi/authorize/?" + urlencode({"user_code": pending["user_code"]}), flush=True) print("Check authorization code: " + pending["user_code"], flush=True) interval = max(5, pending.get("interval", 5)) while time.time() + interval < pending["expires_at"]: time.sleep(interval) write_credentials(directory, {"device_exchange_unknown": True}) try: response = auth_post("token", {"client_id": CLIENT_ID, "grant_type": DEVICE_GRANT, "device_code": pending["device_code"]}) except AuthError as error: if error.code == "authorization_pending": write_credentials(directory, {"pending": pending}) continue if error.code == "slow_down": interval += 5 pending["interval"] = interval write_credentials(directory, {"pending": pending}) continue write_credentials(directory, {}) raise record = token_record(response) write_credentials(directory, record) return record["access_token"] write_credentials(directory, {}) raise AuthError("expired_token") def get_access_token(force=False): # Developer keys remain compatible. Ordinary users need no environment setup. existing = os.environ.get("PIXZOO_API_KEY", "").strip() if existing: return existing directory = credential_directory() with credential_lock(directory): saved = read_credentials(directory) if force: write_credentials(directory, {}) saved = {} if saved.get("access_token") and saved.get("expires_at", 0) > time.time() + 60: return saved["access_token"] if saved.get("refresh_token") and saved.get("refresh_expires_at", 0) > time.time() + 60: # Persist before sending. An interrupted rotation must never replay # an old refresh token and revoke a successfully renewed family. write_credentials(directory, {"refresh_unknown": True}) try: result = auth_post("token", {"client_id": CLIENT_ID, "grant_type": "refresh_token", "refresh_token": saved["refresh_token"]}) except AuthError as error: if error.code != "invalid_grant": raise else: record = token_record(result) write_credentials(directory, record) return record["access_token"] return authorize_device(directory, saved) def login(force=False): get_access_token(force=force) print("PixZoo authorization ready.") if __name__ == "__main__": login() ``` ### Text to image ```python import pixzoo_example as p request = { "schemaVersion": 1, "mediaType": "image", "model": "nano-banana-2", "prompt": "A red panda reading a book, soft natural light", "scene": "text-to-image", "options": { "aspectRatio": "1:1" } } directory = p.submit(request) print("Recovery directory:", directory) # Keep this even if waiting is interrupted. p.wait(directory) print(p.download(directory)) ``` ### Edit a local image ```python import pixzoo_example as p reference_url = p.upload_file("reference.png") request = { "schemaVersion": 1, "mediaType": "image", "model": "nano-banana-2", "prompt": "A red panda reading a book, soft natural light", "scene": "text-to-image", "options": { "aspectRatio": "1:1" } } request["scene"] = "image-to-image" request["prompt"] = "Keep the subject; change the background to a sunny garden" request["options"]["referenceImages"] = [{"url": reference_url, "role": "reference"}] directory = p.submit(request) print("Recovery directory:", directory) p.wait(directory) print(p.download(directory)) ``` ### Text to video ```python import pixzoo_example as p request = { "schemaVersion": 1, "mediaType": "video", "model": "seedance-2.0", "prompt": "A red panda reading a book, slow camera dolly, soft natural light", "scene": "text-to-video", "options": { "aspectRatio": "16:9" } } directory = p.submit(request) print("Recovery directory:", directory) p.wait(directory) print(p.download(directory)) ``` ### Animate a local image ```python import pixzoo_example as p reference_url = p.upload_file("reference.png") request = { "schemaVersion": 1, "mediaType": "video", "model": "seedance-2.0", "prompt": "A red panda reading a book, slow camera dolly, soft natural light", "scene": "text-to-video", "options": { "aspectRatio": "16:9" } } request["scene"] = "image-to-video" request["prompt"] = "Animate the subject naturally, slow camera movement" request["options"]["referenceImages"] = [{"url": reference_url, "role": "firstFrame"}] directory = p.submit(request) print("Recovery directory:", directory) p.wait(directory) print(p.download(directory)) ``` For additional references, upload each once and place URLs in canonical options.referenceImages, referenceVideos or referenceAudios as supported by the selected model. Upload masks similarly and use options.maskUrl only on models with mask support. Check durationOptions, resolutionAspectRatios, supportedQualities, maxReferenceImages, maxOutputCount, and features before adding parameters. Do not guess provider-specific option aliases. ## Resume without generating again ```python import pixzoo_example as p directory = "the-existing-recovery-directory" p.wait(directory, timeout=1800) # Query the SAME ID, even after a process restart. print(p.download(directory)) # If only saving files failed and successful URLs are already in task.json: # print(p.download(directory)) # Skips previously saved outputs. ``` If you know the site task ID but have no recovery directory, create a new local directory, call p.save_state(directory, {"id": site_task_id, "mediaType": "image" or "video", "status": "processing"}), then call p.wait(directory). If task.json has status submission_unknown and no id, inspect task history at https://pixzoo.ai/activity/ai-tasks/ and identify the task before continuing. Do not infer failure from a network error. If storage PUT completed but completion was interrupted, retry only POST /api/storage/complete/ with the same uploadId. Signed uploads expire after 5 minutes and completion must occur within the existing 1-hour window. For a lost PUT response, try complete first; if the file cannot be verified, upload again before generation. The existing service enforces upload types, sizes, ownership, signatures, and quotas. ## Existing endpoints - POST /api/ai/generate/ — Submit one image or video generation task. Never automatically retry a submission. - POST /api/ai/query/ — Query a task owned by this API key user. Poll every 10 seconds; this does not generate again. - POST /api/storage/presign/ — Reserve an ordinary media upload. PUT original file bytes to the returned uploadUrl using only its returned headers. - POST /api/storage/complete/ — Confirm an uploaded file and obtain its final URL. Idempotent for the same uploadId and owner. See [OpenAPI 3.1](/openapi.json) for request/response schemas and [current models](/openapi/models.json) for capabilities and upload limits. These documents do not grant permission to create API keys, purchase credits, or change account settings.