完整代码与下载

# 已在当前进程设置 MXB_API_KEY
python mxb-image.py --prompt "白色桌面上的蓝色陶瓷茶壶"

# 指定模型与尺寸
python mxb-image.py --model gemini-3.1-flash-image-preview-1k \
  --size 1536x864 --prompt "极简茶壶产品摄影，横向构图"

# 单张参考图的 JSON 编辑方式；先确认目标模型/分组支持
python mxb-image.py --reference reference.png --prompt "保留主体，把背景改成浅绿色"

# 使用真实 ID 继续查询，不会创建新任务
python mxb-image.py --task "$MXB_TASK_ID" --output result.png

# 已设置 MXB_API_KEY；创建任务前清除旧的 MXB_TASK_ID
unset MXB_TASK_ID
export MXB_PROMPT='白色桌面上的蓝色陶瓷茶壶'
node mxb-image.mjs

# 继续查询已有任务：MXB_TASK_ID 填实际返回的任务 ID
export MXB_TASK_ID='填写实际任务ID'
node mxb-image.mjs

#!/usr/bin/env python3
"""Python 3.10+, standard library only. Submit once; resume with --task."""
import argparse
import base64
import json
import math
import mimetypes
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None  # Never forward API credentials through a redirect.


class APIError(RuntimeError):
    def __init__(self, status, message, retry_after=None):
        super().__init__(f'HTTP {status}: {message}')
        self.status, self.retry_after = status, retry_after


def interval(value, default=3):
    try:
        number = float(value)
        return min(30, max(1, number)) if math.isfinite(number) else default
    except (ValueError, TypeError):
        return default


class ImageClient:
    def __init__(self, key, base_url='https://api.manxiaobai.online'):
        self.key = key
        self.base = base_url.rstrip('/')
        parsed = urllib.parse.urlsplit(self.base)
        if parsed.scheme != 'https' or parsed.username or parsed.password or parsed.path or parsed.query or parsed.fragment:
            raise ValueError('base_url must be an HTTPS origin, without /v1 or credentials')
        self.opener = urllib.request.build_opener(NoRedirect())

    def request(self, path, payload=None, timeout=30):
        if not path.startswith('/v1/') or path.startswith('//'):
            raise ValueError('Only fixed /v1/ API paths are accepted')
        body = None if payload is None else json.dumps(payload, ensure_ascii=False).encode('utf-8')
        request = urllib.request.Request(self.base + path, data=body, headers={
            'Authorization': 'Bearer ' + self.key,
            'Content-Type': 'application/json',
        }, method='GET' if payload is None else 'POST')
        try:
            with self.opener.open(request, timeout=timeout) as response:
                result = json.load(response)
        except urllib.error.HTTPError as error:
            raw = error.read(2048).decode('utf-8', errors='replace')
            raise APIError(error.code, raw, error.headers.get('Retry-After')) from error
        if not isinstance(result, dict):
            raise RuntimeError('API response must be a JSON object')
        return result

    def submit(self, model, prompt, size='1024x1024', reference=None):
        body = {'model': model, 'prompt': prompt, 'size': size, 'n': 1}
        path = '/v1/images/generations/async'
        if reference:
            file = Path(reference)
            mime = mimetypes.guess_type(file.name)[0]
            if mime not in ('image/png', 'image/jpeg', 'image/webp', 'image/bmp'):
                raise ValueError('Use a real PNG/JPEG/WebP/BMP reference image')
            raw = file.read_bytes()
            if not raw or len(raw) > 25 * 1024 * 1024:
                raise ValueError('Reference image must be nonempty and at most 25 MiB; channel limits may be lower')
            body['images'] = [{'image_url': f'data:{mime};base64,' + base64.b64encode(raw).decode('ascii')}]
            path = '/v1/images/edits/async'
        # Intentionally exactly one POST. A timeout may mean submission succeeded.
        result = self.request(path, body, timeout=60)
        task_id = result.get('task_id') or result.get('id')
        if not isinstance(task_id, str) or not re.fullmatch(r'[A-Za-z0-9_-]+', task_id):
            raise RuntimeError('No valid task ID returned. Check logs before submitting again.')
        return task_id, result

    def wait(self, task_id, timeout=1800):
        if not re.fullmatch(r'[A-Za-z0-9_-]+', task_id):
            raise ValueError('Invalid task ID')
        deadline, delay = time.monotonic() + timeout, 3
        while time.monotonic() < deadline:
            try:
                task = self.request('/v1/images/tasks/' + task_id)
            except APIError as error:
                if error.status not in (429, 500, 502, 503, 504):
                    raise
                time.sleep(interval(error.retry_after, delay))
                delay = min(30, delay * 1.5)
                continue
            except (urllib.error.URLError, TimeoutError):
                time.sleep(delay)
                delay = min(30, delay * 1.5)
                continue
            status = str(task.get('status') or task.get('state') or '').lower()
            if task.get('submission_uncertain'):
                raise RuntimeError(f'Submission uncertain; check task {task_id} and billing before any new POST')
            if status in ('completed', 'succeeded', 'success', 'done'):
                nested = task.get('result')
                data = task.get('data') or (nested.get('data') if isinstance(nested, dict) else None)
                if not isinstance(data, list) or not data or not isinstance(data[0], dict):
                    raise RuntimeError(f'Task {task_id} completed without valid image data; do not resubmit automatically')
                return data, task
            if status in ('failed', 'failure', 'error', 'cancelled', 'canceled', 'expired'):
                raise RuntimeError(f'Task {task_id} failed: {json.dumps(task.get("error"), ensure_ascii=False)}')
            delay = interval(task.get('poll_after_ms', 3000) / 1000 if isinstance(task.get('poll_after_ms', 3000), (int, float)) else 3)
            time.sleep(delay)
        raise TimeoutError(f'Stopped waiting. Resume with --task {task_id}; this does not cancel the server task.')


def download_image(item, destination):
    if item.get('url'):
        url = str(item['url'])
        if urllib.parse.urlsplit(url).scheme not in ('https', 'http'):
            raise ValueError('Unsupported result URL scheme')
        # Separate unauthenticated request; never send the API Key to image hosts.
        with urllib.request.urlopen(url, timeout=120) as response:
            if not response.headers.get('Content-Type', '').lower().startswith('image/'):
                raise RuntimeError('Download did not return an image Content-Type')
            data = response.read()
    elif item.get('b64_json'):
        data = base64.b64decode(item['b64_json'], validate=True)
    else:
        raise RuntimeError('Image result has neither URL nor Base64')
    if not data:
        raise RuntimeError('Empty image result')
    Path(destination).write_bytes(data)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--model', default='gpt-image-2.5')
    parser.add_argument('--prompt')
    parser.add_argument('--size', default='1024x1024')
    parser.add_argument('--reference')
    parser.add_argument('--task', help='Resume an existing task; never creates another task')
    parser.add_argument('--timeout', type=int, default=1800)
    parser.add_argument('--output', default='result.png')
    args = parser.parse_args()
    key = os.environ.get('MXB_API_KEY')
    if not key:
        parser.error('Set MXB_API_KEY in the server environment')
    if args.timeout <= 0:
        parser.error('--timeout must be positive')
    if not args.task and not args.prompt:
        parser.error('--prompt is required when creating a task')
    client = ImageClient(key)
    task_id = args.task
    if not task_id:
        try:
            task_id, response = client.submit(args.model, args.prompt, args.size, args.reference)
        except Exception:
            print('Submission was not retried. If transmission started, check logs before another POST.', file=sys.stderr)
            raise
        print('Task:', task_id, flush=True)
        Path(f'{task_id}.json').write_text(json.dumps(response, ensure_ascii=False, indent=2), encoding='utf-8')
    data, final = client.wait(task_id, args.timeout)
    Path(f'{task_id}.result.json').write_text(json.dumps(final, ensure_ascii=False, indent=2), encoding='utf-8')
    download_image(data[0], args.output)
    print('Saved:', args.output, '(file extension does not convert the image format)')


if __name__ == '__main__':
    main()

// Node.js 22+. No external packages. API requests never follow redirects.
import {readFile, writeFile} from 'node:fs/promises';
import {extname} from 'node:path';
import {pathToFileURL} from 'node:url';

const origin = 'https://api.manxiaobai.online';
const pause = ms => new Promise(resolve => setTimeout(resolve, ms));
const delay = value => Number.isFinite(Number(value)) ? Math.max(1000, Math.min(30000, Number(value))) : 3000;
export class APIError extends Error {
  constructor(status, body, retryAfter) {super(`HTTP ${status}: ${body.slice(0, 2048)}`); this.status=status; this.retryAfter=retryAfter;}
}
export class ImageClient {
  constructor(key) {if(!key) throw new Error('Set MXB_API_KEY'); this.key=key;}
  async request(path, body, timeout=30000) {
    if(!path.startsWith('/v1/')) throw new Error('Expected a fixed /v1/ path');
    const response=await fetch(origin+path, {
      method:body===undefined?'GET':'POST', redirect:'error',
      headers:{Authorization:`Bearer ${this.key}`, 'Content-Type':'application/json'},
      body:body===undefined?undefined:JSON.stringify(body), signal:AbortSignal.timeout(timeout)
    });
    if(!response.ok) throw new APIError(response.status,await response.text(),response.headers.get('retry-after'));
    const result=await response.json();
    if(!result||typeof result!=='object'||Array.isArray(result)) throw new Error('Invalid API response object');
    return result;
  }
  async submit({model='gpt-image-2.5',prompt,size='1024x1024',reference}) {
    if(!prompt) throw new Error('Set MXB_PROMPT when creating a task');
    const body={model,prompt,size,n:1}; let path='/v1/images/generations/async';
    if(reference) {
      const types={'.png':'image/png','.jpg':'image/jpeg','.jpeg':'image/jpeg','.webp':'image/webp','.bmp':'image/bmp'};
      const mime=types[extname(reference).toLowerCase()];
      if(!mime) throw new Error('Use a real PNG/JPEG/WebP/BMP reference image');
      const bytes=await readFile(reference);
      if(!bytes.length||bytes.length>25*1024*1024) throw new Error('Reference must be nonempty and at most 25 MiB; channel limits may be lower');
      body.images=[{image_url:`data:${mime};base64,${bytes.toString('base64')}`}]; path='/v1/images/edits/async';
    }
    const task=await this.request(path,body,60000); // Exactly one POST, no automatic retry.
    const id=task.task_id||task.id;
    if(typeof id!=='string'||!/^[A-Za-z0-9_-]+$/.test(id)) throw new Error('No valid task ID. Check logs before submitting again.');
    return {id,task};
  }
  async wait(id,timeout=1800000) {
    if(!/^[A-Za-z0-9_-]+$/.test(id)) throw new Error('Invalid task ID');
    const deadline=Date.now()+timeout; let interval=3000;
    while(Date.now()<deadline) {
      let task;
      try {task=await this.request('/v1/images/tasks/'+encodeURIComponent(id));}
      catch(error) {
        if(error instanceof APIError && ![429,500,502,503,504].includes(error.status)) throw error;
        if(!(error instanceof APIError)&& !['TypeError','TimeoutError','AbortError'].includes(error.name)) throw error;
        const wait=error.retryAfter===null||error.retryAfter===undefined?interval:delay(Number(error.retryAfter)*1000);
        await pause(wait); interval=Math.min(30000,interval*1.5); continue;
      }
      const state=String(task.status||task.state||'').toLowerCase();
      if(task.submission_uncertain) throw new Error(`Submission uncertain. Check task ${id} and billing before another POST.`);
      if(['completed','succeeded','success','done'].includes(state)) {
        const data=task.data?.length?task.data:task.result?.data;
        if(!Array.isArray(data)||!data.length||!data[0]||typeof data[0]!=='object') throw new Error(`Task ${id} completed without valid image data`);
        return {data,task};
      }
      if(['failed','failure','error','cancelled','canceled','expired'].includes(state)) throw new Error(`Task ${id} failed: ${JSON.stringify(task.error)}`);
      interval=delay(task.poll_after_ms??3000); await pause(interval);
    }
    throw new Error(`Stopped waiting. Resume with MXB_TASK_ID=${id}; server task is not cancelled.`);
  }
}
export async function downloadImage(item,output) {
  let bytes;
  if(item.url) {
    const url=new URL(item.url); if(!['https:','http:'].includes(url.protocol)) throw new Error('Unsupported image URL');
    const response=await fetch(url,{signal:AbortSignal.timeout(120000)}); // No Authorization header.
    if(!response.ok||!response.headers.get('content-type')?.toLowerCase().startsWith('image/')) throw new Error('Image download failed or returned a non-image');
    bytes=Buffer.from(await response.arrayBuffer());
  } else if(item.b64_json) bytes=Buffer.from(item.b64_json,'base64');
  else throw new Error('No image URL or Base64 data');
  if(!bytes.length) throw new Error('Empty image');
  await writeFile(output,bytes);
}
async function main() {
  const client=new ImageClient(process.env.MXB_API_KEY);
  let id=process.env.MXB_TASK_ID;
  if(!id) {
    let created;
    try {created=await client.submit({model:process.env.MXB_MODEL||'gpt-image-2.5',prompt:process.env.MXB_PROMPT,size:process.env.MXB_SIZE||'1024x1024',reference:process.env.MXB_REFERENCE});}
    catch(error) {console.error('Submission not retried. Check logs before another POST if transmission started.');throw error;}
    id=created.id; console.log('Task:',id);
    await writeFile(`${id}.json`,JSON.stringify(created.task,null,2));
  }
  const {data,task}=await client.wait(id);
  await writeFile(`${id}.result.json`,JSON.stringify(task,null,2));
  const output=process.env.MXB_OUTPUT||'result.png'; await downloadImage(data[0],output);
  console.log('Saved:',output,'(extension does not convert image format)');
}
if(process.argv[1]&&import.meta.url===pathToFileURL(process.argv[1]).href) main().catch(error=>{console.error(error.message);process.exitCode=1;});