# Midjourney API 怎么接？Python 出图、轮询与下载

> 用 Python 提交 Midjourney 出图任务、轮询状态并下载四宫格结果，讲清第三方 API 接入、等待超时和当前支持的操作。

- Canonical HTML: https://omniakey.com/zh/blog/midjourney-api-python
- Markdown: https://omniakey.com/zh/blog/midjourney-api-python.md
- Published: 2026-11-12 (updated 2026-11-12)
- Author: Kiki, COO at OmniaKey
- Language: zh
- Category: API setup
- Reading time: 8 min
- All posts: https://omniakey.com/blog/llms.txt

---

在 OmniaKey 上，用 Python 出一张 [Midjourney](https://omniakey.com/zh/midjourney) 图只要三步：创建任务、轮询状态、下载 PNG。2026 年 11 月 12 日的一次接入测试里，一个任务用 197.1 秒返回了一张 2048x2048 的四宫格，四个候选图都在同一张图里。下面的脚本会先把幂等键和请求体写到磁盘，超时或下载失败后重跑，不会多建一个任务。

## 先说结论

- 调用方式是 `POST https://api.omniakey.com/v1/media/tasks`，带上 `Idempotency-Key`；然后轮询 `GET /v1/media/tasks/{id}`，成功后再下载 `assets[0].url`。
- 一个任务返回一张 PNG 四宫格，四个候选在同一张图里，不是四个文件。
- 这条接入只做文生图：没有改图、参考图、blend、放大、vary、平移，也没有版本选择。
- key 来自 OmniaKey，和 Midjourney 官网订阅是两件事。发 POST 前先存请求体和幂等键，任务被接受后再补 task id，状态文件用来保存续跑需要的信息。本地超时不会取消服务端任务。

## 这个 Python 示例用的是哪个 API？

示例调用的是 OmniaKey 面向 Midjourney 的第三方 API。OmniaKey 不是 Midjourney，OmniaKey 的 key 是另一个账户，有自己的余额；Midjourney 官网订阅是另一个产品，不会生成 OmniaKey 的 key。

[Midjourney 模型页](https://omniakey.com/zh/midjourney) 能看到模型和 OmniaKey 侧的当前单价。下面讲这条接入的任务流程。

## 要准备什么

三样东西：

1. OmniaKey 的 API key，放进环境变量 `OMNIAKEY_API_KEY`。
2. Python 3。脚本只用标准库，不用装 SDK。
3. 这个 key/账户能用到 `midjourney`。模型列表里没有它，先确认账户权限，再发创建请求。

```bash
export OMNIAKEY_API_KEY="your-key"
python3 midjourney_grid.py
```

key 不要写进脚本，也不要打进日志。脚本只从环境变量读一次。

## 请求体和参数

Midjourney 要一个提示词加三个参数。bot type 在 OmniaKey 侧已经预设，请求体里不用写模型版本，也不用写 `botType`。

|字段|值|说明|
|-|-|-|
|`input.prompt`|文本|你想要的画面|
|`parameters.aspectRatio`|`1:1`|还可填 `16:9`、`9:16`、`3:2`、`2:3`|
|`parameters.quality`|`"1"`|文档里写的是字符串 `"1"`|
|`parameters.n`|`1`|只能是 1|

图片形状用 `aspectRatio` 控制，没有 `size`、`width`、`height` 或像素值可填。字段是驼峰写法，写 `aspectRatio`，不是 `aspect_ratio`。按上面这五个值填。

样本提示词的请求体是这样：

```json
{
  "model": "midjourney",
  "type": "image",
  "operation": "generate",
  "input": {
    "prompt": "A small red paper boat floating on a quiet blue lake at sunrise, cinematic lighting, elegant composition, no text"
  },
  "parameters": {"aspectRatio": "1:1", "quality": "1", "n": 1}
}
```

## Python 脚本

保存为 `midjourney_grid.py`，直接运行。第一次运行会先写 `midjourney-task.json`，再发 POST；任务被接受后，脚本再把 task id 补进去。

```python
import hashlib
import json
import os
import sys
import time
import urllib.error
import urllib.request
import uuid
from pathlib import Path

API = "https://api.omniakey.com/v1"
KEY = os.environ["OMNIAKEY_API_KEY"]
STATE = Path("midjourney-task.json")
OUT = Path("midjourney-grid.png")
PROMPT = ("A small red paper boat floating on a quiet blue lake at sunrise, "
          "cinematic lighting, elegant composition, no text")


def save(state):
    tmp = STATE.with_suffix(".tmp")
    tmp.write_text(json.dumps(state, indent=2))
    tmp.replace(STATE)


def read_json(response):
    raw = response.read()
    try:
        return json.loads(raw)
    except ValueError:
        return {"raw": raw[:300].decode("utf-8", "replace")}


def call(method, path, body=None, headers=None):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(
        API + path, data=data, method=method,
        headers={"Authorization": f"Bearer {KEY}", **(headers or {})})
    if body is not None:
        req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=60) as response:
            return response.status, read_json(response)
    except urllib.error.HTTPError as error:
        return error.code, read_json(error)
    except (urllib.error.URLError, TimeoutError) as error:
        return None, {"error": str(error)}


def main():
    state = json.loads(STATE.read_text()) if STATE.exists() else None
    if state is None:
        body = {
            "model": "midjourney", "type": "image", "operation": "generate",
            "input": {"prompt": PROMPT},
            "parameters": {"aspectRatio": "1:1", "quality": "1", "n": 1},
        }
        state = {"idempotency_key": str(uuid.uuid4()), "body": body,
                 "task_id": None, "status": "creating"}
        save(state)  # no API key here; write before POST
    elif state["body"]["input"]["prompt"] != PROMPT:
        sys.exit("The state file holds a different prompt. Move it aside to start a new job.")

    if OUT.exists() and state.get("status") == "downloaded":
        saved_sha = state.get("download_sha")
        if not saved_sha or hashlib.sha256(OUT.read_bytes()).hexdigest() == saved_sha:
            print("already saved", OUT)
            return

    if not state["task_id"]:
        code, task = call("POST", "/media/tasks", state["body"],
                          {"Idempotency-Key": state["idempotency_key"]})
        if code is None:
            sys.exit(f"Create not confirmed: {task['error']}. Rerun to retry the same key and body.")
        if code == 409:
            sys.exit("409 idempotency_key_reused: same key, different body. Restore the saved body.")
        if code == 503:
            sys.exit("503 no_available_channel: check model access for this key/account; "
                     "contact support with the request id if it persists.")
        if code not in (200, 202) or "id" not in task:
            sys.exit(f"Create failed: HTTP {code} {task}")
        state.update(task_id=task["id"], status=task["status"])
        save(state)

    deadline = time.monotonic() + 900
    while True:
        code, task = call("GET", f"/media/tasks/{state['task_id']}")
        if code is None:
            sys.exit(f"Polling stopped: {task['error']}. The job is still saved; rerun this script.")
        if code != 200 or "status" not in task:
            sys.exit(f"Poll failed: HTTP {code} {task}")
        state["status"] = task["status"]
        save(state)
        if task["status"] in ("failed", "canceled"):
            sys.exit(f"Job {task['status']}: {task.get('error')}")
        if task["status"] == "succeeded":
            break
        if time.monotonic() > deadline:
            sys.exit(f"Still {task['status']} after 900 s. Rerun with the same state file.")
        time.sleep(min(8, 2 + task.get("progress", {}).get("percent", 0) // 20))

    asset = task["assets"][0]
    if asset.get("mime_type") != "image/png":
        sys.exit(f"Expected a PNG grid, got {asset.get('mime_type')}")
    try:
        with urllib.request.urlopen(asset["url"], timeout=120) as response:
            data = response.read()  # signed URL: no Authorization header
    except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError) as error:
        sys.exit(f"Download failed: {error}. Rerun to get a fresh signed URL and retry.")
    digest = hashlib.sha256(data).hexdigest()
    if asset.get("sha256") and digest != asset["sha256"]:
        sys.exit("SHA-256 mismatch. Rerun to download the asset again.")
    part = OUT.with_suffix(".part")
    part.write_bytes(data)
    part.replace(OUT)
    state.update(status="downloaded", download_sha=digest)
    save(state)
    print("saved", OUT, len(data), "bytes")


if __name__ == "__main__":
    main()
```

脚本沿用 2026 年 11 月 12 日接入测试的请求和响应格式，额外加了断点续跑；生命周期分支用模拟响应离线检查过。

状态文件保存续跑需要的信息：请求体和 UUID 幂等键，不保存 API key，而且在 POST 之前就落盘。网络断在响应回来之前，key 和 body 还在磁盘上，下一次运行可以接着处理。

创建成功有两个码。`202` 表示接了新任务；`200` 表示 OmniaKey 见过同一个 key 和同一个 body，正在返回原来的任务，接着轮询这个任务即可。只有连接失败、创建结果未知时，才用同一个 key 和 body 重发。这个 key 换了 body，就会得到 `409 idempotency_key_reused`。

轮询看 `status`，不要看 `progress.percent`。百分比是估算值，可能一直是 0。到 `succeeded`、`failed` 或 `canceled` 就停。OmniaKey 的 [媒体任务文档](https://docs.omniakey.com/cn/media/overview)（2026 年 11 月 12 日核对）写了这几个状态。

脚本里的轮询上限只是本地设置。到点任务还在跑，重跑脚本会读回 task id 接着轮询。本地超时不等于取消，取消要另外调 cancel 接口。

## 返回的是什么

成功的任务只有一个 asset：一张 PNG 四宫格。下面是 2026 年 11 月 12 日接入测试拿到的样本：

![蓝色水面上漂着四只红色折纸小船，呈 2x2 四宫格排列，晨光在水面反光](https://omniakey.com/_img/blog/midjourney-api-python/paper-boat-grid.03735622.1600.webp)

_2026 年 11 月 12 日接入测试样本：一个 midjourney 任务，一张 2048x2048 PNG，四个候选，197.1 秒完成。下载文件的 SHA-256 与任务记录一致。_

下面的响应有删减：URL 已打码，`expires_at` 是 11 月 12 日那次的历史值。要读的字段是这些：

```json
{
  "id": "task_DbCWJPfPL1jyhSD38IihNHiiFQao8hjV",
  "status": "succeeded",
  "progress": {"percent": 100, "stage": "completed"},
  "assets": [
    {
      "kind": "image",
      "mime_type": "image/png",
      "width": 2048,
      "height": 2048,
      "byte_size": 5834301,
      "sha256": "f503cef038d15fa150b36b5fe7c41874febad3aa1167698d790781e47a4a10b9",
      "expires_at": "2026-11-12T04:16:41Z",
      "url": "https://…redacted-signed-url…"
    }
  ]
}
```

不要写死 2048。宽高从 asset 里读；2048x2048 和 197.1 秒都是这一次样本。四个候选共用一张 PNG，如果需要单张，下载后自己切。

`url` 是带签名的短链接，大约一小时过期。任务记录和图片按 OmniaKey 的 [图片 API 文档](https://docs.omniakey.com/cn/media/image)（2026 年 11 月 12 日核对）保留 24 小时。链接过期就再轮询一次 task id，换一条新的签名链接。

下载 asset 的链接用普通 GET，不加 `Authorization` 头，签名已经在 URL 里。脚本还会比对下载文件的 `sha256`，文件断了就先发现，不会覆盖掉上一份好图。

## 超时或下载失败后怎么重跑

重跑同一条命令，不要新建任务。脚本会读 `midjourney-task.json`，按当前进度走最短的路：

- 创建没确认：用同一个 key、同一个 body 再 POST。
- 有 task id，任务还在跑：继续轮询原 id。
- 任务成功但文件不在：再轮询一次，拿新的签名链接重新下载。
- 文件已经在：先核对 SHA-256，没问题就直接退出；文件坏了就重新轮询、重新下载。

这些都在 24 小时保留期内有效。同一个 key 换了提示词或参数会得到 `409`。脚本发现提示词对不上，会拒绝复用这份状态文件；改参数时，把状态文件移开再跑。

## 报错怎么处理

下面这些是任务和幂等规则会碰到的情况。先改原因，能复用状态文件就复用。

|响应|错误码|怎么办|
|-|-|-|
|400|`missing_idempotency_key`|加上这个请求头，长度控制在 255 个字符内|
|409|`idempotency_key_reused`|发回原来的 body，或者换新 key 新建任务|
|503|`no_available_channel`|先确认这个 key/账户能用 `midjourney`；还不行就联系支持|
|`failed`|`error.code`、`error.message`|看任务错误；失败任务会自动退款|
|`canceled`|—|任务停了；还需要就新建一个|

`503 no_available_channel` 不能证明这个 key 缺少模型权限。这批测试中有两次提交用的是同一个 key，都返回了 `503`，没有生成任务；同一天更早的一次接入测试用的是另一个 key，成功出了图。原因还不确定：可能是 key/账户可用性，也可能是上游通道可用性。先确认当前的 key/账户能用 `midjourney`。`503` 还在，就把响应体和 request id 发给支持。

创建返回 `200` 不是错误，是原任务的重放。只有 `failed` 和 `canceled` 是终止失败；脚本停止轮询，任务还在跑。

## 费用和替代选择

OmniaKey 按成功的任务收一次费，失败和取消会自动退款。当前单价看 [Midjourney 模型页](https://omniakey.com/zh/midjourney)；批量前按成功任务数估。

要四个方向让你挑，Midjourney 合适；这条接入目前只返回四宫格文生图。要一次一张、还要改图，可以看 [GPT Image 2.5 API](https://omniakey.com/zh/blog/gpt-image-2-5-api) 或 [Nano Banana Pro API](https://omniakey.com/zh/blog/nano-banana-pro-api)。不想写代码，直接在 [AI Image Generator](https://omniakey.com/zh/ai-image-generator) 里输入提示词。

## 常见问题

### 用这个 API 需要 Midjourney 订阅吗？

不需要。key 来自 OmniaKey，调用走的是 OmniaKey 的第三方 API。OmniaKey 不是 Midjourney；Midjourney 官网订阅是另一个产品，不会生成 OmniaKey 的 key。

### 为什么一个任务返回四张图，却只有一个文件？

Midjourney 把四个候选拼成一张 2x2 的四宫格 PNG。这条接入没有单张切图和放大、vary 接口，所以四个候选共用一张文件，要单张就下载后自己切。

### 409 idempotency\_key\_reused 是什么意思？

这个幂等键 OmniaKey 已经见过，但请求体变了。发回原来的 body，或者换一个新 key 新建任务。最常见的原因是同一个 key 改了提示词。

### 轮询超时了，任务会丢吗？

任务还在跑。task id 在 `midjourney-task.json` 里，重跑同一份脚本就会接着轮询，或者下载已经完成的图。本地超时不会取消任务，也不会触发退款。