火山即梦 AI 4.0：提交图片生成任务

火山即梦图片生成 4.0：POST ...?Action=CVSync2AsyncSubmitTask&Version=2022-08-31，body req_key=jimeng_t2i_v40 + prompt（建议 force_single 控成本）。

先看清成品

• mock 返回 task_id。
• 1 项 pytest 通过。

合规提醒：即梦网页与火山 API 均须遵守平台规则与国内 AI 生成合成内容标识要求。不上传未授权肖像/作品，不做未经同意的换脸仿声，不生成违法或伪造新闻画面。商用以会员/合同为准。教学客户端用 X-Access-Key 对接 mock；生产请用火山 SDK 做 Signature V4。本文不编造价格。

运行说明：无正式 VOLC AK/SK。A/B 用 mock 提交与轮询；未对 visual.volcengineapi.com 做真实签名调用。C 为发布前标识自检。2026-10-23（Asia/Shanghai）pytest 通过。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

113-jimeng-ai-A-src/
├── jimeng_submit.py
├── mock_jimeng.py
└── test_113a.py

第 1 步：客户端

#!/usr/bin/env python3
"""即梦 AI 图片生成 4.0 提交任务（火山智能视觉）。

正式环境：POST https://visual.volcengineapi.com?Action=CVSync2AsyncSubmitTask&Version=2022-08-31
鉴权为火山 Signature V4（Region=cn-north-1, Service=cv），教学客户端用 X-Access-Key 对接 mock；
生产请用官方 SDK 签名，勿把 SK 写进仓库。

Body: req_key=jimeng_t2i_v40, prompt, 可选 force_single / width / height / size。
"""
from __future__ import annotations
import argparse, json, os, sys, urllib.error, urllib.parse, urllib.request


def submit(base, access_key, prompt, *, force_single=True, width=None, height=None, size=None):
    qs = urllib.parse.urlencode({"Action": "CVSync2AsyncSubmitTask", "Version": "2022-08-31"})
    body = {"req_key": "jimeng_t2i_v40", "prompt": prompt, "force_single": force_single}
    if width and height:
        body["width"] = width; body["height"] = height
    if size is not None:
        body["size"] = size
    req = urllib.request.Request(
        f"{base.rstrip('/')}/?{qs}",
        data=json.dumps(body).encode(),
        headers={"Content-Type": "application/json", "X-Access-Key": access_key},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        raise SystemExit(f"HTTP {e.code}: {e.read().decode(errors='replace')}") from e


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("prompt")
    ap.add_argument("--base", default=os.environ.get("JIMENG_BASE_URL", "https://visual.volcengineapi.com"))
    ap.add_argument("--ak", default=os.environ.get("VOLC_ACCESS_KEY", ""))
    ap.add_argument("--force-single", action="store_true", default=True)
    ap.add_argument("--width", type=int, default=None)
    ap.add_argument("--height", type=int, default=None)
    args = ap.parse_args(argv)
    if not args.ak:
        print("设置 VOLC_ACCESS_KEY（生产还需 SK + SDK 签名）", file=sys.stderr); return 2
    if not args.prompt.strip():
        print("prompt 不能为空", file=sys.stderr); return 2
    r = submit(args.base, args.ak, args.prompt, force_single=args.force_single,
               width=args.width, height=args.height)
    print(json.dumps(r, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

第 2 步：Mock

#!/usr/bin/env python3
"""Mock Jimeng CVSync2AsyncSubmitTask / GetResult (simplified AK header)."""
from __future__ import annotations
import json, uuid
from urllib.parse import parse_qs, urlparse
from http.server import BaseHTTPRequestHandler, HTTPServer

AK = "volc-test-ak"
TASKS = {}

class H(BaseHTTPRequestHandler):
    def log_message(self, *a): pass
    def _send(self, code, obj):
        b = json.dumps(obj).encode()
        self.send_response(code); self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(b))); self.end_headers(); self.wfile.write(b)
    def do_POST(self):
        if self.headers.get("X-Access-Key") != AK:
            self._send(401, {"code": 401, "message": "Unauthorized"}); return
        qs = parse_qs(urlparse(self.path).query)
        action = (qs.get("Action") or [""])[0]
        n = int(self.headers.get("Content-Length", 0))
        body = json.loads(self.rfile.read(n) or b"{}")
        if action == "CVSync2AsyncSubmitTask":
            if body.get("req_key") != "jimeng_t2i_v40" or not body.get("prompt"):
                self._send(400, {"code": 400, "message": "bad body"}); return
            tid = str(uuid.uuid4())
            TASKS[tid] = {"prompt": body["prompt"], "status": "in_queue"}
            self._send(200, {"code": 10000, "data": {"task_id": tid}}); return
        if action == "CVSync2AsyncGetResult":
            tid = body.get("task_id")
            t = TASKS.get(tid)
            if not t: self._send(404, {"code": 404, "message": "unknown task"}); return
            self._send(200, {"code": 10000, "data": {
                "status": "done",
                "image_urls": [f"https://cdn.example/jimeng/{tid}.png"],
            }}); return
        self.send_error(404)

def main():
    import argparse
    ap = argparse.ArgumentParser(); ap.add_argument("--port", type=int, default=9114)
    args = ap.parse_args()
    HTTPServer(("127.0.0.1", args.port), H).serve_forever()

if __name__ == "__main__":
    main()

第 3 步：测试

import json, subprocess, sys, time
from pathlib import Path
HERE = Path(__file__).resolve().parent

def test_submit():
    port = 19114
    proc = subprocess.Popen([sys.executable, str(HERE/"mock_jimeng.py"), "--port", str(port)],
                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        time.sleep(0.3)
        r = subprocess.run(
            [sys.executable, str(HERE/"jimeng_submit.py"), "国潮橘猫故宫红墙",
             "--base", f"http://127.0.0.1:{port}", "--ak", "volc-test-ak"],
            capture_output=True, text=True, check=True)
        data = json.loads(r.stdout)
        assert data["data"]["task_id"]
    finally:
        proc.terminate(); proc.wait(timeout=5)

第 4 步：运行

$ pytest 113a
.                                                                        [100%]
1 passed in 0.36s

怎么检查结果？

• data.task_id 非空。
• 生产务必 SDK 签名，勿只抄 X-Access-Key。

调试示例

签名失败。 核对 AK/SK、Region=cn-north-1、Service=cv；优先官方 Python/Go SDK，不要手写漏字段。

常见问题

模板规则

• SK 不进仓库。
• 先 force_single 试跑。
• 提示词遵守平台规范。

项目资源

任务交出去了，轮询与标识还在后头。