ComfyUI API 사용법 구축 (FAST api)

1. 코드 설명

FastAPI를 사용하여 ComfyUI의 이미지 생성 기능을 API로 제공.

주요 구성 요소:

• FastAPI 앱 설정
• Workflow 모델 정의 (Pydantic 사용)
• queue_prompt 함수: ComfyUI에 워크플로우 전송
• check_progress 함수: 이미지 생성 진행 상황 확인
• /generate 엔드포인트: 이미지 생성 요청 처리

2. 사용 방법

1. ComfyUI 서버 실행:
   ComfyUI 서버가 127.0.0.1:8188에서 실행 중이어야 함.
2. API 서버 실행:
3. python your_script_name.py
4. API 호출:
   • URL: http://localhost:8000/generate
   • 메소드: POST
   • 요청 본문: ComfyUI 워크플로우 JSON
   예시 (Python requests 사용):
5. import requests workflow = { # ComfyUI 워크플로우 JSON } response = requests.post("http://localhost:8000/generate", json={"workflow": workflow}) print(response.json())
6. 응답:
7. { "status": "completed", "image": "http://127.0.0.1:8188/view?filename=generated_image.png&type=temp" }

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import json
from urllib import request
import asyncio

app = FastAPI()

COMFYUI_IP = "127.0.0.1:8188"  # ComfyUI 서버 IP 및 포트

class Workflow(BaseModel):
    workflow: dict

def queue_prompt(prompt_workflow, ip):
    p = {"prompt": prompt_workflow}
    data = json.dumps(p).encode('utf-8')
    req = request.Request(f"http://{ip}/prompt", data=data)
    try:
        res = request.urlopen(req)
        if res.code != 200:
            raise Exception(f"Error: {res.code} {res.reason}")
        return json.loads(res.read().decode('utf-8'))['prompt_id']
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

async def check_progress(prompt_id: str, ip: str):
    while True:
        try:
            req = request.Request(f"http://{ip}/history/{prompt_id}")
            res = request.urlopen(req)
            if res.code == 200:
                history = json.loads(res.read().decode('utf-8'))
                if prompt_id in history:
                    return history[prompt_id]
        except Exception as e:
            print(f"Error checking progress: {str(e)}")
        await asyncio.sleep(1)  # 1초 대기

@app.post("/generate")
async def generate(workflow: Workflow):
    try:
        prompt_id = queue_prompt(workflow.workflow, COMFYUI_IP)
        result = await check_progress(prompt_id, COMFYUI_IP)

        # 결과에서 마지막 이미지 URL 추출
        final_image_url = None
        for node_id, node_output in result['outputs'].items():
            if 'images' in node_output:
                for image in node_output['images']:
                    final_image_url = f"http://{COMFYUI_IP}/view?filename={image['filename']}&type=temp"

        if final_image_url:
            return {"status": "completed", "image": final_image_url}
        else:
            return {"status": "completed", "image": None}
    except HTTPException as e:
        raise e
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)