# Midjourney APIをPythonで使う：画像生成と保存

> PythonでMidjourneyの画像タスクを作成し、状態をポーリングして画像グリッドを保存する手順。使える操作と、タイムアウト後の再実行方法を解説します。

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

---

OmniaKeyが提供するサードパーティの[Midjourney](https://omniakey.com/ja/midjourney) APIをPythonから呼び出す手順は3つです。タスクを作成し、完了するまで状態をポーリングし、署名付きアセットURLからPNGをダウンロードします。2026年11月12日の接続テストでは、1つのタスクが197.1秒で2048x2048の画像を返しました。この画像は4つの候補を並べた2x2グリッドです。以下のスクリプトは状態ファイルを追加するので、タイムアウトやダウンロード失敗の後に再実行しても、有料ジョブが二重に作られません。

## 短い答え

- `POST https://api.omniakey.com/v1/media/tasks` に `Idempotency-Key` を付けて送信し、`GET /v1/media/tasks/{id}` をポーリングして、`assets[0].url` をダウンロードします。
- 1つのタスクが返すのは、4枚の画像をまとめたPNGグリッド1枚です。4つの別ファイルではありません。
- このAPIのMidjourneyはテキストからの画像生成のみです。編集、参照画像、blend、アップスケール、vary、パン、バージョン選択はありません。
- キーはOmniaKeyから取得し、Midjourneyのサブスクリプションとは別です。POSTの前にリクエスト本文と冪等キーを保存し、APIが受け付けたらタスクIDを追加します。状態ファイルを残せば後から再開できます。ローカルのタイムアウトでサーバー側のタスクがキャンセルされることはありません。

## このPython例はどのAPIを使いますか？

この例では、OmniaKeyが提供するMidjourney向けのサードパーティAPIを呼び出します。OmniaKeyはMidjourneyではありません。OmniaKeyのキーは独自の残高を持つ別アカウントです。Midjourney公式サイトのサブスクリプションは別製品で、OmniaKeyのキーは作成されません。

[Midjourneyモデルページ](https://omniakey.com/ja/midjourney)には、OmniaKey側のモデルと現在の料金が表示されています。以下では、この接続とタスクの流れを説明します。

## 必要なもの

用意するものは3つです。

1. OmniaKeyのAPIキー。環境変数 `OMNIAKEY_API_KEY` に設定します。
2. Python 3。スクリプトは標準ライブラリだけを使うので、SDKのインストールは不要です。
3. そのキー/アカウントで `midjourney` が使えること。キーのモデル一覧にない場合は、作成リクエストを送る前にアカウントのアクセス権を確認してください。

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

キーはスクリプトにもログにも書かないでください。コードは環境変数から一度だけ読み取ります。

## リクエスト本文とパラメータ

Midjourneyにはプロンプトと3つのパラメータを渡します。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` で指定します。フィールドはキャメルケースで、`aspect_ratio` ではなく `aspectRatio` です。上の5つの値から選んでください。

サンプルプロンプトの本文は次のとおりです。

```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 APIを呼び出す

`midjourney_grid.py` として保存して実行します。初回はPOSTの前に `midjourney-task.json` を書き出し、OmniaKeyがリクエストを受け付けたらタスク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()
```

リクエストとレスポンスの形は11月12日の接続テストに基づいています。再開用の状態管理はここで追加した部分で、ライフサイクルの分岐はモック応答を使ってオフラインで確認しました。

状態ファイルには、再開に必要なものが残ります。JSON本文と、冪等キーとして使うUUIDです。APIキーは保存しません。POSTの前に書き出すので、接続が切れても復旧に必要な2つの値がディスクに残ります。

作成リクエストは2つの成功コードを返すことがあります。`202` は新しいジョブを受け付けた状態です。`200` はOmniaKeyが同じキーと同じ本文を以前に見ていて、元のタスクを返している状態です。そのタスクのポーリングを続けてください。同じ本文とキーを再送するのは、接続障害の後に作成結果が不明な場合だけです。本文を変えると `409 idempotency_key_reused` になります。

ポーリングは `status` で制御します。パーセントは目安で、ジョブが動いていても0のままのことがあります。`succeeded`、`failed`、`canceled` で停止します。OmniaKeyの[メディアタスクのライフサイクル](https://docs.omniakey.com/en/media/overview)（2026年11月12日確認）にこれらの状態が記載されています。

ポーリングの上限はローカル側の設定です。上限に達してもジョブは実行中か待機中で、同じスクリプトを再実行すると保存済みのタスクIDを読み取ってポーリングを続けます。ローカルのタイムアウトはキャンセルではありません。キャンセルには別のエンドポイントを使います。

## 返ってくるMidjourneyの画像グリッド

成功したタスクのアセットは1つだけです。PNGグリッドです。以下は2026年11月12日の接続テストで得たサンプルです。

![青い水面に浮かぶ赤い折り紙の舟4隻。2x2のグリッドに並び、朝日が暖かく反射している](https://omniakey.com/_img/blog/midjourney-api-python/paper-boat-grid.03735622.1600.webp)

_2026年11月12日の接続テストのサンプル：midjourneyタスク1件、2048x2048のPNG 1枚、候補4つ、作成からダウンロードまで197.1秒。ダウンロードしたバイト列はタスクのSHA-256と一致しました。_

以下のレスポンスは短縮版です。URLは伏せてあり、`expires_at` は当時の値です。必要なアセットフィールドを追加しています。

```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を固定値にせず、`width` と `height` はアセットから読み取ってください。2048x2048と197.1秒は1つのサンプルです。4つの候補は1枚のPNGを共有するので、単体画像が必要な場合はダウンロード後にグリッドを分割します。

`url` は署名付きで短時間だけ有効です。有効期限は約1時間です。OmniaKeyの[画像APIドキュメント](https://docs.omniakey.com/en/media/image)（2026年11月12日確認）によると、タスク記録と画像は作成後24時間保持されます。URLの期限が切れる前にダウンロードしてください。期限切れの場合は、タスクIDを再度ポーリングして新しい署名付きURLを取得します。

アセットURLは通常のGETでダウンロードし、`Authorization` ヘッダーは付けません。署名はURLに含まれています。スクリプトはダウンロードしたバイト列をタスクの `sha256` と比較するので、途中で切れたファイルを良いコピーに上書きする前に検出できます。

## タイムアウトやダウンロード失敗後の再実行

同じコマンドを再実行し、ジョブを引き継ぎます。スクリプトは `midjourney-task.json` を読み、最も短い安全な経路を選びます。

- 作成が未確認：同じ本文を同じ冪等キーでPOSTします。
- タスクIDがあり、ジョブがまだ実行中：既存のIDをポーリングします。
- タスクは成功したがPNGがない：新しい署名付きURLを取得するために一度ポーリングし、もう一度ダウンロードします。
- PNGが保存済みで記録したSHA-256と一致する：追加のリクエストなしで終了します。壊れたファイルはポーリングして再ダウンロードします。

これは24時間の保持期間内で機能します。同じキーでプロンプトやパラメータを変えると `409` になります。スクリプトはプロンプトが一致しない状態ファイルを拒否します。パラメータを変えるときは、状態ファイルを別の場所へ移してください。

## エラーと対処

以下はAPIのタスクルールと冪等ルールによる失敗です。原因を直し、該当する場合は同じ状態ファイルで再実行してください。

|レスポンス|コード|対処|
|-|-|-|
|400|`missing_idempotency_key`|ヘッダーを追加し、255文字以内に収める|
|409|`idempotency_key_reused`|元の本文を送るか、新しいキーで新しいジョブを開始する|
|503|`no_available_channel`|このキー/アカウントのアクセスを確認し、続く場合はサポートに連絡する|
|`failed`|`error.code`、`error.message`|タスクのエラーを確認する。失敗したジョブは返金される|
|`canceled`|—|ジョブは停止済み。必要なら新しく作成する|

`503 no_available_channel` は、キーにモデル権限がないことを証明するものではありません。このバッチで使った1つのキーでは、2回の送信が `503` を返し、タスクは作成されませんでした。別のキーは、同じ日の早い時間の接続テストで成功しています。原因は不明で、キー/アカウントの可用性か上流チャネルの可用性が考えられます。現在のキー/アカウントで `midjourney` が使えるか確認してください。`503` が続く場合は、レスポンス本文とリクエストIDをサポートに送ってください。

作成時の `200` は元のタスクの再送です。終了状態になる失敗は `failed` と `canceled` だけです。スクリプトを止めてもジョブは動き続けます。

## 料金と、別の画像APIを選ぶとき

OmniaKeyの課金は成功したジョブ1件につき1回で、失敗またはキャンセルされたジョブは自動的に返金されます。現在の料金は[Midjourneyモデルページ](https://omniakey.com/ja/midjourney)にあります。バッチを見積もる前に確認してください。

Midjourneyは、1つのグリッドで4つのバリエーションを見比べたいときに向いています。この接続が返すのはテキストからの画像グリッドです。1回の呼び出しで1枚の画像が欲しい場合や、元の場面を保つ編集が必要な場合は、[GPT Image 2.5 API](https://omniakey.com/ja/blog/gpt-image-2-5-api)または[Nano Banana Pro API](https://omniakey.com/ja/blog/nano-banana-pro-api)を参照してください。

コードを書かずにプロンプトを入力したい場合は、[AI Image Generator](https://omniakey.com/ja/ai-image-generator)でブラウザから画像モデルを実行できます。

## よくある質問

### Midjourneyのサブスクリプションは必要ですか？

いいえ。キーはOmniaKeyから取得し、呼び出しはOmniaKeyのサードパーティAPIに送られます。OmniaKeyはMidjourneyではありません。Midjourney公式サイトのサブスクリプションは別製品で、OmniaKeyのキーは作成されません。

### 1つのタスクで4枚の画像が1ファイルで返るのはなぜですか？

Midjourneyは4つの候補を2x2グリッドの1枚のPNGとして返します。この接続ではアップスケールとvaryが使えないため、4つの候補が同じファイルを共有します。別々のファイルが必要な場合は、ダウンロード後に自分でグリッドを分割してください。

### 409 idempotency\_key\_reused とは何ですか？

OmniaKeyがすでに持っている冪等キーを送信しましたが、リクエスト本文が変わっています。元の本文を送るか、新しいキーで新しいジョブを開始してください。同じキーでプロンプトを変えるのがよくある原因です。

### ポーリング中にスクリプトがタイムアウトしたらどうなりますか？

ジョブは動き続けます。タスクIDは `midjourney-task.json` にあるので、同じスクリプトを再実行するとそのIDをポーリングするか、完了したアセットをダウンロードします。ローカルのタイムアウトではジョブも課金もそのまま残ります。
