ComfyUI API 구축 (FAST api) - AIX-GANGNAM/mini-project GitHub Wiki

1. 코드 설명

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

주요 구성 요소:

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

2. 사용 방법

1. ComfyUI 서버 실행: ComfyUI 서버가 127.0.0.1:8188에서 실행 중이어야 함.

2. API 서버 실행:

   python your_script_name.py
   

3. API 호출:

   • URL: http://localhost:8000/generate
   • 메소드: POST
   • 요청 본문: ComfyUI 워크플로우 JSON

   예시 (Python requests 사용):

   import requests
   
   workflow = {
       # ComfyUI 워크플로우 JSON
   }
   
   response = requests.post("http://localhost:8000/generate", json={"workflow": workflow})
   print(response.json())
   

4. 응답:

   {
     "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)