ComfyUI APIでACE-Step音楽生成を自動化する全体像まとめ

ここまでの記事では、ComfyUIとAPIを利用してACE-StepをPythonから制御する方法を段階的に紹介してきました。

本記事では、その内容を一度整理し、全体像としてPythonコードをまとめます。

ComfyUIでACE-StepをAPI用JSONをエクスポート
音楽生成ジョブをAPIで投入する
WebSocketで進捗を確認する
生成された音楽ファイルを取得する
サンプルコードまとめ
まとめ

ComfyUIでACE-StepをAPI用JSONをエクスポート

ComfyUIは、ワークフローをHTTP API経由で実行できるノードベースの生成AIツールです。

ACE-Stepはテキストから音楽を生成するオープンソースモデルで、ComfyUI公式ドキュメントでも音楽生成チュートリアルが提供されています。

API利用時は、UIで作成したワークフローを「API形式JSON」としてエクスポートし、それをそのままサーバーに送信します。

音楽生成ジョブをAPIで投入する

API操作の起点は POST /prompt です。Pythonなど外部プログラムから、ACE-StepワークフローJSONを送信すると、ComfyUIは生成ジョブをキューに登録し、prompt_id を返します。

このIDが、以降の進捗確認や結果取得の鍵になります。

WebSocketで進捗を確認する

ACE-Stepでの音楽生成は数十秒から数分かかるため、進捗確認が重要です。

ComfyUIは WebSocket（/ws） を提供しており、生成中の進捗や実行状態をリアルタイムで受信できます。

PythonではWebSocketクライアントとTQDMを組み合わせることで、CLI上に進捗バーを表示できます。

生成された音楽ファイルを取得する

生成が完了したら GET /history/{prompt_id} を呼び出し、どのファイルが出力されたかを確認します。

ACE-Stepの音楽出力は audio として履歴に記録され、ファイル名や保存先サブフォルダを取得できます。その情報をもとに GET /view を使って実ファイルをダウンロードします。

サンプルコードまとめ

これまでの紹介記事をまとめ、実行可能な一つのコードとしてまとめたものを示します。

コードはChatGPTに生成してもらい、ComfyUI v0.9.1 + ACE-Step v1.2 + Python 3.10系で動作を確認していますが、将来のバージョンでは動作しない可能性があります。動作しない場合はAIに見せて、最新バージョンや自分の環境に合わせてもらうのも有効な手段です。

事前に必要パッケージをいれておきます。

pip install requests websocket-client tqdm

実行時には以下の点に注意してください。

同じフォルダに、ACE-StepワークフローをAPI向けにエクスポートしたJSONファイルをaudio-ace-step.jsonとして配置しておきます。
ComfyUIサーバーはローカルで起動しておきます。ポート番号はデフォルトの8188を指定しています。
URLを適宜書き換えればリモートでも動きます。リモートの場合は、ComfyUI起動時に –listen オプションを指定し外部アクセスを許可しておきます。
実行後は、sample_generated_audio.wav に生成音声が保存されます。

import json
import time
import uuid
import requests
import websocket
from tqdm import tqdm
import os

COMFY_BASE = "http://127.0.0.1:8188"
COMFY_PROMPT = f"{COMFY_BASE}/prompt"
COMFY_HISTORY = f"{COMFY_BASE}/history"
COMFY_VIEW = f"{COMFY_BASE}/view"
WS_URL = "ws://127.0.0.1:8188/ws"

WORKFLOW_JSON = "audio_ace_step.json"
OUTPUT_FILE = "sample_generated_audio.wav"


def load_and_patch(json_path: str, prompt_text: str, duration_s: int = 60):
    wf = json.load(open(json_path, "r", encoding="utf-8"))
    for node in wf.values():
        if not isinstance(node, dict):
            continue
        cls = node.get("class_type", "")
        inputs = node.get("inputs", {}) or {}
        if cls == "TextEncodeAceStepAudio":
            if "tags" in inputs:
                inputs["tags"] = prompt_text
            inputs["lyrics"] = "[inst]"
        if cls == "EmptyAceStepLatentAudio":
            if "seconds" in inputs:
                inputs["seconds"] = duration_s
        if cls == "KSampler" and "seed" in inputs:
            inputs["seed"] = int(uuid.uuid4().int & (2**32 - 1))
    return wf


def submit_workflow(wf: dict, client_id: str) -> str:
    r = requests.post(COMFY_PROMPT, json={"prompt": wf, "client_id": client_id}, timeout=30)
    r.raise_for_status()
    j = r.json()
    pid = j.get("prompt_id")
    if not pid:
        raise RuntimeError("no prompt_id returned")
    return pid


def wait_and_show_progress(prompt_id: str, client_id: str, timeout: int = 1800):
    ws_url = f"{WS_URL}?clientId={client_id}"
    ws = websocket.create_connection(ws_url)
    start = time.time()
    bar = tqdm(total=100, desc="Comfy", unit="%")
    last = -1
    try:
        while True:
            if time.time() - start > timeout:
                raise TimeoutError("WS timeout")
            raw = ws.recv()
            try:
                data = json.loads(raw)
            except Exception:
                continue
            mtype = data.get("type")
            pdata = data.get("data") or {}
            pid = pdata.get("prompt_id")

            if pid != prompt_id:
                continue

            if mtype == "progress":
                v = pdata.get("value")
                maxv = pdata.get("max", 1)
                try:
                    pct = int(v / maxv * 100)
                except Exception:
                    pct = None
                if pct is not None and pct != last:
                    delta = max(0, min(100, pct)) - bar.n
                    if delta > 0:
                        bar.update(delta)
                    last = pct

            if mtype == "execution_error":
                raise RuntimeError("execution_error from ComfyUI")

            if (mtype == "execution_success") or (mtype == "executing" and pdata.get("node") is None):
                if bar.n < 100:
                    bar.update(100 - bar.n)
                bar.close()
                return
    finally:
        try:
            ws.close()
        except Exception:
            pass


def download_first_audio(prompt_id: str, out_path: str):
    r = requests.get(f"{COMFY_HISTORY}/{prompt_id}", timeout=30)
    r.raise_for_status()
    hist = r.json()
    entry = hist.get(prompt_id, {})
    outputs = entry.get("outputs") or {}
    type = "audio"

    for out in outputs.values():
        if not isinstance(out, dict):
            continue
        # ACEStep の出力は 'audio' に入る
        if type in out and isinstance(out[type], list) and out[type]:
            item = out[type][0]
            fn = item.get("filename")
            sub = item.get("subfolder", "")
            if not fn:
                continue

            # type は履歴のエントリに従う（例: 'output'）。デフォルトは 'output'
            t = item.get("type", "output")
            params = {"filename": fn, "type": t}
            
            if sub:
                params["subfolder"] = sub

            vr = requests.get(COMFY_VIEW, params=params, timeout=60)
            vr.raise_for_status()
            with open(out_path, "wb") as f:
                f.write(vr.content)
            return out_path

    raise RuntimeError("no audio found in history outputs")


PROMPT_TEXT = "instrumental, medieval fantasy, calm background"  # サンプル
CLIENT_ID = str(uuid.uuid4())

def main():
    wf = load_and_patch(WORKFLOW_JSON, PROMPT_TEXT, duration_s=60)
    pid = submit_workflow(wf, CLIENT_ID)
    print("submitted prompt_id:", pid)
    wait_and_show_progress(pid, CLIENT_ID)
    saved = download_first_audio(pid, OUTPUT_FILE)
    print("saved to:", saved)


if __name__ == "__main__":
    main()